diff -r 70b8858199f5 -r 5fe4f179975f MqttMonitor/MqttClient.py
--- a/MqttMonitor/MqttClient.py Wed Jul 21 20:10:36 2021 +0200
+++ b/MqttMonitor/MqttClient.py Thu Jul 22 19:02:32 2021 +0200
@@ -122,6 +122,7 @@
self.__cleanSession = cleanSession
self.__protocol = protocol
+ self.__disconnectUserProperties = []
if protocol == MqttProtocols.MQTTv5:
cleanSession = None
@@ -278,7 +279,7 @@
"""
self.__mqttClient.user_data_set(userdata)
- # TODO: MQTTv5: add support for properties
+ # TODO: MQTTv5: add support for WILL properties
def setLastWill(self, topic, payload=None, qos=0, retain=False):
"""
Public method to set the last will of the client.
@@ -350,7 +351,7 @@
self.__loopStarted = False
def connectToServer(self, host, port=1883, keepalive=60, bindAddress="",
- reinit=True):
+ properties=None):
"""
Public method to connect to a remote MQTT broker.
@@ -365,14 +366,18 @@
@param bindAddress IP address of a local network interface to bind
this client to
@type str
- @param reinit flag indicating to reinitialize the MQTT client before
- trying to connect with the given parameters
- @type bool
+ @param properties list of user properties to be sent with the
+ subscription
+ @type list of tuple of (str, str)
"""
- # TODO: MQTTv5: add support for MQTTv5 properties
+ props = (
+ self.__createPropertiesObject(PacketTypes.CONNECT, properties)
+ if properties else
+ None
+ )
self.__mqttClient.connect_async(
host, port=port, keepalive=keepalive, bind_address=bindAddress,
- clean_start=self.__cleanSession)
+ clean_start=self.__cleanSession, properties=props)
self.__connectTimeoutTimer.start()
@@ -393,10 +398,10 @@
this client to
@type str
@param options dictionary containing the connection options. This
- dictionary should contain the keys "ClientId", "Keepalive",
- "CleanSession", "Username", "Password", "WillTopic", "WillMessage",
- "WillQos", "WillRetain", "TlsEnable", "TlsCaCert", "TlsClientCert",
- "TlsClientKey", "ConnectionTimeout"
+ dictionary should contain the keys "ClientId", "ConnectionTimeout",
+ "Keepalive", "CleanSession", "Username", "Password", "WillTopic",
+ "WillMessage", "WillQos", "WillRetain", "TlsEnable", "TlsCaCert",
+ "TlsClientCert", "TlsClientKey", "UserProperties".
@type dict
"""
if options:
@@ -443,10 +448,25 @@
# use default TLS configuration
self.setTLS()
+ # step 4: get the connect user properties
+ if self.__protocol == MqttProtocols.MQTTv5:
+ try:
+ userProperties = parametersDict["UserProperties"]
+ properties = userProperties["connect"][:]
+ self.__disconnectUserProperties = (
+ userProperties["connect"][:]
+ if userProperties["use_connect"] else
+ userProperties["disconnect"][:]
+ )
+ except KeyError:
+ properties = None
+ else:
+ properties = None
# step 4: connect to server
self.__cleanSession = parametersDict["CleanSession"]
self.connectToServer(host, port=port,
- keepalive=parametersDict["Keepalive"])
+ keepalive=parametersDict["Keepalive"],
+ properties=properties)
else:
keepalive = self.defaultConnectionOptions["Keepalive"]
self.connectToServer(host, port=port, keepalive=keepalive,
@@ -462,7 +482,7 @@
the keys "ClientId", "Protocol", "ConnectionTimeout", "Keepalive",
"CleanSession", "Username", "Password", "WillTopic", "WillMessage",
"WillQos", "WillRetain", "TlsEnable", "TlsCaCert", "TlsClientCert",
- "TlsClientKey".
+ "TlsClientKey", "UserProperties".
@rtype dict
"""
return {
@@ -481,6 +501,11 @@
"TlsCaCert": "",
"TlsClientCert": "",
"TlsClientKey": "",
+ "UserProperties": {
+ "connect": [],
+ "disconnect": [],
+ "use_connect": True,
+ },
}
def reconnectToServer(self):
@@ -500,9 +525,13 @@
"""
self.__connectTimeoutTimer.stop()
- # TODO: MQTTv5: add support for properties (?)
- # TODO: MQTTv5: add support for reason code
- self.__mqttClient.disconnect()
+ props = (
+ self.__createPropertiesObject(
+ PacketTypes.DISCONNECT, self.__disconnectUserProperties)
+ if self.__disconnectUserProperties else
+ None
+ )
+ self.__mqttClient.disconnect(properties=props)
def subscribe(self, topic, qos=0, properties=None):
"""
@@ -539,14 +568,14 @@
@rtype tuple of (int, int)
"""
props = (
- self.__createPropertiesObject(PacketTypes.SUBSCRIBE, properties)
+ self.__createPropertiesObject(PacketTypes.UNSUBSCRIBE, properties)
if properties else
None
)
return self.__mqttClient.unsubscribe(topic, properties=props)
- # TODO: MQTTv5: add support for properties
- def publish(self, topic, payload=None, qos=0, retain=False):
+ def publish(self, topic, payload=None, qos=0, retain=False,
+ properties=None):
"""
Public method to publish to a topic.
@@ -559,11 +588,19 @@
@param retain flag indicating to set as the "last known good"/retained
message for the topic
@type bool
+ @param properties list of user properties to be sent with the
+ subscription
+ @type list of tuple of (str, str)
@return message info object
@rtype mqtt.MQTTMessageInfo
"""
+ props = (
+ self.__createPropertiesObject(PacketTypes.PUBLISH, properties)
+ if properties else
+ None
+ )
return self.__mqttClient.publish(topic, payload=payload, qos=qos,
- retain=retain)
+ retain=retain, properties=props)
def __createPropertiesObject(self, packetType, properties):
"""
@@ -577,8 +614,7 @@
@rtype Properties
"""
props = Properties(packetType)
- for userProperty in properties:
- props.UserProperty = tuple(userProperty)
+ props.UserProperty = properties
return props
