Fix unawaited critical_tasks cancel, use None for infinite reconnects.

Author: flyte

Makefile

@@ -12,12 +12,12 @@ graphs:
 	dot -Tsvg -O interrupt_callbacks.dot
 
 coverage:
-	coverage run --source "." --omit "mqtt_io/tests/*,mqtt_io/modules/*" -m behave mqtt_io/tests/features -t ~skip
-	coverage report -m
+	poetry run coverage run --source "." --omit "mqtt_io/tests/*,mqtt_io/modules/*" -m behave mqtt_io/tests/features -t ~skip
+	poetry run coverage report -m
 
 lint:
-	pylint -d fixme mqtt_io
-	mypy --show-error-codes --strict --no-warn-unused-ignores mqtt_io
+	poetry run pylint -d fixme mqtt_io
+	poetry run mypy --show-error-codes --strict --no-warn-unused-ignores mqtt_io
 
 build:
 	poetry build

config.stdio.yml

@@ -1,8 +1,11 @@
 mqtt:
-  host: test.mosquitto.org
+  host: localhost
   topic_prefix: mqtt_io
   clean_session: yes
   protocol: 3.1.1
+  keepalive: 10
+  reconnect_delay: 4
+  reconnect_count: 4
 
 gpio_modules:
   - name: stdio
@@ -25,6 +28,7 @@ digital_inputs:
     interrupt: rising
     interrupt_for:
       - test
+    poll_interval: 2
 
 digital_outputs:
   - name: bell
@@ -35,6 +39,15 @@ digital_outputs:
     publish_initial: yes
     initial: high
 
+sensor_modules:
+  - name: mock
+    module: mock
+
+sensor_inputs:
+  - name: mock_sensor
+    module: mock
+    interval: 2
+
 logging:
   version: 1
   handlers:

docs/config/reference/mqtt/README.md

@@ -433,13 +433,13 @@ Default: 2
 *mqtt*.**reconnect_count**
 
 Max number of retries of connections before giving up and exiting.
--1 (the default) means infinite reconnects.
+Null value means infinite reconnects (default).
 The counter is reset when the connection is reestablished successfully.
 
 
 ```yaml
 Type: integer
 Required: False
-Default: -1
+Default: None
 ```
 

docs/schema.json

@@ -284,13 +284,14 @@
       },
       "reconnect_count": {
         "meta": {
-          "description": "Max number of retries of connections before giving up and exiting.\n-1 (the default) means infinite reconnects.\nThe counter is reset when the connection is reestablished successfully.\n",
+          "description": "Max number of retries of connections before giving up and exiting.\nNull value means infinite reconnects (default).\nThe counter is reset when the connection is reestablished successfully.\n",
           "title_id": "mqtt-reconnect_count"
         },
         "type": "integer",
         "required": false,
-        "default": -1,
-        "min": -1
+        "default": null,
+        "nullable": true,
+        "min": 0
       }
     }
   },

mqtt_io/config/config.schema.yml

@@ -262,12 +262,13 @@ mqtt:
       meta:
         description: |
           Max number of retries of connections before giving up and exiting.
-          -1 (the default) means infinite reconnects.
+          Null value means infinite reconnects (default).
           The counter is reset when the connection is reestablished successfully.
       type: integer
       required: no
-      default: -1
-      min: -1
+      default: null
+      nullable: yes
+      min: 0
 gpio_modules:
   meta:
     description: |

mqtt_io/mqtt/asyncio_mqtt.py

@@ -66,6 +66,7 @@ def __init__(self, options: MQTTClientOptions):
             username=options.username,
             password=options.password,
             client_id=options.client_id,
+            # keepalive=options.keepalive,  # This isn't implemented yet on 0.8.1
             tls_context=tls_context,
             protocol=protocol_map[options.protocol],
             will=will,

mqtt_io/server.py

@@ -1030,6 +1030,8 @@ async def _mqtt_task_loop(self) -> None:
             )
             try:
                 await entry.coro
+            except MQTTException:
+                raise
             except Exception:  # pylint: disable=broad-except
                 _LOG.exception("Exception while handling MQTT task:")
             self.mqtt_task_queue.task_done()
@@ -1149,13 +1151,12 @@ async def stream_output_loop(
                 self.event_bus.fire(StreamDataSentEvent(stream_conf["name"], data))
 
     async def _main_loop(self) -> None:
-        counter = self.config["mqtt"].get("reconnect_count")
         reconnect = True
+        reconnect_delay = self.config["mqtt"]["reconnect_delay"]
         while reconnect:
             try:
                 await self._connect_mqtt()
-                # Reset counter
-                counter = self.config["mqtt"].get("reconnect_count")
+                reconnects_remaining = self.config["mqtt"]["reconnect_count"]
                 self.critical_tasks = [
                     self.loop.create_task(coro)
                     for coro in (
@@ -1179,18 +1180,29 @@ async def _main_loop(self) -> None:
             except asyncio.CancelledError:
                 break
             except MQTTException:
-                reconnect = counter > 0
-                if counter > 0:
-                    counter -= 1
+                if reconnects_remaining is not None:
+                    reconnect = reconnects_remaining > 0
+                    reconnects_remaining -= 1
                 _LOG.exception("Connection to MQTT broker failed")
 
             finally:
+                _LOG.debug("Clearing events and cancelling 'critical_tasks'")
                 self.running.clear()
                 self.mqtt_connected.clear()
                 if self.critical_tasks:
-                    asyncio.gather(*self.critical_tasks, return_exceptions=True).cancel()
+                    for task in self.critical_tasks:
+                        task.cancel()
+                    await asyncio.gather(*self.critical_tasks, return_exceptions=True)
+                    _LOG.debug("'critical_tasks' cancelled and gathered")
             if reconnect:
-                await asyncio.sleep(self.config["mqtt"].get("reconnect_delay"))
+                _LOG.debug(
+                    "Waiting for MQTT reconnect delay (%s second(s))", reconnect_delay
+                )
+                await asyncio.sleep(reconnect_delay)
+                _LOG.info(
+                    "Reconnecting to MQTT broker (%s retries remaining)",
+                    "infinite" if reconnects_remaining is None else reconnects_remaining,
+                )
         await self.shutdown()
 
     # Main entry point