Refactor WorkerManager

dev-requirements.txt

@@ -4,4 +4,6 @@ pytest-mock
 pytest-asyncio
 bump2version
 pytest-automock
-pytest-cov
+pytest-cov
+asyncmock
+asynctest

octoprint_nanny/clients/mqtt.py

@@ -56,7 +56,7 @@ def __init__(
         private_key_file: str,
         ca_certs,
         algorithm="RS256",
-        remote_control_queue=None,
+        mqtt_receive_queue=None,
         mqtt_bridge_hostname=MQTT_BRIDGE_HOSTNAME,
         mqtt_bridge_port=MQTT_BRIDGE_PORT,
         on_connect=None,
@@ -69,7 +69,8 @@ def __init__(
         project_id=GCP_PROJECT_ID,
         region=IOT_DEVICE_REGISTRY_REGION,
         registry_id=IOT_DEVICE_REGISTRY,
-        tls_version=ssl.PROTOCOL_TLSv1_2,
+        tls_version=ssl.PROTOCOL_TLS,
+        keepalive=900,  # 15 minutes
         trace_context={},
         message_callbacks=[],  # see message_callback_add() https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#subscribe-unsubscribe
     ):
@@ -86,12 +87,13 @@ def __init__(
         self.ca_certs = ca_certs
         self.region = region
         self.registry_id = registry_id
+        self.keepalive = keepalive
 
         self.tls_version = tls_version
         self.region = region
         self.algorithm = algorithm
 
-        self.remote_control_queue = remote_control_queue
+        self.mqtt_receive_queue = mqtt_receive_queue
         self._honeycomb_tracer = HoneycombTracer(service_name="octoprint_plugin")
 
         self.client = mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv311)
@@ -106,12 +108,12 @@ def __init__(
         self.client.on_unsubscribe = self._on_unsubscribe
 
         # device receives configuration updates on this topic
-        self.mqtt_config_topic = f"/devices/{self.device_cloudiot_id}/config"
+        self.config_topic = f"/devices/{self.device_cloudiot_id}/config"
 
         # device receives commands on this topic
-        self.mqtt_command_topic = f"/devices/{self.device_cloudiot_id}/commands/#"
+        self.commands_topic = f"/devices/{self.device_cloudiot_id}/commands/#"
         # remote_control app commmands are routed to this subfolder
-        self.remote_control_command_topic = (
+        self.remote_control_commands_topic = (
             f"/devices/{self.device_cloudiot_id}/commands/remote_control"
         )
         # this permits routing on a per-app basis, e.g.
@@ -137,23 +139,29 @@ def __init__(
     ##
     @beeline.traced("MQTTClient._on_message")
     def _on_message(self, client, userdata, message):
-        if message.topic == self.remote_control_command_topic:
+        if message.topic == self.remote_control_commands_topic:
             parsed_message = json.loads(message.payload.decode("utf-8"))
             logger.info(
                 f"Received remote control command on topic={message.topic} payload={parsed_message}"
             )
-            self.remote_control_queue.put_nowait(
-                {"topic": self.remote_control_command_topic, "message": parsed_message}
+            self.mqtt_receive_queue.put_nowait(
+                {"topic": self.remote_control_commands_topic, "message": parsed_message}
             )
             # callback to api to indicate command was received
-        elif message.topic == self.mqtt_config_topic:
-            parsed_message = json.loads(message.payload.decode("utf-8"))
-            logger.info(
-                f"Received config update on topic={message.topic} payload={parsed_message}"
-            )
-            self.remote_control_queue.put_nowait(
-                {"topic": self.mqtt_config_topic, "message": parsed_message}
-            )
+        elif message.topic == self.config_topic:
+            try:
+                parsed_message = message.payload.decode("utf-8")
+                parsed_message = json.loads(parsed_message)
+                logger.info(
+                    f"Received config update on topic={message.topic} payload={parsed_message}"
+                )
+                self.mqtt_receive_queue.put_nowait(
+                    {"topic": self.config_topic, "message": parsed_message}
+                )
+            except json.decoder.JSONDecodeError:
+                logger.error(
+                    f"Failed to decode message on topic={message.topic} payload={message.payload} message={parsed_message}"
+                )
         else:
             logger.info(
                 f"MQTTClient._on_message called with userdata={userdata} topic={message.topic} payload={message}"
@@ -200,13 +208,13 @@ def _on_connect(self, client, userdata, flags, rc):
             should_backoff = False
             minimum_backoff_time = 1
             logger.info("Device successfully connected to MQTT broker")
-            self.client.subscribe(self.mqtt_config_topic, qos=1)
+            self.client.subscribe(self.config_topic, qos=1)
             logger.info(
-                f"Subscribing to config updates device_cloudiot_id={self.device_cloudiot_id} to topic {self.mqtt_config_topic}"
+                f"Subscribing to config updates device_cloudiot_id={self.device_cloudiot_id} to topic {self.config_topic}"
             )
-            self.client.subscribe(self.mqtt_command_topic, qos=1)
+            self.client.subscribe(self.commands_topic, qos=1)
             logger.info(
-                f"Subscribing to remote commands device_cloudiot_id={self.device_cloudiot_id} to topic {self.mqtt_command_topic}"
+                f"Subscribing to remote commands device_cloudiot_id={self.device_cloudiot_id} to topic {self.commands_topic}"
             )
         else:
             logger.error(f"Connection refused by MQTT broker with reason code rc={rc}")
@@ -241,7 +249,9 @@ def connect(self):
             username="unused",
             password=create_jwt(self.project_id, self.private_key_file, self.algorithm),
         )
-        self.client.connect(self.mqtt_bridge_hostname, self.mqtt_bridge_port)
+        self.client.connect(
+            self.mqtt_bridge_hostname, self.mqtt_bridge_port, keepalive=self.keepalive
+        )
         logger.info(f"MQTT client connected to {self.mqtt_bridge_hostname}")
 
     def publish(self, payload, topic=None, retain=False, qos=1):