WIP

changedetectionio/custom_queue.py

@@ -1,4 +1,5 @@
 import queue
+import asyncio
 from blinker import signal
 from loguru import logger
 
@@ -50,3 +51,59 @@ class SignalPriorityQueue(queue.PriorityQueue):
         except Exception as e:
             logger.critical(f"Exception: {e}")
         return item
+
+
+class AsyncSignalPriorityQueue(asyncio.PriorityQueue):
+    """
+    Async version of SignalPriorityQueue that sends signals when items are added/removed.
+    
+    This class extends asyncio.PriorityQueue and maintains the same signal behavior
+    as the synchronous version for real-time UI updates.
+    """
+    
+    def __init__(self, maxsize=0):
+        super().__init__(maxsize)
+        try:
+            self.queue_length_signal = signal('queue_length')
+        except Exception as e:
+            logger.critical(f"Exception: {e}")
+
+    async def put(self, item):
+        # Call the parent's put method first
+        await super().put(item)
+        
+        # After putting the item in the queue, check if it has a UUID and emit signal
+        if hasattr(item, 'item') and isinstance(item.item, dict) and 'uuid' in item.item:
+            uuid = item.item['uuid']
+            # Get the signal and send it if it exists
+            watch_check_update = signal('watch_check_update')
+            if watch_check_update:
+                # Send the watch_uuid parameter
+                watch_check_update.send(watch_uuid=uuid)
+        
+        # Send queue_length signal with current queue size
+        try:
+            if self.queue_length_signal:
+                self.queue_length_signal.send(length=self.qsize())
+        except Exception as e:
+            logger.critical(f"Exception: {e}")
+
+    async def get(self):
+        # Call the parent's get method first
+        item = await super().get()
+        
+        # Send queue_length signal with current queue size
+        try:
+            if self.queue_length_signal:
+                self.queue_length_signal.send(length=self.qsize())
+        except Exception as e:
+            logger.critical(f"Exception: {e}")
+        return item
+    
+    @property
+    def queue(self):
+        """
+        Provide compatibility with sync PriorityQueue.queue access
+        Returns the internal queue for template access
+        """
+        return self._queue if hasattr(self, '_queue') else []