dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-07-29 06:49:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor MQTT connection handling in PacketCaptureService and WeatherService to be non-blocking

Browse Source 
- Updated connection and reconnection logic in PacketCaptureService to use asyncio's run_in_executor, preventing blocking of the event loop during MQTT operations.
- Enhanced WeatherService to implement non-blocking connection and subscription handling, ensuring smoother operation and improved error logging for connection failures.
This commit is contained in:
agessaman
2025-12-27 09:35:34 -08:00
parent 35bc6260a4
commit d7f16f31e4
2 changed files with 37 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/service_plugins/packet_capture_service.py
+21 -4
View File
@@ -1030,10 +1030,22 @@ class PacketCaptureService(BaseServicePlugin):
ws_path = broker_config.get('websocket_path', '/mqtt')
self.logger.debug(f"Connecting to MQTT broker {host}:{port} via WebSockets (path: {ws_path}, TLS: {broker_config.get('use_tls', False)})")
# For WebSockets, connect without path parameter (path set via ws_set_options)
client.connect(host, port, keepalive=60)
# Run connect in executor to avoid blocking the event loop
loop = asyncio.get_event_loop()
try:
await loop.run_in_executor(None, client.connect, host, port, 60)
except Exception as connect_error:
# Connection failed, but don't block - let loop_start handle retries
self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}")
else:
self.logger.debug(f"Connecting to MQTT broker {host}:{port} via TCP (TLS: {broker_config.get('use_tls', False)})")
client.connect(host, port, keepalive=60)
# Run connect in executor to avoid blocking the event loop
loop = asyncio.get_event_loop()
try:
await loop.run_in_executor(None, client.connect, host, port, 60)
except Exception as connect_error:
# Connection failed, but don't block - let loop_start handle retries
self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}")
# Start network loop (non-blocking)
client.loop_start()
@@ -1634,10 +1646,15 @@ class PacketCaptureService(BaseServicePlugin):
except Exception as e:
self.logger.debug(f"Error renewing auth token for {broker_host}: {e}")
# Attempt reconnection
# Attempt reconnection (non-blocking to avoid blocking event loop)
host = config['host']
port = config['port']
client.reconnect()
loop = asyncio.get_event_loop()
try:
await loop.run_in_executor(None, client.reconnect)
except Exception as reconnect_error:
# Reconnection failed, but don't block - will retry on next cycle
self.logger.debug(f"Reconnect() call failed (non-blocking): {reconnect_error}")
# Give it a moment to connect
await asyncio.sleep(2)
modules/service_plugins/weather_service.py
+16 -3
View File
@@ -637,9 +637,22 @@ class WeatherService(BaseServicePlugin):
client.on_message = on_message
# Connect and subscribe
client.connect(broker_host, broker_port, 60)
client.subscribe(topic)
# Connect and subscribe (non-blocking to avoid blocking event loop)
loop = asyncio.get_event_loop()
try:
await loop.run_in_executor(None, client.connect, broker_host, broker_port, 60)
except Exception as connect_error:
# Connection failed, but don't block - will retry on next cycle
self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}")
raise # Re-raise to trigger retry logic
# Subscribe is non-blocking, but wrap it anyway for consistency
try:
client.subscribe(topic)
except Exception as subscribe_error:
self.logger.debug(f"Subscribe() call failed: {subscribe_error}")
raise
client.loop_start()
self.logger.info(f"Connected to Blitzortung MQTT, subscribed to {topic}")