CCCHH/dooris
1
1
Fork 0
Code Issues 1 Pull requests 1 Releases Packages 1 Activity Actions

api: fix mqtt reconnect logic errors

Browse source
This commit is contained in:
lilly 2026-05-31 20:53:14 +02:00
commit b752888812
Signed by: lilly
SSH key fingerprint: SHA256:y9T5GFw2A20WVklhetIxG1+kcg/Ce0shnQmbu1LQ37g
2 changed files with 23 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

9
api/src/dooris_api/ccujack.py
View file

@ -162,15 +162,18 @@ class CCUJackClient:
while True: while True:
try: try:
await asyncio.sleep(15 * 60) # 15 minutes
logger.info("Running CCUJack cron") logger.info("Running CCUJack cron")
await self.find_locks() if not self.mqtt.is_connected:
if not self.mqtt.is_connected():
logger.warning("MQTT client was discovered to be disconnected; reconnecting now") logger.warning("MQTT client was discovered to be disconnected; reconnecting now")
await self.mqtt.connect() await self.mqtt.connect()
await self.find_locks()
except Exception as e: except Exception as e:
logger.exception(f"Error in CCUJack cron task: {e}") logger.exception(f"Error in CCUJack cron task: {e}")
finally:
await asyncio.sleep(15 * 60) # 15 minutes
async def query_param_value(self, address: str) -> CCUValue: async def query_param_value(self, address: str) -> CCUValue:
if address in self.param_values: if address in self.param_values:

14
api/src/dooris_api/mqtt_client.py
View file

@ -128,19 +128,21 @@ class AsyncMqttClient:
to_add = topics.difference(self.active_subscriptions) to_add = topics.difference(self.active_subscriptions)
if to_add: if to_add:
self.active_subscriptions.update(to_add)
if self.is_connected:
logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}") logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}")
self.fut_subscribe = asyncio.get_running_loop().create_future() self.fut_subscribe = asyncio.get_running_loop().create_future()
self.client.subscribe([(i, qos) for i in to_add]) self.client.subscribe([(i, qos) for i in to_add])
await self.fut_subscribe await self.fut_subscribe
self.active_subscriptions.update(to_add)
to_remove = self.active_subscriptions.difference(topics) to_remove = self.active_subscriptions.difference(topics)
if to_remove: if to_remove:
self.active_subscriptions.difference_update(to_remove)
if self.is_connected:
logger.info(f"mqtt client unsubscribing from topics {','.join(to_remove)}") logger.info(f"mqtt client unsubscribing from topics {','.join(to_remove)}")
self.fut_unsubscribe = asyncio.get_running_loop().create_future() self.fut_unsubscribe = asyncio.get_running_loop().create_future()
self.client.unsubscribe(list(to_remove)) self.client.unsubscribe(list(to_remove))
await self.fut_unsubscribe await self.fut_unsubscribe
self.active_subscriptions.difference_update(to_remove)
async def connect(self): async def connect(self):
server_host, server_port = self.connection_string.rsplit(":", maxsplit=1) server_host, server_port = self.connection_string.rsplit(":", maxsplit=1)
@ -157,13 +159,19 @@ class AsyncMqttClient:
# re-establish all supposed mqtt subscriptions # re-establish all supposed mqtt subscriptions
if len(self.active_subscriptions) > 0: if len(self.active_subscriptions) > 0:
qos = 1 qos = 1
await self.client.subscribe((i, qos) for i in self.active_subscriptions) self.fut_subscribe = asyncio.get_running_loop().create_future()
self.client.subscribe([(i, qos) for i in self.active_subscriptions])
await self.fut_subscribe
async def disconnect(self): async def disconnect(self):
if not self.is_connected:
return
logger.info("Disconnecting mqtt client from broker") logger.info("Disconnecting mqtt client from broker")
self.fut_disconnect = asyncio.get_running_loop().create_future() self.fut_disconnect = asyncio.get_running_loop().create_future()
self.client.disconnect() self.client.disconnect()
await self.fut_disconnect await self.fut_disconnect
@property
def is_connected(self) -> bool: def is_connected(self) -> bool:
return self.client.is_connected() return self.client.is_connected()