Add unsubscribe mechanism and enhance configuration dynamic loading:

- Implement `unsubscribe` method in `ProcessingQueue` for subscriber cleanup.
- Refactor subscriber ID handling in forwarders.
- Improve configuration loading with validation for `.yaml` files.
- Comment out unsupported event handling logic.

Author: U039b

mongoose/core/processing.py

@@ -1,6 +1,6 @@
 import enum
-from collections import defaultdict
 import logging
+from collections import defaultdict
 from queue import Queue, Full
 from threading import Event
 from typing import Any, Dict, List
@@ -74,7 +74,6 @@ def publish(self, topic: ProcessingTopic, data: Any):
                 q.put_nowait(data)
             except Full as e:
                 logger.error(f"Failed to publish data to {topic}: {e}")
-                raise e
 
     def subscribe(self, topic: ProcessingTopic | List[ProcessingTopic], subscriber_id: str, queue_size=100) -> Queue:
         """Subscribe to one or more topics and receive a dedicated queue for receiving data.
@@ -118,6 +117,29 @@ def subscribe(self, topic: ProcessingTopic | List[ProcessingTopic], subscriber_i
             self.queues[t].append(q)
         return q
 
+    def unsubscribe(self, subscriber_id: str):
+        """Completely unsubscribe a subscriber and remove all associated queues.
+
+        This method removes the subscriber from all topics they were subscribed to
+        and cleans up their dedicated queues. If a topic has no more subscribers
+        after this operation, it is removed from the system.
+
+        Args:
+            subscriber_id: Unique identifier for the subscriber to unsubscribe.
+        """
+        logger.info(f"Unsubscribing {subscriber_id}")
+        if subscriber_id not in self.subscribers:
+            logger.warning(f"Subscriber {subscriber_id} not found")
+            return
+
+        subscribed_topics = self.subscribers.pop(subscriber_id)
+        for topic, q in subscribed_topics.items():
+            if topic in self.queues:
+                if q in self.queues[topic]:
+                    self.queues[topic].remove(q)
+                if not self.queues[topic]:
+                    del self.queues[topic]
+
     def stop_processing(self):
         """Signal all processing operations to stop.
 

mongoose/core/watchdogs.py

@@ -32,6 +32,8 @@ def __init__(self, config_class: Type, callback: Callable):
         super().__init__()
 
     def _load_configuration(self, config_file: Path):
+        if not config_file.name.endswith(".yaml"):
+            return None
         with config_file.open(mode="r") as f:
             config_data = yaml.safe_load(f)
         if config_data:
@@ -53,10 +55,11 @@ def on_any_event(self, event: FileSystemEvent):
             new_configuration = self._load_configuration(src_file)
             if new_configuration:
                 created.append(new_configuration)
-        elif event.event_type == "modified":
-            new_configuration = self._load_configuration(src_file)
-            if new_configuration:
-                modified.append(new_configuration)
+        # Not supported yet
+        # elif event.event_type == "modified":
+        #     new_configuration = self._load_configuration(src_file)
+        #     if new_configuration:
+        #         modified.append(new_configuration)
         elif event.event_type == "deleted":
             deleted.append(src_file)
 

mongoose/forward/base.py

@@ -51,6 +51,10 @@ def __init__(self, topics: List[str]):
         self.thread: Optional[threading.Thread] = None
         self.queue = None
 
+    @property
+    def subscriber_id(self):
+        return f"{self.__class__.__name__.lower()}_{id(self)}"
+
     def start(self):
         """Start the forwarder worker thread.
 
@@ -66,8 +70,7 @@ def start(self):
             return
 
         # Unique subscriber ID to avoid collisions
-        subscriber_id = f"{self.__class__.__name__.lower()}_{id(self)}"
-        self.queue = self.processing_queue.subscribe(topics, subscriber_id=subscriber_id)
+        self.queue = self.processing_queue.subscribe(topics, subscriber_id=self.subscriber_id)
 
         self.thread = threading.Thread(target=self._run, daemon=True)
         self.thread.start()

mongoose/forward/webhook.py

@@ -232,6 +232,7 @@ def _forward_batch(self, batch: list):
     def disable(self):
         """Disable the forwarder."""
         self.config.enable = False
+        self.processing_queue.unsubscribe(self.subscriber_id)
 
     def forward(self, data: Any):
         """Send formatted data to the webhook URL with retries.

tests/test_processing_unsubscribe.py

@@ -0,0 +1,70 @@
+import pytest
+
+from mongoose.core.processing import ProcessingQueue, ProcessingTopic
+from mongoose.utils.exceptions import TopicNotFoundException
+
+
+def test_unsubscribe():
+    pq = ProcessingQueue()
+    subscriber_id = "test_subscriber"
+    topic = ProcessingTopic.NETWORK_DPI
+
+    # Reset singleton-like behavior if any (though it looks like a normal class with class attributes)
+    # Actually, subscribers and queues are class attributes, which is a bit strange if multiple instances are intended.
+    # Let's check if they are indeed class attributes.
+    ProcessingQueue.subscribers.clear()
+    ProcessingQueue.queues.clear()
+
+    # Subscribe
+    pq.subscribe(topic, subscriber_id)
+    assert subscriber_id in pq.subscribers
+    assert topic in pq.queues
+    assert len(pq.queues[topic]) == 1
+
+    # Unsubscribe (method not yet implemented)
+    if hasattr(pq, 'unsubscribe'):
+        pq.unsubscribe(subscriber_id)
+        assert subscriber_id not in pq.subscribers
+        assert topic not in pq.queues or len(pq.queues[topic]) == 0
+
+        # Verify that publishing to the topic now raises TopicNotFoundException if no other subscribers
+        with pytest.raises(TopicNotFoundException):
+            pq.publish(topic, "test data")
+    else:
+        pytest.fail("ProcessingQueue has no unsubscribe method")
+
+
+def test_unsubscribe_not_found():
+    pq = ProcessingQueue()
+    ProcessingQueue.subscribers.clear()
+    ProcessingQueue.queues.clear()
+
+    # Should not raise any exception
+    pq.unsubscribe("non_existent_subscriber")
+
+
+def test_unsubscribe_partial_others_remain():
+    pq = ProcessingQueue()
+    sub1 = "sub1"
+    sub2 = "sub2"
+    topic = ProcessingTopic.NETWORK_DPI
+
+    ProcessingQueue.subscribers.clear()
+    ProcessingQueue.queues.clear()
+
+    pq.subscribe(topic, sub1)
+    pq.subscribe(topic, sub2)
+
+    assert len(pq.queues[topic]) == 2
+
+    pq.unsubscribe(sub1)
+
+    assert sub1 not in pq.subscribers
+    assert sub2 in pq.subscribers
+    assert topic in pq.queues
+    assert len(pq.queues[topic]) == 1
+
+    # Can still publish to sub2
+    pq.publish(topic, "data")
+    q2 = pq.subscribers[sub2][topic]
+    assert q2.get_nowait() == "data"