MQTT and nodes update
This commit is contained in:
@@ -208,13 +208,14 @@ class MQTTHandler:
|
|||||||
# Outbound — send a command to a specific node
|
# Outbound — send a command to a specific node
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
def send_command(self, node_id: str, payload: dict):
|
def send_command(self, node_id: str, payload: dict) -> bool:
|
||||||
topic = f"nodes/{node_id}/commands"
|
topic = f"nodes/{node_id}/commands"
|
||||||
if self._client and self._connected:
|
if self._client and self._connected:
|
||||||
self._client.publish(topic, json.dumps(payload), qos=1)
|
self._client.publish(topic, json.dumps(payload), qos=1)
|
||||||
logger.info(f"Command sent to {node_id}: {payload.get('action')}")
|
logger.info(f"Command sent to {node_id}: {payload.get('action')}")
|
||||||
else:
|
return True
|
||||||
logger.warning(f"MQTT not connected — could not send command to {node_id}")
|
logger.warning(f"MQTT not connected — could not send command to {node_id}")
|
||||||
|
return False
|
||||||
|
|
||||||
def push_config(self, node_id: str, system_config: dict):
|
def push_config(self, node_id: str, system_config: dict):
|
||||||
topic = f"nodes/{node_id}/config"
|
topic = f"nodes/{node_id}/config"
|
||||||
@@ -240,12 +241,22 @@ class MQTTHandler:
|
|||||||
async def connect(self):
|
async def connect(self):
|
||||||
self._loop = asyncio.get_event_loop()
|
self._loop = asyncio.get_event_loop()
|
||||||
self._client = self._build_client()
|
self._client = self._build_client()
|
||||||
|
# Start the paho network loop first so it drives reconnects automatically,
|
||||||
|
# then keep attempting the initial TCP connect until it succeeds.
|
||||||
|
self._client.loop_start()
|
||||||
|
asyncio.create_task(self._connect_with_retry())
|
||||||
|
|
||||||
|
async def _connect_with_retry(self):
|
||||||
|
delay = 5
|
||||||
|
logger.info(f"MQTT connecting to {settings.mqtt_broker}:{settings.mqtt_port}")
|
||||||
|
while True:
|
||||||
try:
|
try:
|
||||||
self._client.connect(settings.mqtt_broker, settings.mqtt_port, keepalive=60)
|
self._client.connect(settings.mqtt_broker, settings.mqtt_port, keepalive=60)
|
||||||
self._client.loop_start()
|
return # paho loop_start + reconnect_delay_set handles the rest
|
||||||
logger.info(f"MQTT connecting to {settings.mqtt_broker}:{settings.mqtt_port}")
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"MQTT connection error: {e}")
|
logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s")
|
||||||
|
await asyncio.sleep(delay)
|
||||||
|
delay = min(delay * 2, 60)
|
||||||
|
|
||||||
async def disconnect(self):
|
async def disconnect(self):
|
||||||
if self._client:
|
if self._client:
|
||||||
|
|||||||
@@ -62,7 +62,8 @@ async def send_command(node_id: str, cmd: CommandPayload):
|
|||||||
elif cmd.action == "discord_leave":
|
elif cmd.action == "discord_leave":
|
||||||
await release_token(node_id)
|
await release_token(node_id)
|
||||||
|
|
||||||
mqtt_handler.send_command(node_id, payload)
|
if not mqtt_handler.send_command(node_id, payload):
|
||||||
|
raise HTTPException(503, "MQTT broker unavailable — command not delivered.")
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user