Broker compatibility: add option for brokers not 100% mqtt compliant

Some brokers (like AWS iot core) do not support qos=2 or retain
messages.
Add option to transport to specify the max qos value to use and also
if the retain flag is supported by broker.

In case of retain flag, behavior of transport is a bit different.

Author: GwendalRaoul

python_transport/wirepas_gateway/protocol/mqtt_wrapper.py

@@ -30,6 +30,7 @@ def __init__(
         logger,
         on_termination_cb=None,
         on_connect_cb=None,
+        on_disconnect_cb=None,
         last_will_topic=None,
         last_will_data=None,
     ):
@@ -39,11 +40,16 @@ def __init__(
         self.logger = logger
         self.on_termination_cb = on_termination_cb
         self.on_connect_cb = on_connect_cb
+        self.on_disconnect_cb = on_disconnect_cb
         # Set to track the unpublished packets
         self._unpublished_mid_set = set()
         # Variable to keep track of latest published packet
         self._timestamp_last_publish = datetime.now()
 
+        # load special settings for borker compatibility
+        self.max_qos = settings.mqtt_max_qos_supported
+        self.retain_not_supported = settings.mqtt_retain_flag_not_supported
+
         self._client = mqtt.Client(
             client_id=settings.gateway_id,
             clean_session=not settings.mqtt_persist_session,
@@ -65,6 +71,7 @@ def __init__(
 
         self._client.username_pw_set(settings.mqtt_username, settings.mqtt_password)
         self._client.on_connect = self._on_connect
+        self._client.on_disconnect = self._on_disconnect
         self._client.on_publish = self._on_publish
 
         if last_will_topic is not None and last_will_data is not None:
@@ -102,6 +109,13 @@ def _on_connect(self, client, userdata, flags, rc):
         if self.on_connect_cb is not None:
             self.on_connect_cb()
 
+    def _on_disconnect(self, client, userdata, rc):
+        if rc != 0:
+            self.logger.error("MQTT disconnect: %s (%s)", connack_string(rc), rc)
+
+        if self.on_disconnect_cb is not None:
+            self.on_disconnect_cb()
+
     def _on_publish(self, client, userdata, mid):
         self._unpublished_mid_set.remove(mid)
         self._timestamp_last_publish = datetime.now()
@@ -196,7 +210,7 @@ def _get_socket(self):
 
     def _set_last_will(self, topic, data):
         # Set Last wil message
-        self._client.will_set(topic, data, qos=2, retain=True)
+        self._client.will_set(topic, data, qos=1, retain=not self.retain_not_supported)
 
     def run(self):
         self.running = True
@@ -238,6 +252,14 @@ def _publish_from_wrapper_thread(self, topic, payload, qos, retain):
             retain: Is it a retain message
 
         """
+        # Limit qos in case broker has a limit
+        if qos > self.max_qos:
+            qos = self.max_qos
+
+        # Clear retain flag if not supported
+        if self.retain_not_supported:
+            retain = False
+
         mid = self._client.publish(topic, payload, qos=qos, retain=retain).mid
         if self.publish_queue_size == 0:
             # Reset last published packet
@@ -250,14 +272,28 @@ def publish(self, topic, payload, qos=1, retain=False) -> None:
         Args:
             topic: Topic to publish on
             payload: Payload
-            qos: Qos to use
-            retain: Is it a retain message
+            qos: Qos to use. Can be less than requested if broker does not support it
+            retain: Is it a retain message. Can be discarded if broker does not support it
 
         """
+        # No need to check qos or retain at this stage as done in real publish to broker
+
         # Send it to the queue to be published from Mqtt thread
         self._publish_queue.put((topic, payload, qos, retain))
 
     def subscribe(self, topic, cb, qos=2) -> None:
+        """ Method to subscribe to mqtt topic
+
+        Args:
+            topic: Topic to publish on
+            cb: Callback to call on message reception
+            qos: Qos to use. Can be less than requested if broker does not support it
+
+        """
+        # Limit qos in case broker has a limit
+        if qos > self.max_qos:
+            qos = self.max_qos
+
         self.logger.debug("Subscribing to: {}".format(topic))
         self._client.subscribe(topic, qos)
         self._client.message_callback_add(topic, cb)

python_transport/wirepas_gateway/transport_service.py

@@ -168,6 +168,48 @@ def initialize_sink(self, name):
                 sink.cost = self.minimum_sink_cost
 
 
+class PeriodicPublishStatusThread(Thread):
+    def __init__(self, logger, period, mqtt_wrapper, topic, payload):
+        """
+        Thread to periodically publish gateway status if retain flag is not
+        supported by broker.
+
+        Args:
+            logger: logger used to log messages
+            period: the period to check the buffer status
+            mqtt_wrapper: the mqtt wrapper to get access to queue level
+        """
+        Thread.__init__(self)
+
+        self.logger = logger
+
+        # Daemonize thread to exit with full process
+        self.daemon = True
+        self.running = False
+
+        # How often to publish
+        self.period = period
+        self.mqtt_wrapper = mqtt_wrapper
+        self.topic = topic
+        self.payload = payload
+
+    def run(self):
+        """
+        Main loop to publish status
+        """
+        self.running = True
+        while self.running:
+            self.logger.debug("Publish status")
+            self.mqtt_wrapper.publish(self.topic, self.payload, qos=1, retain=False)
+            sleep(self.period)
+
+    def stop(self):
+        """
+        Stop the publish script
+        """
+        self.running = False
+
+
 class TransportService(BusClient):
     """
     Implementation of gateway to backend protocol
@@ -182,6 +224,9 @@ class TransportService(BusClient):
     # Period in s to check for black hole issue
     MONITORING_BUFFERING_PERIOD_S = 1
 
+    # Period to send status in case broker do not support retain flag
+    GATEWAY_STATUS_PERIOD_S = 30
+
     def __init__(self, settings, logger=None, **kwargs):
         self.logger = logger or logging.getLogger(__name__)
         self.logger.info("Version is: %s", transport_version)
@@ -197,6 +242,10 @@ def __init__(self, settings, logger=None, **kwargs):
         self.gw_model = settings.gateway_model
         self.gw_version = settings.gateway_version
 
+        # Does broker support retain flag
+        self.retain_not_supported = settings.mqtt_retain_flag_not_supported
+        self.status_publish_thread = None
+
         self.whitened_ep_filter = settings.whitened_endpoints_filter
 
         last_will_topic = TopicGenerator.make_status_topic(self.gw_id)
@@ -209,6 +258,7 @@ def __init__(self, settings, logger=None, **kwargs):
             self.logger,
             self._on_mqtt_wrapper_termination_cb,
             self._on_connect,
+            self._on_disconnect,
             last_will_topic,
             last_will_message,
         )
@@ -254,7 +304,20 @@ def _set_status(self):
 
         topic = TopicGenerator.make_status_topic(self.gw_id)
 
-        self.mqtt_wrapper.publish(topic, event_online.payload, qos=1, retain=True)
+        if self.retain_not_supported:
+            if self.status_publish_thread is None:
+                self.status_publish_thread = PeriodicPublishStatusThread(
+                    self.logger,
+                    self.GATEWAY_STATUS_PERIOD_S,
+                    self.mqtt_wrapper,
+                    topic,
+                    event_online.payload,
+                )
+                self.status_publish_thread.start()
+            else:
+                self.logger.warning("Publish thread already running")
+        else:
+            self.mqtt_wrapper.publish(topic, event_online.payload, qos=1, retain=True)
 
     def _on_connect(self):
         # Register for get gateway info
@@ -295,6 +358,12 @@ def _on_connect(self):
 
         self.logger.info("MQTT connected!")
 
+    def _on_disconnect(self):
+        self.logger.info("MQTT disconnected!")
+        if self.status_publish_thread is not None:
+            self.status_publish_thread.stop()
+            self.status_publish_thread = None
+
     def on_data_received(
         self,
         sink_id,

python_transport/wirepas_gateway/utils/argument_tools.py

@@ -344,6 +344,28 @@ def add_mqtt(self):
             ),
         )
 
+        self.mqtt.add_argument(
+            "--mqtt_max_qos_supported",
+            default=os.environ.get("WM_SERVICES_MQTT_MAX_QOS_SUPPORTED", 2),
+            action="store",
+            type=self.str2int,
+            help=(
+                "Max qos supported by broker in case it has limitation"
+                "compare to MQTT specification that is 2."
+            ),
+        )
+
+        self.mqtt.add_argument(
+            "--mqtt_retain_flag_not_supported",
+            default=bool(
+                os.environ.get("WM_SERVICES_MQTT_RETAIN_FLAG_SUPPORTED", False)
+            ),
+            type=self.str2bool,
+            nargs="?",
+            const=True,
+            help=("Set to false if broker do not support retain flag"),
+        )
+
     def add_buffering_settings(self):
         """ Parameters used to avoid black hole case """
         self.buffering.add_argument(