lee/changedetection.io
1
0
Fork 0
mirror of https://github.com/dgtlmoon/changedetection.io.git synced 2026-09-25 23:06:51 +00:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
dgtlmoon
2025-05-30 18:49:43 +02:00
parent 3d61ce8df7
commit 6c1ed57032
2 changed files with 79 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changedetectionio/async_update_worker.py
+29 -34
View File
@@ -30,7 +30,6 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore):
while not app.config.exit.is_set():
update_handler = None
watch = None
current_uuid = None
try:
# Use asyncio wait_for to make queue.get() cancellable
@@ -43,16 +42,14 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore):
await asyncio.sleep(0.1)
continue
uuid = None
uuid = queued_item_data.item.get('uuid')
fetch_start_time = round(time.time())
# Mark this UUID as being processed
from changedetectionio import worker_handler
worker_handler.set_uuid_processing(uuid, processing=True)
try:
uuid = queued_item_data.item.get('uuid')
fetch_start_time = round(time.time())
current_uuid = uuid
# Mark this UUID as being processed
from changedetectionio import worker_handler
worker_handler.set_uuid_processing(uuid, processing=True)
if uuid in list(datastore.data['watching'].keys()) and datastore.data['watching'][uuid].get('url'):
changed_detected = False
contents = b''
@@ -354,39 +351,37 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore):
datastore.update_watch(uuid=uuid, update_obj={'fetch_time': round(time.time() - fetch_start_time, 3),
'check_count': count})
current_uuid = None
# Mark UUID as no longer being processed
worker_handler.set_uuid_processing(uuid, processing=False)
# Send completion signal
if watch:
logger.info(f"Worker {worker_id} sending completion signal for UUID {watch['uuid']}")
watch_check_update.send(watch_uuid=watch['uuid'])
update_handler = None
logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s")
# Yield control to other coroutines
await asyncio.sleep(0.01)
except Exception as e:
logger.error(f"Worker {worker_id} unexpected error processing {uuid}: {e}")
logger.error(f"Worker {worker_id} traceback:", exc_info=True)
# Make sure to mark UUID as completed even on error
# Also update the watch with error information
if datastore and uuid in datastore.data['watching']:
datastore.update_watch(uuid=uuid, update_obj={'last_error': f"Worker error: {str(e)}"})
finally:
# Always cleanup - this runs whether there was an exception or not
if uuid:
try:
# Mark UUID as no longer being processed
worker_handler.set_uuid_processing(uuid, processing=False)
# Also update the watch with error information
if datastore and uuid in datastore.data['watching']:
datastore.update_watch(uuid=uuid, update_obj={'last_error': f"Worker error: {str(e)}"})
# Send completion signal
if watch:
#logger.info(f"Worker {worker_id} sending completion signal for UUID {watch['uuid']}")
watch_check_update.send(watch_uuid=watch['uuid'])
update_handler = None
logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s")
except Exception as cleanup_error:
logger.error(f"Worker {worker_id} error during cleanup: {cleanup_error}")
current_uuid = None
# Brief pause before continuing to avoid tight error loops
await asyncio.sleep(1.0)
# Brief pause before continuing to avoid tight error loops (only on error)
if 'e' in locals():
await asyncio.sleep(1.0)
else:
# Small yield for normal completion
await asyncio.sleep(0.01)
# Check if we should exit
if app.config.exit.is_set():
changedetectionio/realtime/socket_server.py
+50 -53
View File
@@ -8,8 +8,10 @@ from blinker import signal
from changedetectionio import strtobool
class SignalHandler:
"""A standalone class to receive signals"""
def __init__(self, socketio_instance, datastore):
self.socketio_instance = socketio_instance
self.datastore = datastore
@@ -17,18 +19,17 @@ class SignalHandler:
# Connect to the watch_check_update signal
from changedetectionio.flask_app import watch_check_update as wcc
wcc.connect(self.handle_signal, weak=False)
# logger.info("SignalHandler: Connected to signal from direct import")
# logger.info("SignalHandler: Connected to signal from direct import")
# Connect to the queue_length signal
queue_length_signal = signal('queue_length')
queue_length_signal.connect(self.handle_queue_length, weak=False)
# logger.info("SignalHandler: Connected to queue_length signal")
# logger.info("SignalHandler: Connected to queue_length signal")
# Create and start the queue update thread using eventlet
import eventlet
self.polling_emitter_thread = eventlet.spawn(self.polling_emit_running_or_queued_watches)
# Store the thread reference in socketio for clean shutdown
self.socketio_instance.polling_emitter_thread = self.polling_emitter_thread
@@ -43,7 +44,7 @@ class SignalHandler:
watch = self.datastore.data['watching'].get(watch_uuid)
if watch:
if app_context:
#note
# note
with app_context.app_context():
with app_context.test_request_context():
# Forward to handle_watch_update with the watch parameter
@@ -60,17 +61,16 @@ class SignalHandler:
try:
queue_length = kwargs.get('length', 0)
logger.debug(f"SignalHandler: Queue length update received: {queue_length}")
# Emit the queue size to all connected clients
self.socketio_instance.emit("queue_size", {
"q_length": queue_length,
"event_timestamp": time.time()
})
except Exception as e:
logger.error(f"Socket.IO error in handle_queue_length: {str(e)}")
def polling_emit_running_or_queued_watches(self):
"""Greenlet that periodically updates the browser/frontend with current state of who is being checked or queued
This is because sometimes the browser page could reload (like on clicking on a link) but the data is old
@@ -81,7 +81,7 @@ class SignalHandler:
from changedetectionio.flask_app import app
from changedetectionio import worker_handler
watch_check_update = signal('watch_check_update')
# Use eventlet sleep for non-blocking operation
from eventlet import sleep as eventlet_sleep
@@ -91,28 +91,22 @@ class SignalHandler:
# Run until explicitly stopped
while stop_event is None or not stop_event.ready():
try:
# For each running UUID, send a signal, so we update the UI
running_uuids = worker_handler.get_running_uuids()
for uuid in running_uuids:
logger.trace(f"Sending update for {uuid}")
# Send with app_context to ensure proper URL generation
with app.app_context():
watch_check_update.send(app_context=app, watch_uuid=uuid)
# Yield control back to eventlet after each send to prevent blocking
eventlet_sleep(0.1) # Small sleep to yield control
# Check if we need to stop in the middle of processing
if stop_event is not None and stop_event.ready():
break
# Get current running UUIDs from async workers
running_uuids = set(worker_handler.get_running_uuids())
# Sleep between polling/update cycles
eventlet_sleep(2)
# Send updates for newly running UUIDs
with app.app_context():
for uuid in running_uuids:
watch_check_update.send(app_context=app, watch_uuid=uuid)
eventlet_sleep(0.01) # Small yield
except Exception as e:
logger.error(f"Error in queue update greenlet: {str(e)}")
# Sleep a bit to avoid flooding logs in case of persistent error
eventlet_sleep(0.5)
eventlet_sleep(10)
logger.info("Queue update eventlet greenlet stopped")
@@ -141,26 +135,29 @@ def handle_watch_update(socketio, **kwargs):
error_texts = watch.compile_error_texts()
# Create a simplified watch data object to send to clients
watch_uuid = watch.get('uuid')
watch_data = {
'checking_now': True if watch.get('uuid') in running_uuids else False,
'checking_now': True if watch_uuid in running_uuids else False,
'fetch_time': watch.get('fetch_time'),
'has_error': True if error_texts else False,
'last_changed': watch.get('last_changed'),
'last_checked': watch.get('last_checked'),
'error_text': error_texts,
'last_checked_text': _jinja2_filter_datetime(watch),
'last_changed_text': timeago.format(int(watch['last_changed']), time.time()) if watch.history_n >= 2 and int(watch.get('last_changed', 0)) > 0 else 'Not yet',
'queued': True if watch.get('uuid') in queue_list else False,
'last_changed_text': timeago.format(int(watch['last_changed']), time.time()) if watch.history_n >= 2 and int(
watch.get('last_changed', 0)) > 0 else 'Not yet',
'queued': True if watch_uuid in queue_list else False,
'paused': True if watch.get('paused') else False,
'notification_muted': True if watch.get('notification_muted') else False,
'unviewed': watch.has_unviewed,
'uuid': watch.get('uuid'),
'uuid': watch_uuid,
'event_timestamp': time.time()
}
errored_count =0
for uuid, watch in datastore.data['watching'].items():
if watch.get('last_error'):
errored_count = 0
for watch_uuid_iter, watch_iter in datastore.data['watching'].items():
if watch_iter.get('last_error'):
errored_count += 1
general_stats = {
@@ -169,13 +166,13 @@ def handle_watch_update(socketio, **kwargs):
}
# Debug what's being emitted
#logger.debug(f"Emitting 'watch_update' event for {watch.get('uuid')}, data: {watch_data}")
# logger.debug(f"Emitting 'watch_update' event for {watch.get('uuid')}, data: {watch_data}")
# Emit to all clients (no 'broadcast' parameter needed - it's the default behavior)
socketio.emit("watch_update", {'watch': watch_data, 'general_stats': general_stats})
# Log after successful emit
logger.info(f"Socket.IO: Emitted update for watch {watch.get('uuid')}, Checking now: {watch_data['checking_now']}")
# Log after successful emit - use watch_data['uuid'] to avoid variable shadowing issues
logger.trace(f"Socket.IO: Emitted update for watch {watch_data['uuid']}, Checking now: {watch_data['checking_now']}")
except Exception as e:
logger.error(f"Socket.IO error in handle_watch_update: {str(e)}")
@@ -190,30 +187,31 @@ def init_socketio(app, datastore):
# Restrict SocketIO CORS to same origin by default, can be overridden with env var
cors_origins = os.environ.get('SOCKETIO_CORS_ORIGINS', None)
socketio = SocketIO(app,
async_mode=async_mode,
cors_allowed_origins=cors_origins, # None means same-origin only
logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')),
engineio_logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')))
async_mode=async_mode,
cors_allowed_origins=cors_origins, # None means same-origin only
logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')),
engineio_logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')))
# Set up event handlers
logger.info("Socket.IO: Registering connect event handler")
@socketio.on('connect')
def handle_connect():
"""Handle client connection"""
# logger.info("Socket.IO: CONNECT HANDLER CALLED - Starting connection process")
# logger.info("Socket.IO: CONNECT HANDLER CALLED - Starting connection process")
from flask import request
from flask_login import current_user
from changedetectionio.flask_app import update_q
# Access datastore from socketio
datastore = socketio.datastore
# logger.info(f"Socket.IO: Current user authenticated: {current_user.is_authenticated if hasattr(current_user, 'is_authenticated') else 'No current_user'}")
# logger.info(f"Socket.IO: Current user authenticated: {current_user.is_authenticated if hasattr(current_user, 'is_authenticated') else 'No current_user'}")
# Check if authentication is required and user is not authenticated
has_password_enabled = datastore.data['settings']['application'].get('password') or os.getenv("SALTED_PASS", False)
# logger.info(f"Socket.IO: Password enabled: {has_password_enabled}")
# logger.info(f"Socket.IO: Password enabled: {has_password_enabled}")
if has_password_enabled and not current_user.is_authenticated:
logger.warning("Socket.IO: Rejecting unauthenticated connection")
return False # Reject the connection
@@ -231,7 +229,7 @@ def init_socketio(app, datastore):
logger.info("Socket.IO: Client connected")
# logger.info("Socket.IO: Registering disconnect event handler")
# logger.info("Socket.IO: Registering disconnect event handler")
@socketio.on('disconnect')
def handle_disconnect():
"""Handle client disconnection"""
@@ -246,24 +244,23 @@ def init_socketio(app, datastore):
# Store the datastore reference on the socketio object for later use
socketio.datastore = datastore
# Create a stop event for our queue update thread using eventlet Event
import eventlet.event
stop_event = eventlet.event.Event()
socketio.stop_event = stop_event
# Add a shutdown method to the socketio object
def shutdown():
"""Shutdown the SocketIO server gracefully"""
try:
logger.info("Socket.IO: Shutting down server...")
# Signal the queue update thread to stop
if hasattr(socketio, 'stop_event'):
socketio.stop_event.send()
logger.info("Socket.IO: Signaled queue update thread to stop")
# Wait for the greenlet to exit (with timeout)
if hasattr(socketio, 'polling_emitter_thread'):
try:
@@ -275,14 +272,14 @@ def init_socketio(app, datastore):
logger.info("Socket.IO: Queue update eventlet greenlet already dead")
except Exception as e:
logger.error(f"Error killing eventlet greenlet: {str(e)}")
# Close any remaining client connections
#if hasattr(socketio, 'server'):
# if hasattr(socketio, 'server'):
# socketio.server.disconnect()
logger.info("Socket.IO: Server shutdown complete")
except Exception as e:
logger.error(f"Socket.IO error during shutdown: {str(e)}")
# Attach the shutdown method to the socketio object
socketio.shutdown = shutdown