552. Subscribes to all event topics
663. Routes events to appropriate handlers
774. Persists data to database
8+ 5. Dispatches events to configured webhooks
89"""
910
11+ import asyncio
1012import logging
1113import signal
1214import threading
1315import time
14- from typing import Any , Callable , Optional
16+ from typing import Any , Callable , Optional , TYPE_CHECKING
1517
1618from meshcore_hub .common .database import DatabaseManager
1719from meshcore_hub .common .health import HealthReporter
1820from meshcore_hub .common .mqtt import MQTTClient , MQTTConfig
1921
22+ if TYPE_CHECKING :
23+ from meshcore_hub .collector .webhook import WebhookDispatcher
24+
2025logger = logging .getLogger (__name__ )
2126
2227
@@ -31,21 +36,28 @@ def __init__(
3136 self ,
3237 mqtt_client : MQTTClient ,
3338 db_manager : DatabaseManager ,
39+ webhook_dispatcher : Optional ["WebhookDispatcher" ] = None ,
3440 ):
3541 """Initialize subscriber.
3642
3743 Args:
3844 mqtt_client: MQTT client instance
3945 db_manager: Database manager instance
46+ webhook_dispatcher: Optional webhook dispatcher for event forwarding
4047 """
4148 self .mqtt = mqtt_client
4249 self .db = db_manager
50+ self ._webhook_dispatcher = webhook_dispatcher
4351 self ._running = False
4452 self ._shutdown_event = threading .Event ()
4553 self ._handlers : dict [str , EventHandler ] = {}
4654 self ._mqtt_connected = False
4755 self ._db_connected = False
4856 self ._health_reporter : Optional [HealthReporter ] = None
57+ # Webhook processing
58+ self ._webhook_queue : list [tuple [str , dict [str , Any ], str ]] = []
59+ self ._webhook_lock = threading .Lock ()
60+ self ._webhook_thread : Optional [threading .Thread ] = None
4961
5062 @property
5163 def is_healthy (self ) -> bool :
@@ -117,6 +129,78 @@ def _handle_mqtt_message(
117129 except Exception as e :
118130 logger .error (f"Error logging event { event_type } { e } )
119131
132+ # Queue event for webhook dispatch
133+ if self ._webhook_dispatcher and self ._webhook_dispatcher .webhooks :
134+ self ._queue_webhook_event (event_type , payload , public_key )
135+
136+ def _queue_webhook_event (
137+ self , event_type : str , payload : dict [str , Any ], public_key : str
138+ ) -> None :
139+ """Queue an event for webhook dispatch.
140+
141+ Args:
142+ event_type: Event type name
143+ payload: Event payload
144+ public_key: Source node public key
145+ """
146+ with self ._webhook_lock :
147+ self ._webhook_queue .append ((event_type , payload , public_key ))
148+
149+ def _start_webhook_processor (self ) -> None :
150+ """Start background thread for webhook processing."""
151+ if not self ._webhook_dispatcher or not self ._webhook_dispatcher .webhooks :
152+ return
153+
154+ # Capture dispatcher in local variable for closure (avoids Optional issues)
155+ dispatcher = self ._webhook_dispatcher
156+
157+ def run_webhook_loop () -> None :
158+ """Run async webhook dispatch in background thread."""
159+ loop = asyncio .new_event_loop ()
160+ asyncio .set_event_loop (loop )
161+
162+ try :
163+ loop .run_until_complete (dispatcher .start ())
164+ logger .info ("Webhook processor started" )
165+
166+ while self ._running :
167+ # Get queued events
168+ events_to_process : list [tuple [str , dict [str , Any ], str ]] = []
169+ with self ._webhook_lock :
170+ if self ._webhook_queue :
171+ events_to_process = self ._webhook_queue .copy ()
172+ self ._webhook_queue .clear ()
173+
174+ # Process events
175+ for event_type , payload , public_key in events_to_process :
176+ try :
177+ loop .run_until_complete (
178+ dispatcher .dispatch (event_type , payload , public_key )
179+ )
180+ except Exception as e :
181+ logger .error (f"Webhook dispatch error: { e } )
182+
183+ # Small sleep to prevent busy-waiting
184+ time .sleep (0.01 )
185+
186+ finally :
187+ loop .run_until_complete (dispatcher .stop ())
188+ loop .close ()
189+ logger .info ("Webhook processor stopped" )
190+
191+ self ._webhook_thread = threading .Thread (
192+ target = run_webhook_loop , daemon = True , name = "webhook-processor"
193+ )
194+ self ._webhook_thread .start ()
195+
196+ def _stop_webhook_processor (self ) -> None :
197+ """Stop the webhook processor thread."""
198+ if self ._webhook_thread and self ._webhook_thread .is_alive ():
199+ # Thread will exit when self._running becomes False
200+ self ._webhook_thread .join (timeout = 5.0 )
201+ if self ._webhook_thread .is_alive ():
202+ logger .warning ("Webhook processor thread did not stop cleanly" )
203+
120204 def start (self ) -> None :
121205 """Start the subscriber."""
122206 logger .info ("Starting collector subscriber" )
@@ -149,6 +233,9 @@ def start(self) -> None:
149233
150234 self ._running = True
151235
236+ # Start webhook processor if configured
237+ self ._start_webhook_processor ()
238+
152239 # Start health reporter for Docker health checks
153240 self ._health_reporter = HealthReporter (
154241 component = "collector" ,
@@ -181,6 +268,9 @@ def stop(self) -> None:
181268 self ._running = False
182269 self ._shutdown_event .set ()
183270
271+ # Stop webhook processor
272+ self ._stop_webhook_processor ()
273+
184274 # Stop health reporter
185275 if self ._health_reporter :
186276 self ._health_reporter .stop ()
@@ -201,6 +291,7 @@ def create_subscriber(
201291 mqtt_password : Optional [str ] = None ,
202292 mqtt_prefix : str = "meshcore" ,
203293 database_url : str = "sqlite:///./meshcore.db" ,
294+ webhook_dispatcher : Optional ["WebhookDispatcher" ] = None ,
204295) -> Subscriber :
205296 """Create a configured subscriber instance.
206297
@@ -211,6 +302,7 @@ def create_subscriber(
211302 mqtt_password: MQTT password
212303 mqtt_prefix: MQTT topic prefix
213304 database_url: Database connection URL
305+ webhook_dispatcher: Optional webhook dispatcher for event forwarding
214306
215307 Returns:
216308 Configured Subscriber instance
@@ -230,7 +322,7 @@ def create_subscriber(
230322 db_manager = DatabaseManager (database_url )
231323
232324 # Create subscriber
233- subscriber = Subscriber (mqtt_client , db_manager )
325+ subscriber = Subscriber (mqtt_client , db_manager , webhook_dispatcher )
234326
235327 # Register handlers
236328 from meshcore_hub .collector .handlers import register_all_handlers
@@ -247,6 +339,7 @@ def run_collector(
247339 mqtt_password : Optional [str ] = None ,
248340 mqtt_prefix : str = "meshcore" ,
249341 database_url : str = "sqlite:///./meshcore.db" ,
342+ webhook_dispatcher : Optional ["WebhookDispatcher" ] = None ,
250343) -> None :
251344 """Run the collector (blocking).
252345
@@ -257,6 +350,7 @@ def run_collector(
257350 mqtt_password: MQTT password
258351 mqtt_prefix: MQTT topic prefix
259352 database_url: Database connection URL
353+ webhook_dispatcher: Optional webhook dispatcher for event forwarding
260354 """
261355 subscriber = create_subscriber (
262356 mqtt_host = mqtt_host ,
@@ -265,6 +359,7 @@ def run_collector(
265359 mqtt_password = mqtt_password ,
266360 mqtt_prefix = mqtt_prefix ,
267361 database_url = database_url ,
362+ webhook_dispatcher = webhook_dispatcher ,
268363 )
269364
270365 # Set up signal handlers
0 commit comments