Fix adding handlers at every reconnect cycle. Improve disconnect error handling.

Author: cbpowell

senselink/mqtt/mqtt_controller.py

@@ -24,31 +24,12 @@ def __init__(self, host, port=1883, username=None, password=None):
         self.data_sources = []
         self.listeners = {}
 
+        self.listen_task = None
     async def connect(self):
         # Create task
         await self.client_handler()
 
     async def client_handler(self):
-        logging.info(f"Starting MQTT client to URL: {self.host}")
-        reconnect_interval = 5  # [seconds]
-
-        client = Client(self.host, self.port, username=self.username, password=self.password)
-        while True:
-            try:
-                async with client:
-                    await self.listen(client)
-            except MqttError as error:
-                logging.error(f'Disconnected from MQTT broker with error: {error}')
-                logging.debug(f'MQTT client disconnected/ended, reconnecting in {reconnect_interval}...')
-                await asyncio.sleep(reconnect_interval)
-            except (KeyboardInterrupt, asyncio.CancelledError):
-                return False
-            except Exception as error:
-                logging.error(f'Stopping MQTT client with error: {error}')
-                return False
-
-    async def listen(self, client):
-        logging.info(f'MQTT client connected')
         # Add tasks for each data source handler
         for ds in self.data_sources:
             # Get handlers from data source
@@ -68,19 +49,46 @@ async def listen(self, client):
                     logging.debug(f'Creating new prime Listener for topic: {topic}')
                     self.listeners[topic] = MQTTListener(topic, funcs)
 
-        async with client.messages() as messages:
-            # Subscribe to specified topics
-            for topic, handlers in self.listeners.items():
-                await client.subscribe(topic)
-            # Handle messages that come in
-            async for message in messages:
-                topic = message.topic.value
-                handlers = self.listeners[topic].handlers
-                logging.debug(f'Got message for topic: {topic}')
-                for func in handlers:
-                    # Decode to UTF-8
-                    payload = message.payload.decode()
-                    await func(payload)
+        logging.info(f"Starting MQTT client to URL: {self.host}")
+        reconnect_interval = 10  # [seconds]
+        loop = asyncio.get_event_loop()
+
+        while True:
+            try:
+                # Listen for MQTT messages in (unawaited) asyncio task
+                self.listen_task = loop.create_task(self.listen())
+                await self.listen_task
+            except MqttError as error:
+                logging.error(f'Disconnected from MQTT broker with error: {error}')
+                logging.debug(f'MQTT client disconnected/ended, reconnecting in {reconnect_interval}...')
+                # Cancel task and wait
+                self.listen_task.cancel()
+                await asyncio.sleep(reconnect_interval)
+            except Exception as error:
+                logging.error(f'Stopping MQTT client with error: {error}')
+                logging.debug(f'MQTT client disconnected/ended, reconnecting in {reconnect_interval}...')
+                # Cancel task and wait
+                self.listen_task.cancel()
+                await asyncio.sleep(reconnect_interval)
+            except (KeyboardInterrupt, asyncio.CancelledError):
+                return False
+
+    async def listen(self):
+        logging.info(f'MQTT client connected')
+        async with Client(self.host, self.port, username=self.username, password=self.password) as client:
+            async with client.messages() as messages:
+                # Subscribe to specified topics
+                for topic, handlers in self.listeners.items():
+                    await client.subscribe(topic)
+                # Handle messages that come in
+                async for message in messages:
+                    topic = message.topic.value
+                    handlers = self.listeners[topic].handlers
+                    logging.debug(f'Got message for topic: {topic}')
+                    for func in handlers:
+                        # Decode to UTF-8
+                        payload = message.payload.decode()
+                        await func(payload)
 
 
 if __name__ == "__main__":