Mercurial Repositories > MQTT Monitor / changeset

--- a/MqttMonitor/MqttClient.py	Mon Jul 19 20:00:17 2021 +0200
+++ b/MqttMonitor/MqttClient.py	Tue Jul 20 17:48:21 2021 +0200
@@ -52,11 +52,13 @@
     onDisconnectedV3 = pyqtSignal(int)
     onDisconnectedV5 = pyqtSignal(int, int)
     onLog = pyqtSignal(int, str)
-    onMessage = pyqtSignal(str, bytes, int, bool)
+    onMessageV3 = pyqtSignal(str, bytes, int, bool)
+    onMessageV5 = pyqtSignal(str, bytes, int, bool)
     onPublish = pyqtSignal(int)
     onSubscribeV3 = pyqtSignal(int, tuple)
     onSubscribeV5 = pyqtSignal(int, list)
-    onUnsubscribe = pyqtSignal(int)
+    onUnsubscribeV3 = pyqtSignal(int)
+    onUnsubscribeV5 = pyqtSignal(int, int, int)
     
     connectTimeout = pyqtSignal()
     
@@ -139,37 +141,43 @@
                 lambda client, userdata, mid, grantedQos, properties=None:
                     self.onSubscribeV3.emit(mid, grantedQos)
             )
+            self.__mqttClient.on_unsubscribe = (
+                lambda client, userdata, mid:
+                    self.onUnsubscribeV3.emit(mid)
+            )
+            self.__mqttClient.on_message = (
+                lambda client, userdata, message:
+                    self.onMessageV3.emit(message.topic, message.payload,
+                                          message.qos, message.retain)
+            )
         else:
-            # TODO: add properties to signal
+            # TODO: add properties to signals
             self.__mqttClient.on_connect = (
                 lambda client, userdata, flags, rc, properties=None:
                     self.onConnectV5.emit(flags, rc.value, rc.packetType)
             )
             self.__mqttClient.on_disconnect = self.__onDisconnectedV5
-            # TODO: add properties to signal
             self.__mqttClient.on_subscribe = (
                 lambda client, userdata, mid, reasonCodes, properties=None:
                     self.onSubscribeV5.emit(mid, reasonCodes)
             )
-        # TODO: MQTTv5: add support for MQTTv5 signature
+            self.__mqttClient.on_unsubscribe = (
+                lambda client, userdata, mid, properties, rc:
+                    self.onUnsubscribeV5.emit(mid, rc.value, rc.packetType)
+            )
+            self.__mqttClient.on_message = (
+                lambda client, userdata, message:
+                    self.onMessageV5.emit(message.topic, message.payload,
+                                          message.qos, message.retain)
+            )
         self.__mqttClient.on_log = (
             lambda client, userdata, level, buf:
                 self.onLog.emit(level, buf)
         )
-        self.__mqttClient.on_message = (
-            lambda client, userdata, message:
-                self.onMessage.emit(message.topic, message.payload,
-                                    message.qos, message.retain)
-        )
         self.__mqttClient.on_publish = (
             lambda client, userdata, mid:
                 self.onPublish.emit(mid)
         )
-        # TODO: MQTTv5: add support for MQTTv5 signature
-        self.__mqttClient.on_unsubscribe = (
-            lambda client, userdata, mid, properties=None, reasoncodes=None:
-                self.onUnsubscribe.emit(mid)
-        )
     
     def __onDisconnectedV5(self, client, userdata, rc, properties=None):
         """
@@ -178,7 +186,7 @@
         @param client reference to the client object
         @type paho.mqtt.Client
         @param userdata user data
-        @type list
+        @type Any
         @param rc result code or reason code
         @type int or ReasonCodes
         @param properties optional properties (defaults to None)
@@ -355,10 +363,10 @@
         """
 ##        if reinit:
 ##            self.reinitialise()
-        # TODO: MQTTv5: use 'clean_start' set to the remembered 'cleanSession'
         # TODO: MQTTv5: add support for MQTTv5 properties
         self.__mqttClient.connect_async(
-            host, port=port, keepalive=keepalive, bind_address=bindAddress)
+            host, port=port, keepalive=keepalive, bind_address=bindAddress,
+            clean_start=self.__cleanSession)
         
         self.__connectTimeoutTimer.start()
         
@@ -430,6 +438,7 @@
                     self.setTLS()
             
             # step 4: connect to server
+            self.__cleanSession = parametersDict["CleanSession"]
             self.connectToServer(host, port=port,
                                  keepalive=parametersDict["Keepalive"])
 ##                                 reinit=False)

--- a/MqttMonitor/MqttMonitorWidget.py	Mon Jul 19 20:00:17 2021 +0200
+++ b/MqttMonitor/MqttMonitorWidget.py	Tue Jul 20 17:48:21 2021 +0200
@@ -217,11 +217,13 @@
         client.onDisconnectedV3.connect(self.__brokerDisconnected)
         client.onDisconnectedV5.connect(self.__brokerDisconnected)
         client.onLog.connect(self.__clientLog)
-        client.onMessage.connect(self.__messageReceived)
+        client.onMessageV3.connect(self.__messageReceived)
+        client.onMessageV5.connect(self.__messageReceived)
         client.onPublish.connect(self.__messagePublished)
         client.onSubscribeV3.connect(self.__topicSubscribed)
         client.onSubscribeV5.connect(self.__topicSubscribedV5)
