Merge pull request #32 from bcmi-labs/reconnects

ucloud: Restart MQTT tasks on disconnects.

Author: iabdalkader

.github/workflows/python-linter.yml

@@ -37,4 +37,4 @@ jobs:
       run: |
         # stop the build if there are Python syntax errors or undefined names
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        flake8 . --count --max-complexity=10 --max-line-length=120 --statistics
+        flake8 . --count --max-complexity=15 --max-line-length=120 --statistics

arduino_iot_cloud/ucloud.py

@@ -49,6 +49,10 @@ class InvalidStateError(Exception):
 _DEFAULT_PORT = (8883, 8884)
 
 
+class DoneException(Exception):
+    pass
+
+
 def timestamp():
     return int(time.time())
 
@@ -278,12 +282,16 @@ def mqtt_callback(self, topic, message):
         logging.debug(f"mqtt topic: {topic[-8:]}... message: {message[:8]}...")
         self.senmlpack.clear()
         for record in self.records.values():
+            # If the object is uninitialized, updates are always allowed even if it's a read-only
+            # object. Otherwise, for initialized objects, updates are only allowed if the object
+            # is writable (on_write function is set) and the value is received from the out topic.
             if not record.initialized or (record.on_write is not None and b"shadow" not in topic):
                 record.add_to_pack(self.senmlpack)
         self.senmlpack.from_cbor(message)
         self.senmlpack.clear()
 
     async def discovery_task(self, interval=0.100):
+        self.mqtt.subscribe(self.device_topic, qos=1)
         while self.thing_id is None:
             self.mqtt.check_msg()
             if self.records.get("thing_id").value is not None:
@@ -300,6 +308,25 @@ async def discovery_task(self, interval=0.100):
                     self.mqtt.publish(self.create_topic("shadow", "o"), self.senmlpack.to_cbor(), qos=1)
                 logging.info("Device configured via discovery protocol.")
             await asyncio.sleep(interval)
+        raise DoneException()
+
+    async def conn_task(self, interval=1.0, backoff=1.2):
+        logging.info("Connecting to Arduino IoT cloud...")
+        while True:
+            try:
+                self.mqtt.connect()
+                break
+            except Exception as e:
+                logging.warning(f"Connection failed {e}, retrying after {interval}s")
+                await asyncio.sleep(interval)
+                interval = min(interval * backoff, 4.0)
+
+        if self.thing_id is None:
+            self.create_task("discovery", self.discovery_task)
+        else:
+            self.mqtt.subscribe(self.create_topic("e", "i"))
+        self.create_task("mqtt_task", self.mqtt_task)
+        raise DoneException()
 
     async def mqtt_task(self, interval=0.100):
         while True:
@@ -320,34 +347,33 @@ async def mqtt_task(self, interval=0.100):
                     self.last_ping = timestamp()
                     logging.debug("No records to push, sent a ping request.")
             await asyncio.sleep(interval)
+        raise DoneException()
 
     async def run(self, user_main=None):
-        logging.info("Connecting to Arduino IoT cloud...")
-        if not self.mqtt.connect():
-            logging.error("Failed to connect Arduino IoT cloud.")
-            return
-
-        self.mqtt.subscribe(self.device_topic, qos=1)
+        self.create_task("conn_task", self.conn_task)
         if user_main is not None:
             self.create_task("user_main", user_main, self)
-        self.create_task("mqtt_task", self.mqtt_task)
-        self.create_task("discovery", self.discovery_task)
 
         while True:
+            task_except = None
             try:
                 await asyncio.gather(*self.tasks.values(), return_exceptions=False)
-                logging.info("All tasks finished!")
-                break
+                break   # All tasks are done, not likely.
             except Exception as e:
-                except_msg = str(e)
-                pass  # import traceback; traceback.print_exc()
+                task_except = e
+                pass    # import traceback; traceback.print_exc()
 
             for name in list(self.tasks):
                 task = self.tasks[name]
                 try:
                     if task.done():
                         self.tasks.pop(name)
                         self.records.pop(name, None)
-                        logging.error(f"Removed task: {name}. Raised exception: {except_msg}.")
+                        if isinstance(task_except, DoneException):
+                            logging.error(f"task: {name} complete.")
+                        elif task_except is not None:
+                            logging.error(f"task: {name} raised exception: {str(task_except)}.")
+                        if name == "mqtt_task":
+                            self.create_task("conn_task", self.conn_task)
                 except (CancelledError, InvalidStateError):
                     pass

arduino_iot_cloud/umqtt.py

@@ -22,8 +22,6 @@
 #
 # Based on: https://github.com/micropython/micropython-lib/tree/master/micropython/umqtt.simple
 
-import time
-
 try:
     from ussl import wrap_socket
     import usocket as socket
@@ -92,8 +90,13 @@ def set_last_will(self, topic, msg, retain=False, qos=0):
         self.lw_qos = qos
         self.lw_retain = retain
 
-    def _connect(self, clean_session=True):
+    def connect(self, clean_session=True):
         addr = socket.getaddrinfo(self.server, self.port)[0][-1]
+
+        if self.sock is not None:
+            self.sock.close()
+            self.sock = None
+
         try:
             self.sock = socket.socket()
             self.sock = wrap_socket(self.sock, **self.ssl_params)
@@ -130,7 +133,6 @@ def _connect(self, clean_session=True):
 
         self.sock.write(premsg[0:i + 2])
         self.sock.write(msg)
-        # print(hex(len(msg)), hexlify(msg, ":"))
         self._send_str(self.client_id)
         if self.lw_topic:
             self._send_str(self.lw_topic)
@@ -144,18 +146,6 @@ def _connect(self, clean_session=True):
             raise MQTTException(resp[3])
         return resp[2] & 1
 
-    def connect(self, retry=10, interval=1.0, clean_session=True):
-        for i in range(0, retry):
-            try:
-                self._connect(clean_session)
-                return True
-            except Exception as e:
-                if self.sock is not None:
-                    self.sock.close()
-                logging.warning(f"Connection failed {e}, retrying after {interval}s")
-                time.sleep(interval)
-        return False
-
     def disconnect(self):
         self.sock.write(b"\xe0\0")
         self.sock.close()