-        client.onUnsubscribe.connect(self.__topicUnsubscribed)
+        client.onUnsubscribeV3.connect(self.__topicUnsubscribed)
+        client.onUnsubscribeV5.connect(self.__topicUnsubscribedV5)
         
         client.connectTimeout.connect(self.__connectTimeout)
         
@@ -231,7 +233,6 @@
     ## Slots handling MQTT related signals
     #######################################################################
     
-    # TODO: change to accept ReasonCode for rc
     @pyqtSlot(dict, int)
     @pyqtSlot(dict, int, int)
     def __brokerConnected(self, flags, rc, packetType=None):
@@ -240,9 +241,10 @@
         
         @param flags flags set for the connection
         @type dict
-        @param rc CONNACK result code or tuple containing the result code and
-            the packet type of the MQTTv5 reason code
-        @type int or tuple of (int, int)
+        @param rc CONNACK result code or MQTTv5 reason code
+        @type int
+        @param packetType packet type as reported by the client
+        @type int
         """
         self.brokerStatusLabel.hide()
         
@@ -367,8 +369,9 @@
         else:
             self.logEdit.verticalScrollBar().setValue(scrollbarValue)
     
+    # TODO: add support for MQTT v5 properties
     @pyqtSlot(str, bytes, int, bool)
-    def __messageReceived(self, topic, payload, qos, retain):
+    def __messageReceived(self, topic, payload, qos, retain, properties=None):
         """
         Private slot to handle the receipt of a message.
         
@@ -380,12 +383,15 @@
         @type int
         @param retain flag indicating a retained message
         @type bool
+        @param properties properties sent with the message (MQTT v5)
+        @type Properties
         """
         if topic.startswith(MqttMonitorWidget.BrokerStatusTopicPrefix):
             # handle broker status messages
             self.__handleBrokerStatusMessage(topic, payload)
         else:
-            self.__appendMessage(topic, payload, qos)
+            self.__appendMessage(topic, payload, qos, retain,
+                                 properties=properties)
     
     @pyqtSlot(int)
     def __messagePublished(self, mid):
@@ -424,8 +430,8 @@
         
         @param mid ID of the subscribe request
         @type int
-        @param grantedQos tuple of granted quality of service
-        @type tuple of int
+        @param reasonCodes list of reason codes, one for each topic
+        @type list of ReasonCodes
         """
         msg = mqttReasonCode(reasonCodes[0].value, reasonCodes[0].packetType)
         self.__flashBrokerStatusLabel(msg)
@@ -434,7 +440,8 @@
     @pyqtSlot(int)
     def __topicUnsubscribed(self, mid):
         """
-        Private slot to handle being unsubcribed from a topic.
+        Private slot to handle being unsubcribed from a topic (MQTT v3.1,
+        MQTT v3.1.1).
         
         @param mid ID of the unsubscribe request
         @type int
@@ -446,6 +453,22 @@
                 self.__updateUnsubscribeTopicComboBox()
                 self.__updatePublishTopicComboBox()
     
+    @pyqtSlot(int, int, int)
+    def __topicUnsubscribedV5(self, mid, rc, packetType):
+        """
+        Private slot to handle being unsubscribed to topics (MQTT v5).
+        
+        @param mid ID of the subscribe request
+        @type int
+        @param rc MQTTv5 reason code
+        @type int
+        @param packetType packet type as reported by the client
+        @type int
+        """
+        msg = mqttReasonCode(rc, packetType)
+        self.__flashBrokerStatusLabel(msg)
+        self.__topicUnsubscribed(mid)
+    
     #######################################################################
     ## Slots handling UI interactions
     #######################################################################
@@ -913,7 +936,7 @@
             topicIndex = self.publishTopicComboBox.findText(currentTopic)
             self.publishTopicComboBox.setCurrentIndex(topicIndex)
     
-    def __appendMessage(self, topic, payload, qos):
+    def __appendMessage(self, topic, payload, qos, retain, properties=None):
         """
         Private method to append a received message to the output.
         
@@ -923,7 +946,12 @@
         @type bytes
         @param qos quality of service indicator (0, 1, 2)
         @type int
+        @param retain flag indicating a retained message
+        @type bool
+        @param properties properties sent with the message (MQTT v5)
+        @type Properties
         """
+        # TODO: add Output for retain and properties
         scrollbarValue = self.messagesEdit.verticalScrollBar().value()
         
         textCursor = self.messagesEdit.textCursor()
@@ -949,6 +977,10 @@
         self.messagesEdit.setCurrentCharFormat(self.__messagesQosFormat)
         self.messagesEdit.insertPlainText(self.tr("QoS: {0}\n").format(qos))
         
+        if retain:
+            self.messagesEdit.setCurrentCharFormat(self.__messagesQosFormat)
+            self.messagesEdit.insertPlainText(self.tr("Retained Message\n"))
+        
         payloadStr = str(payload, encoding="utf-8", errors="replace")
         self.messagesEdit.setCurrentCharFormat(self.__messagesFormat)
         self.messagesEdit.insertPlainText(