From a892672057e4fe5caee7bedf652f7907c0cb87fc Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Thu, 31 May 2018 17:10:34 +0100 Subject: [PATCH] Add full test1 tests for V5 --- src/MQTTClient.c | 47 +- src/MQTTClient.h | 6 +- test/CMakeLists.txt | 36 ++ test/MQTTV311.py | 123 ++-- test/MQTTV5.py | 1464 +++++++++++++++++++++++++++++++++++++++++++ test/mqttsas.py | 44 +- test/test1.c | 2 +- test/test15.c | 191 +++--- test/test5.c | 3 +- test/test6.c | 15 +- 10 files changed, 1751 insertions(+), 180 deletions(-) create mode 100644 test/MQTTV5.py diff --git a/src/MQTTClient.c b/src/MQTTClient.c index 374569ad..fff84440 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -801,7 +801,7 @@ static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason client->connected = 0; client->connect_state = 0; - if (client->cleansession) + if (client->MQTTVersion < MQTTVERSION_5 && client->cleansession) MQTTClient_cleanSession(client); FUNC_EXIT; } @@ -1551,7 +1551,13 @@ exit: int MQTTClient_subscribeMany(MQTTClient handle, int count, char* const* topic, int* qos) { - MQTTResponse response = MQTTClient_subscribeMany5(handle, count, topic, qos, NULL, NULL); + MQTTClients* m = handle; + MQTTResponse response = {MQTTCLIENT_SUCCESS, NULL}; + + if (m->c->MQTTVersion >= MQTTVERSION_5) + response.reasonCode = MQTTCLIENT_BAD_MQTT_VERSION; + else + response = MQTTClient_subscribeMany5(handle, count, topic, qos, NULL, NULL); return response.reasonCode; } @@ -1575,7 +1581,13 @@ MQTTResponse MQTTClient_subscribe5(MQTTClient handle, const char* topic, int qos int MQTTClient_subscribe(MQTTClient handle, const char* topic, int qos) { - MQTTResponse response = MQTTClient_subscribe5(handle, topic, qos, NULL, NULL); + MQTTClients* m = handle; + MQTTResponse response = {MQTTCLIENT_SUCCESS, NULL}; + + if (m->c->MQTTVersion >= MQTTVERSION_5) + response.reasonCode = MQTTCLIENT_BAD_MQTT_VERSION; + else + response = MQTTClient_subscribe5(handle, topic, qos, NULL, NULL); return response.reasonCode; } @@ -1698,8 +1710,6 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p rc = MQTTCLIENT_DISCONNECTED; else if (!UTF8_validateString(topicName)) rc = MQTTCLIENT_BAD_UTF8_STRING; - else if (m->c->MQTTVersion >= MQTTVERSION_5 && properties == NULL) - rc = MQTTCLIENT_NULL_PARAMETER; if (rc != MQTTCLIENT_SUCCESS) goto exit; @@ -1738,7 +1748,15 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p p->msgId = msgid; p->MQTTVersion = m->c->MQTTVersion; if (m->c->MQTTVersion >= MQTTVERSION_5) - p->properties = *properties; + { + if (properties) + p->properties = *properties; + else + { + MQTTProperties props = MQTTProperties_initializer; + p->properties = props; + } + } rc = MQTTProtocol_startPublish(m->c, p, qos, retained, &msg); @@ -1781,7 +1799,13 @@ exit: int MQTTClient_publish(MQTTClient handle, const char* topicName, int payloadlen, void* payload, int qos, int retained, MQTTClient_deliveryToken* deliveryToken) { - MQTTResponse rc = MQTTClient_publish5(handle, topicName, payloadlen, payload, qos, retained, NULL, deliveryToken); + MQTTClients* m = handle; + MQTTResponse rc = {MQTTCLIENT_SUCCESS, NULL}; + + if (m->c->MQTTVersion >= MQTTVERSION_5) + rc.reasonCode = MQTTCLIENT_BAD_MQTT_VERSION; + else + rc = MQTTClient_publish5(handle, topicName, payloadlen, payload, qos, retained, NULL, deliveryToken); return rc.reasonCode; } @@ -1820,13 +1844,16 @@ exit: int MQTTClient_publishMessage(MQTTClient handle, const char* topicName, MQTTClient_message* message, MQTTClient_deliveryToken* deliveryToken) { + MQTTClients* m = handle; MQTTResponse rc = {MQTTCLIENT_SUCCESS, NULL}; if (strncmp(message->struct_id, "MQTM", 4) != 0 || (message->struct_version != 0 && message->struct_version != 1)) - return MQTTCLIENT_BAD_STRUCTURE; - - rc = MQTTClient_publishMessage5(handle, topicName, message, deliveryToken); + rc.reasonCode = MQTTCLIENT_BAD_STRUCTURE; + else if (m->c->MQTTVersion >= MQTTVERSION_5) + rc.reasonCode = MQTTCLIENT_BAD_MQTT_VERSION; + else + rc = MQTTClient_publishMessage5(handle, topicName, message, deliveryToken); return rc.reasonCode; } diff --git a/src/MQTTClient.h b/src/MQTTClient.h index 720c9cfb..b4f51a5e 100644 --- a/src/MQTTClient.h +++ b/src/MQTTClient.h @@ -712,6 +712,7 @@ typedef struct * MQTTVERSION_DEFAULT (0) = default: start with 3.1.1, and if that fails, fall back to 3.1 * MQTTVERSION_3_1 (3) = only try version 3.1 * MQTTVERSION_3_1_1 (4) = only try version 3.1.1 + * MQTTVERSION_5 (5) = only try version 5.0 */ int MQTTVersion; /** @@ -726,8 +727,9 @@ typedef struct /** * Optional binary password. Only checked and used if the password option is NULL */ - struct { - int len; /**< binary password length */ + struct + { + int len; /**< binary password length */ const void* data; /**< binary password data */ } binarypwd; } MQTTClient_connectOptions; diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 52170d8f..c88c3cb4 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -110,8 +110,44 @@ ADD_TEST( COMMAND "test15" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} ) +ADD_TEST( + NAME test15-2-multithread-callbacks + COMMAND "test15" "--test_no" "2" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test15-3-connack-return-codes + COMMAND "test15" "--test_no" "3" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test15-4-client-persistence + COMMAND "test15" "--test_no" "4" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test15-5-disconnect-with-quiesce + COMMAND "test15" "--test_no" "5" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test15-6-connlost-will-message + COMMAND "test15" "--test_no" "6" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test15-7-connlost-binary-will-message + COMMAND "test15" "--test_no" "7" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + SET_TESTS_PROPERTIES( test15-1-single-thread-client + test15-2-multithread-callbacks + test15-3-connack-return-codes + test15-4-client-persistence + test15-5-disconnect-with-quiesce + test15-6-connlost-will-message + test15-7-connlost-binary-will-message PROPERTIES TIMEOUT 540 ) diff --git a/test/MQTTV311.py b/test/MQTTV311.py index 0faa9f70..36cea823 100644 --- a/test/MQTTV311.py +++ b/test/MQTTV311.py @@ -1,16 +1,16 @@ """ ******************************************************************* - Copyright (c) 2013, 2014 IBM Corp. - + Copyright (c) 2013, 2018 IBM Corp. + All rights reserved. This program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 - and Eclipse Distribution License v1.0 which accompany this distribution. - - The Eclipse Public License is available at + and Eclipse Distribution License v1.0 which accompany this distribution. + + The Eclipse Public License is available at http://www.eclipse.org/legal/epl-v10.html - and the Eclipse Distribution License is available at + and the Eclipse Distribution License is available at http://www.eclipse.org/org/documents/edl-v10.php. - + Contributors: Ian Craggs - initial implementation and/or documentation ******************************************************************* @@ -31,7 +31,7 @@ logger = logging.getLogger("mqttsas") class MQTTException(Exception): pass - + # Message types CONNECT, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL, \ @@ -106,10 +106,10 @@ class FixedHeaders: self.RETAIN == fh.RETAIN # and \ # self.remainingLength == fh.remainingLength - def __repr__(self): - "return printable representation of our data" - return classNames[self.MessageType]+'(DUP='+repr(self.DUP)+ \ - ", QoS="+repr(self.QoS)+", Retain="+repr(self.RETAIN) + def __str__(self): + "return printable stresentation of our data" + return classNames[self.MessageType]+'(DUP='+str(self.DUP)+ \ + ", QoS="+str(self.QoS)+", Retain="+str(self.RETAIN) def pack(self, length): "pack data into string buffer ready for transmission down socket" @@ -185,7 +185,7 @@ def readUTF(buffer, maxlen): if zz != -1: raise MQTTException("[MQTT-1.5.3-1] D800-DFFF found in UTF data "+buf) if buf.find("\uFEFF") != -1: - logger.info("[MQTT-1.5.3-3] U+FEFF in UTF string") + logger.info("[MQTT-1.5.3-3] U+FEFF in UTF string") return buf def writeBytes(buffer): @@ -202,8 +202,8 @@ class Packets: buffer = self.fh.pack(0) return buffer - def __repr__(self): - return repr(self.fh) + def __str__(self): + return str(self.fh) def __eq__(self, packet): return self.fh == packet.fh if packet else False @@ -235,20 +235,20 @@ class Connects(Packets): if buffer != None: self.unpack(buffer) - def pack(self): + def pack(self): connectFlags = bytes([(self.CleanSession << 1) | (self.WillFlag << 2) | \ (self.WillQoS << 3) | (self.WillRETAIN << 5) | \ (self.usernameFlag << 6) | (self.passwordFlag << 7)]) buffer = writeUTF(self.ProtocolName) + bytes([self.ProtocolVersion]) + \ connectFlags + writeInt16(self.KeepAliveTimer) - buffer += writeUTF(self.ClientIdentifier) + buffer += writeUTF(self.ClientIdentifier) if self.WillFlag: - buffer += writeUTF(self.WillTopic) - buffer += writeBytes(self.WillMessage) + buffer += writeUTF(self.WillTopic) + buffer += writeBytes(self.WillMessage) if self.usernameFlag: - buffer += writeUTF(self.username) + buffer += writeUTF(self.username) if self.passwordFlag: - buffer += writeBytes(self.password) + buffer += writeBytes(self.password) buffer = self.fh.pack(len(buffer)) + buffer return buffer @@ -331,15 +331,15 @@ class Connects(Packets): - def __repr__(self): - buf = repr(self.fh)+", ProtocolName="+str(self.ProtocolName)+", ProtocolVersion=" +\ - repr(self.ProtocolVersion)+", CleanSession="+repr(self.CleanSession) +\ - ", WillFlag="+repr(self.WillFlag)+", KeepAliveTimer=" +\ - repr(self.KeepAliveTimer)+", ClientId="+str(self.ClientIdentifier) +\ - ", usernameFlag="+repr(self.usernameFlag)+", passwordFlag="+repr(self.passwordFlag) + def __str__(self): + buf = str(self.fh)+", ProtocolName="+str(self.ProtocolName)+", ProtocolVersion=" +\ + str(self.ProtocolVersion)+", CleanSession="+str(self.CleanSession) +\ + ", WillFlag="+str(self.WillFlag)+", KeepAliveTimer=" +\ + str(self.KeepAliveTimer)+", ClientId="+str(self.ClientIdentifier) +\ + ", usernameFlag="+str(self.usernameFlag)+", passwordFlag="+str(self.passwordFlag) if self.WillFlag: - buf += ", WillQoS=" + repr(self.WillQoS) +\ - ", WillRETAIN=" + repr(self.WillRETAIN) +\ + buf += ", WillQoS=" + str(self.WillQoS) +\ + ", WillRETAIN=" + str(self.WillRETAIN) +\ ", WillTopic='"+ self.WillTopic +\ "', WillMessage='"+str(self.WillMessage)+"'" if self.username: @@ -393,8 +393,8 @@ class Connacks(Packets): assert self.fh.QoS == 0, "[MQTT-2.1.2-1]" assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" - def __repr__(self): - return repr(self.fh)+", Session present="+str((self.flags & 0x01) == 1)+", ReturnCode="+repr(self.returnCode)+")" + def __str__(self): + return str(self.fh)+", Session present="+str((self.flags & 0x01) == 1)+", ReturnCode="+str(self.returnCode)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -421,8 +421,8 @@ class Disconnects(Packets): assert self.fh.QoS == 0, "[MQTT-2.1.2-1]" assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" - def __repr__(self): - return repr(self.fh)+")" + def __str__(self): + return str(self.fh)+")" class Publishes(Packets): @@ -475,11 +475,11 @@ class Publishes(Packets): assert self.fh.DUP == False, "[MQTT-2.1.2-4]" return fhlen + self.fh.remainingLength - def __repr__(self): - rc = repr(self.fh) + def __str__(self): + rc = str(self.fh) if self.fh.QoS != 0: - rc += ", MsgId="+repr(self.messageIdentifier) - rc += ", TopicName="+repr(self.topicName)+", Payload="+repr(self.data)+")" + rc += ", MsgId="+str(self.messageIdentifier) + rc += ", TopicName="+str(self.topicName)+", Payload="+str(self.data)+")" return rc def __eq__(self, packet): @@ -520,8 +520,8 @@ class Pubacks(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] Puback reserved bits must be 0" return fhlen + 2 - def __repr__(self): - return repr(self.fh)+", MsgId "+repr(self.messageIdentifier) + def __str__(self): + return str(self.fh)+", MsgId "+str(self.messageIdentifier) def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -557,8 +557,8 @@ class Pubrecs(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] Pubrec reserved bits must be 0" return fhlen + 2 - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -595,8 +595,8 @@ class Pubrels(Packets): logger.info("[MQTT-3.6.1-1] bits in fixed header for pubrel are ok") return fhlen + 2 - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -632,8 +632,8 @@ class Pubcomps(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] Retain should be false in Pubcomp" return fhlen + 2 - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -685,9 +685,9 @@ class Subscribes(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] RETAIN must be false in subscribe" return fhlen + self.fh.remainingLength - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+\ - ", Data="+repr(self.data)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+\ + ", Data="+str(self.data)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -735,9 +735,9 @@ class Subacks(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] Retain should be false in suback" return fhlen + self.fh.remainingLength - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+\ - ", Data="+repr(self.data)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+\ + ", Data="+str(self.data)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -787,9 +787,9 @@ class Unsubscribes(Packets): logger.info("[MQTT-3-10.1-1] fixed header bits are 0,0,1,0") return fhlen + self.fh.remainingLength - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+\ - ", Data="+repr(self.data)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+\ + ", Data="+str(self.data)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -827,8 +827,8 @@ class Unsubacks(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" return fhlen + self.fh.remainingLength - def __repr__(self): - return repr(self.fh)+", MsgId="+repr(self.messageIdentifier)+")" + def __str__(self): + return str(self.fh)+", MsgId="+str(self.messageIdentifier)+")" def __eq__(self, packet): return Packets.__eq__(self, packet) and \ @@ -855,8 +855,8 @@ class Pingreqs(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" return fhlen - def __repr__(self): - return repr(self.fh)+")" + def __str__(self): + return str(self.fh)+")" class Pingresps(Packets): @@ -879,8 +879,8 @@ class Pingresps(Packets): assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" return fhlen - def __repr__(self): - return repr(self.fh)+")" + def __str__(self): + return str(self.fh)+")" classes = [None, Connects, Connacks, Publishes, Pubacks, Pubrecs, Pubrels, Pubcomps, Subscribes, Subacks, Unsubscribes, @@ -910,11 +910,10 @@ if __name__ == "__main__": pass for packet in classes[1:]: - before = str(packet()) + before = str(packet()) after = str(unpackPacket(packet().pack())) try: assert before == after except: print("before:", before, "\nafter:", after) print("End") - diff --git a/test/MQTTV5.py b/test/MQTTV5.py new file mode 100644 index 00000000..567aa77f --- /dev/null +++ b/test/MQTTV5.py @@ -0,0 +1,1464 @@ +""" +******************************************************************* + Copyright (c) 2013, 2018 IBM Corp. + + All rights reserved. This program and the accompanying materials + are made available under the terms of the Eclipse Public License v1.0 + and Eclipse Distribution License v1.0 which accompany this distribution. + + The Eclipse Public License is available at + http://www.eclipse.org/legal/epl-v10.html + and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + + Contributors: + Ian Craggs - initial implementation and/or documentation + Ian Craggs - take MQTT 3.1.1 and create MQTT 5.0 version +******************************************************************* +""" + +""" + +Assertions are used to validate incoming data, but are omitted from outgoing packets. This is +so that the tests that use this package can send invalid data for error testing. + +""" + +import logging, struct + +logger = logging.getLogger('MQTTV5') + +# Low-level protocol interface + +class MQTTException(Exception): + pass + +class MalformedPacket(MQTTException): + pass + +class ProtocolError(MQTTException): + pass + +MAX_PACKET_SIZE = 2**28-1 +MAX_PACKETID = 2**16-1 + +class PacketTypes: + + indexes = range(1, 16) + + # Packet types + CONNECT, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL, \ + PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK, \ + PINGREQ, PINGRESP, DISCONNECT, AUTH = indexes + + # Dummy packet type for properties use - will delay only applies to will + WILLMESSAGE = 99 + + +class Packets(object): + + Names = [ "reserved", \ + "Connect", "Connack", "Publish", "Puback", "Pubrec", "Pubrel", \ + "Pubcomp", "Subscribe", "Suback", "Unsubscribe", "Unsuback", \ + "Pingreq", "Pingresp", "Disconnect", "Auth"] + + classNames = [name+'es' if name == "Publish" else + name+'s' if name != "reserved" else name for name in Names] + + def pack(self): + buffer = self.fh.pack(0) + return buffer + + def __str__(self): + return str(self.fh) + + def __eq__(self, packet): + return self.fh == packet.fh if packet else False + + def __setattr__(self, name, value): + if name not in self.names: + raise MQTTException(name + " Attribute name must be one of "+str(self.names)) + object.__setattr__(self, name, value) + + +def PacketType(byte): + """ + Retrieve the message type from the first byte of the fixed header. + """ + if byte != None: + rc = byte[0] >> 4 + else: + rc = None + return rc + +class ReasonCodes: + """ + The reason code used in MQTT V5.0 + + """ + + def __getName__(self, packetType, identifier): + """ + used when displaying the reason code + """ + assert identifier in self.names.keys(), identifier + names = self.names[identifier] + namelist = [name for name in names.keys() if packetType in names[name]] + assert len(namelist) == 1 + return namelist[0] + + def getId(self, name): + """ + used when setting the reason code for a packetType + check that only valid codes for the packet are set + """ + identifier = None + for code in self.names.keys(): + if name in self.names[code].keys(): + if self.packetType in self.names[code][name]: + identifier = code + break + assert identifier != None, name + return identifier + + def set(self, name): + self.value = self.getId(name) + + def unpack(self, buffer): + name = self.__getName__(self.packetType, buffer[0]) + self.value = self.getId(name) + return 1 + + def getName(self): + return self.__getName__(self.packetType, self.value) + + def __str__(self): + return self.getName() + + def pack(self): + return bytes([self.value]) + + def __init__(self, packetType, aName="Success", identifier=-1): + self.packetType = packetType + self.names = { + 0 : { "Success" : [PacketTypes.CONNACK, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP, + PacketTypes.UNSUBACK, PacketTypes.AUTH], + "Normal disconnection" : [PacketTypes.DISCONNECT], + "Granted QoS 0" : [PacketTypes.SUBACK] }, + 1 : { "Granted QoS 1" : [PacketTypes.SUBACK] }, + 2 : { "Granted QoS 2" : [PacketTypes.SUBACK] }, + 4 : { "Disconnect with will message" : [PacketTypes.DISCONNECT] }, + 16 : { "No matching subscribers" : + [PacketTypes.PUBACK, PacketTypes.PUBREC] }, + 17 : { "No subscription found" : [PacketTypes.UNSUBACK] }, + 24 : { "Continue authentication" : [PacketTypes.AUTH] }, + 25 : { "Re-authenticate" : [PacketTypes.AUTH] }, + 128 : { "Unspecified error" : [PacketTypes.CONNACK, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK, + PacketTypes.DISCONNECT], }, + 129 : { "Malformed packet" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 130 : { "Protocol error" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 131 : { "Implementation specific error": [PacketTypes.CONNACK, + PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.SUBACK, + PacketTypes.UNSUBACK, PacketTypes.DISCONNECT], }, + 132 : { "Unsupported protocol version" : [PacketTypes.CONNACK] }, + 133 : { "Client identifier not valid" : [PacketTypes.CONNACK] }, + 134 : { "Bad user name or password" : [PacketTypes.CONNACK] }, + 135 : { "Not authorized" : [PacketTypes.CONNACK, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK, + PacketTypes.DISCONNECT], }, + 136 : { "Server unavailable" : [PacketTypes.CONNACK] }, + 137 : { "Server busy" : [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 138 : { "Banned" : [PacketTypes.CONNACK] }, + 139 : { "Server shutting down" : [PacketTypes.DISCONNECT] }, + 140 : { "Bad authentication method" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 141 : { "Keep alive timeout" : [PacketTypes.DISCONNECT] }, + 142 : { "Session taken over" : [PacketTypes.DISCONNECT] }, + 143 : { "Topic filter invalid" : + [PacketTypes.SUBACK, PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]}, + 144 : { "Topic name invalid" : + [PacketTypes.CONNACK, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.DISCONNECT]}, + 145 : { "Packet identifier in use" : + [PacketTypes.PUBACK, PacketTypes.PUBREC, + PacketTypes.SUBACK, PacketTypes.UNSUBACK]}, + 146 : { "Packet identifier not found" : + [PacketTypes.PUBREL, PacketTypes.PUBCOMP] }, + 147 : { "Receive maximum exceeded": [PacketTypes.DISCONNECT] }, + 148 : { "Topic alias invalid": [PacketTypes.DISCONNECT] }, + 149 : { "Packet too large": [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 150 : { "Message rate too high": [PacketTypes.DISCONNECT] }, + 151 : { "Quota exceeded": [PacketTypes.CONNACK, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.DISCONNECT], }, + 152 : { "Administrative action" : [PacketTypes.DISCONNECT] }, + 153 : { "Payload format invalid" : + [PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.DISCONNECT]}, + 154 : { "Retain not supported" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 155 : { "QoS not supported" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 156 : { "Use another server" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 157 : { "Server moved" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 158 : { "Shared subscription not supported" : + [PacketTypes.SUBACK, PacketTypes.DISCONNECT] }, + 159 : { "Connection rate exceeded" : + [PacketTypes.CONNACK, PacketTypes.DISCONNECT] }, + 160 : { "Maximum connect time" : + [PacketTypes.DISCONNECT] }, + 161 : { "Subscription identifiers not supported" : + [PacketTypes.SUBACK, PacketTypes.DISCONNECT] }, + 162 : { "Wildcard subscription not supported" : + [PacketTypes.SUBACK, PacketTypes.DISCONNECT] }, + } + if identifier == -1: + self.set(aName) + else: + self.value = identifier + self.getName() # check it's good + + +class VBIs: # Variable Byte Integer + + @staticmethod + def encode(x): + """ + Convert an integer 0 <= x <= 268435455 into multi-byte format. + Returns the buffer convered from the integer. + """ + assert 0 <= x <= 268435455 + buffer = b'' + while 1: + digit = x % 128 + x //= 128 + if x > 0: + digit |= 0x80 + buffer += bytes([digit]) + if x == 0: + break + return buffer + + @staticmethod + def decode(buffer): + """ + Get the value of a multi-byte integer from a buffer + Return the value, and the number of bytes used. + + [MQTT-1.5.5-1] the encoded value MUST use the minimum number of bytes necessary to represent the value + """ + multiplier = 1 + value = 0 + bytes = 0 + while 1: + bytes += 1 + digit = buffer[0] + buffer = buffer[1:] + value += (digit & 127) * multiplier + if digit & 128 == 0: + break + multiplier *= 128 + return (value, bytes) + +def getPacket(aSocket): + "receive the next packet" + buf = aSocket.recv(1) # get the first byte fixed header + if buf == b"": + return None + if str(aSocket).find("[closed]") != -1: + closed = True + else: + closed = False + if closed: + return None + # now get the remaining length + multiplier = 1 + remlength = 0 + while 1: + next = aSocket.recv(1) + while len(next) == 0: + next = aSocket.recv(1) + buf += next + digit = buf[-1] + remlength += (digit & 127) * multiplier + if digit & 128 == 0: + break + multiplier *= 128 + # receive the remaining length if there is any + rest = bytes([]) + if remlength > 0: + while len(rest) < remlength: + rest += aSocket.recv(remlength-len(rest)) + assert len(rest) == remlength + return buf + rest + + +class FixedHeaders(object): + + def __init__(self, aPacketType): + self.PacketType = aPacketType + self.DUP = False + self.QoS = 0 + self.RETAIN = False + self.remainingLength = 0 + + def __eq__(self, fh): + return self.PacketType == fh.PacketType and \ + self.DUP == fh.DUP and \ + self.QoS == fh.QoS and \ + self.RETAIN == fh.RETAIN # and \ + # self.remainingLength == fh.remainingLength + + def __setattr__(self, name, value): + names = ["PacketType", "DUP", "QoS", "RETAIN", "remainingLength"] + if name not in names: + raise MQTTException(name + " Attribute name must be one of "+str(names)) + object.__setattr__(self, name, value) + + def __str__(self): + "return printable representation of our data" + return Packets.classNames[self.PacketType]+'(fh.DUP='+str(self.DUP)+ \ + ", fh.QoS="+str(self.QoS)+", fh.RETAIN="+str(self.RETAIN) + + def pack(self, length): + "pack data into string buffer ready for transmission down socket" + buffer = bytes([(self.PacketType << 4) | (self.DUP << 3) |\ + (self.QoS << 1) | self.RETAIN]) + self.remainingLength = length + buffer += VBIs.encode(length) + return buffer + + def unpack(self, buffer, maximumPacketSize): + "unpack data from string buffer into separate fields" + b0 = buffer[0] + self.PacketType = b0 >> 4 + self.DUP = ((b0 >> 3) & 0x01) == 1 + self.QoS = (b0 >> 1) & 0x03 + self.RETAIN = (b0 & 0x01) == 1 + (self.remainingLength, bytes) = VBIs.decode(buffer[1:]) + if self.remainingLength + bytes + 1 > maximumPacketSize: + raise ProtocolError("Packet too large") + return bytes + 1 # length of fixed header + + +def writeInt16(length): + return bytes([length // 256, length % 256]) + +def readInt16(buf): + return buf[0]*256 + buf[1] + +def writeInt32(length): + buffer = [length // 16777216] + length %= 16777216 + buffer += [length // 65536] + length %= 65536 + buffer += [length // 256, length % 256] + return bytes(buffer) + +def readInt32(buf): + return buf[0]*16777216 + buf[1]*65536 + buf[2]*256 + buf[3] + +def writeUTF(data): + # data could be a string, or bytes. If string, encode into bytes with utf-8 + return writeInt16(len(data)) + (data if type(data) == type(b"") else bytes(data, "utf-8")) + +def readUTF(buffer, maxlen): + if maxlen >= 2: + length = readInt16(buffer) + else: + raise MalformedPacket("Not enough data to read string length") + maxlen -= 2 + if length > maxlen: + raise MalformedPacket("Length delimited string too long") + buf = buffer[2:2+length].decode("utf-8") + logger.info("[MQTT-4.7.3-2] topic names and filters must not include null") + zz = buf.find("\x00") # look for null in the UTF string + if zz != -1: + raise MalformedPacket("[MQTT-1.5.4-2] Null found in UTF data "+buf) + for c in range (0xD800, 0xDFFF): + zz = buf.find(chr(c)) # look for D800-DFFF in the UTF string + if zz != -1: + raise MalformedPacket("[MQTT-1.5.4-1] D800-DFFF found in UTF data "+buf) + if buf.find("\uFEFF") != -1: + logger.info("[MQTT-1.5.4-3] U+FEFF in UTF string") + return buf, length+2 + +def writeBytes(buffer): + return writeInt16(len(buffer)) + buffer + +def readBytes(buffer): + length = readInt16(buffer) + return buffer[2:2+length], length+2 + + +class Properties(object): + + def __init__(self, packetType): + self.packetType = packetType + self.types = ["Byte", "Two Byte Integer", "Four Byte Integer", "Variable Byte Integer", + "Binary Data", "UTF-8 Encoded String", "UTF-8 String Pair"] + + self.names = { + "Payload Format Indicator" : 1, + "Message Expiry Interval" : 2, + "Content Type" : 3, + "Response Topic" : 8, + "Correlation Data" : 9, + "Subscription Identifier" : 11, + "Session Expiry Interval" : 17, + "Assigned Client Identifier" : 18, + "Server Keep Alive" : 19, + "Authentication Method" : 21, + "Authentication Data" : 22, + "Request Problem Information" : 23, + "Will Delay Interval" : 24, + "Request Response Information" : 25, + "Response Information" : 26, + "Server Reference" : 28, + "Reason String" : 31, + "Receive Maximum" : 33, + "Topic Alias Maximum" : 34, + "Topic Alias" : 35, + "Maximum QoS" : 36, + "Retain Available" : 37, + "User Property List" : 38, + "Maximum Packet Size" : 39, + "Wildcard Subscription Available" : 40, + "Subscription Identifier Available" : 41, + "Shared Subscription Available" : 42 + } + + self.properties = { + # id: type, packets + 1 : (self.types.index("Byte"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]), # payload format indicator + 2 : (self.types.index("Four Byte Integer"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]), + 3 : (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]), + 8 : (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]), + 9 : (self.types.index("Binary Data"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]), + 11 : (self.types.index("Variable Byte Integer"), + [PacketTypes.PUBLISH, PacketTypes.SUBSCRIBE]), + 17 : (self.types.index("Four Byte Integer"), + [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.DISCONNECT]), + 18 : (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]), + 19 : (self.types.index("Two Byte Integer"), [PacketTypes.CONNACK]), + 21 : (self.types.index("UTF-8 Encoded String"), + [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]), + 22 : (self.types.index("Binary Data"), + [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]), + 23 : (self.types.index("Byte"), + [PacketTypes.CONNECT]), + 24 : (self.types.index("Four Byte Integer"), [PacketTypes.WILLMESSAGE]), + 25 : (self.types.index("Byte"), [PacketTypes.CONNECT]), + 26 : (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]), + 28 : (self.types.index("UTF-8 Encoded String"), + [PacketTypes.CONNACK, PacketTypes.DISCONNECT]), + 31 : (self.types.index("UTF-8 Encoded String"), + [PacketTypes.CONNACK, PacketTypes.PUBACK, PacketTypes.PUBREC, + PacketTypes.PUBREL, PacketTypes.PUBCOMP, PacketTypes.SUBACK, + PacketTypes.UNSUBACK, PacketTypes.DISCONNECT, PacketTypes.AUTH]), + 33 : (self.types.index("Two Byte Integer"), + [PacketTypes.CONNECT, PacketTypes.CONNACK]), + 34 : (self.types.index("Two Byte Integer"), + [PacketTypes.CONNECT, PacketTypes.CONNACK]), + 35 : (self.types.index("Two Byte Integer"), [PacketTypes.PUBLISH]), + 36 : (self.types.index("Byte"), [PacketTypes.CONNACK]), + 37 : (self.types.index("Byte"), [PacketTypes.CONNACK]), + 38 : (self.types.index("UTF-8 String Pair"), + [PacketTypes.CONNECT, PacketTypes.CONNACK, + PacketTypes.PUBLISH, PacketTypes.PUBACK, + PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP, + PacketTypes.SUBSCRIBE, PacketTypes.SUBACK, + PacketTypes.UNSUBSCRIBE, PacketTypes.UNSUBACK, + PacketTypes.DISCONNECT, PacketTypes.AUTH, PacketTypes.WILLMESSAGE]), + 39 : (self.types.index("Four Byte Integer"), + [PacketTypes.CONNECT, PacketTypes.CONNACK]), + 40 : (self.types.index("Byte"), [PacketTypes.CONNACK]), + 41 : (self.types.index("Byte"), [PacketTypes.CONNACK]), + 42 : (self.types.index("Byte"), [PacketTypes.CONNACK]), + } + + def getIdentFromName(self, compressedName): + # return the identifier corresponding to the property name + result = -1 + for name in self.names.keys(): + if compressedName == name.replace(' ', ''): + result = self.names[name] + break + return result + + def __setattr__(self, name, value): + name = name.replace(' ', '') + privateVars = ["packetType", "types", "names", "properties"] + if name in privateVars: + object.__setattr__(self, name, value) + else: + # the name could have spaces in, or not. Remove spaces before assignment + if name not in [name.replace(' ', '') for name in self.names.keys()]: + raise MQTTException("Attribute name must be one of "+str(self.names.keys())) + # check that this attribute applies to the packet type + if self.packetType not in self.properties[self.getIdentFromName(name)][1]: + raise MQTTException("Attribute %s does not apply to packet type %s" + % (name, Packets.Names[self.packetType]) ) + object.__setattr__(self, name, value) + + def __str__(self): + buffer = "[" + first = True + for name in self.names.keys(): + compressedName = name.replace(' ', '') + if hasattr(self, compressedName): + if not first: + buffer += ", " + buffer += compressedName +" : "+str(getattr(self, compressedName)) + first = False + buffer += "]" + return buffer + + def isEmpty(self): + rc = True + for name in self.names.keys(): + compressedName = name.replace(' ', '') + if hasattr(self, compressedName): + rc = False + break + return rc + + def clear(self): + for name in self.names.keys(): + compressedName = name.replace(' ', '') + if hasattr(self, compressedName): + delattr(self, compressedName) + + def writeProperty(self, identifier, type, value): + buffer = b"" + buffer += VBIs.encode(identifier) # identifier + if type == self.types.index("Byte"): # value + buffer += bytes([value]) + elif type == self.types.index("Two Byte Integer"): + buffer += writeInt16(value) + elif type == self.types.index("Four Byte Integer"): + buffer += writeInt32(value) + elif type == self.types.index("Variable Byte Integer"): + buffer += VBIs.encode(value) + elif type == self.types.index("Binary Data"): + buffer += writeBytes(value) + elif type == self.types.index("UTF-8 Encoded String"): + buffer += writeUTF(value) + elif type == self.types.index("UTF-8 String Pair"): + buffer += writeUTF(value[0]) + writeUTF(value[1]) + return buffer + + def pack(self): + # serialize properties into buffer for sending over network + buffer = b"" + for name in self.names.keys(): + compressedName = name.replace(' ', '') + isList = False + if compressedName.endswith('List'): + isList = True + if hasattr(self, compressedName): + identifier = self.getIdentFromName(compressedName) + attr_type = self.properties[identifier][0] + if isList: + for prop in getattr(self, compressedName): + buffer += self.writeProperty(identifier, attr_type, prop) + else: + buffer += self.writeProperty(identifier, attr_type, + getattr(self, compressedName)) + return VBIs.encode(len(buffer)) + buffer + + def readProperty(self, buffer, type, propslen): + if type == self.types.index("Byte"): + value = buffer[0] + valuelen = 1 + elif type == self.types.index("Two Byte Integer"): + value = readInt16(buffer) + valuelen = 2 + elif type == self.types.index("Four Byte Integer"): + value = readInt32(buffer) + valuelen = 4 + elif type == self.types.index("Variable Byte Integer"): + value, valuelen = VBIs.decode(buffer) + elif type == self.types.index("Binary Data"): + value, valuelen = readBytes(buffer) + elif type == self.types.index("UTF-8 Encoded String"): + value, valuelen = readUTF(buffer, propslen) + elif type == self.types.index("UTF-8 String Pair"): + value, valuelen = readUTF(buffer, propslen) + buffer = buffer[valuelen:] # strip the bytes used by the value + value1, valuelen1 = readUTF(buffer, propslen - valuelen) + value = (value, value1) + valuelen += valuelen1 + return value, valuelen + + def getNameFromIdent(self, identifier): + rc = None + for name in self.names: + if self.names[name] == identifier: + rc = name + return rc + + def unpack(self, buffer): + self.clear() + # deserialize properties into attributes from buffer received from network + propslen, VBIlen = VBIs.decode(buffer) + buffer = buffer[VBIlen:] # strip the bytes used by the VBI + propslenleft = propslen + while propslenleft > 0: # properties length is 0 if there are none + identifier, VBIlen = VBIs.decode(buffer) # property identifier + buffer = buffer[VBIlen:] # strip the bytes used by the VBI + propslenleft -= VBIlen + attr_type = self.properties[identifier][0] + value, valuelen = self.readProperty(buffer, attr_type, propslenleft) + buffer = buffer[valuelen:] # strip the bytes used by the value + propslenleft -= valuelen + propname = self.getNameFromIdent(identifier) + compressedName = propname.replace(' ', '') + if propname.endswith('List'): + if not hasattr(self, compressedName): + setattr(self, propname, [value]) + else: + setattr(self, propname, getattr(self, compressedName) + [value]) + else: + if hasattr(self, compressedName): + raise MQTTException("Property '%s' must not exist more than once" % property) + setattr(self, propname, value) + return self, propslen + VBIlen + + +class Connects(Packets): + + def __init__(self, buffer = None): + object.__setattr__(self, "names", + ["fh", "properties", "willProperties", "ProtocolName", "ProtocolVersion", + "ClientIdentifier", "CleanStart", "KeepAliveTimer", + "WillFlag", "WillQoS", "WillRETAIN", "WillTopic", "WillMessage", + "usernameFlag", "passwordFlag", "username", "password"]) + + self.fh = FixedHeaders(PacketTypes.CONNECT) + + # variable header + self.ProtocolName = "MQTT" + self.ProtocolVersion = 5 + self.CleanStart = True + self.WillFlag = False + self.WillQoS = 0 + self.WillRETAIN = 0 + self.KeepAliveTimer = 30 + self.usernameFlag = False + self.passwordFlag = False + + self.properties = Properties(PacketTypes.CONNECT) + self.willProperties = Properties(PacketTypes.WILLMESSAGE) + + # Payload + self.ClientIdentifier = "" # UTF-8 + self.WillTopic = None # UTF-8 + self.WillMessage = None # binary + self.username = None # UTF-8 + self.password = None # binary + + #self.properties = Properties() + if buffer != None: + self.unpack(buffer) + + def pack(self): + connectFlags = bytes([(self.CleanStart << 1) | (self.WillFlag << 2) | \ + (self.WillQoS << 3) | (self.WillRETAIN << 5) | \ + (self.usernameFlag << 6) | (self.passwordFlag << 7)]) + buffer = writeUTF(self.ProtocolName) + bytes([self.ProtocolVersion]) + \ + connectFlags + writeInt16(self.KeepAliveTimer) + buffer += self.properties.pack() + buffer += writeUTF(self.ClientIdentifier) + if self.WillFlag: + assert self.willProperties.packetType == PacketTypes.WILLMESSAGE + buffer += self.willProperties.pack() + buffer += writeUTF(self.WillTopic) + buffer += writeBytes(self.WillMessage) + if self.usernameFlag: + buffer += writeUTF(self.username) + if self.passwordFlag: + buffer += writeBytes(self.password) + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.CONNECT + + try: + fhlen = self.fh.unpack(buffer, maximumPacketSize) + packlen = fhlen + self.fh.remainingLength + assert len(buffer) >= packlen, "buffer length %d packet length %d" % (len(buffer), packlen) + curlen = fhlen # points to after header + remaining length + assert self.fh.DUP == False, "[MQTT-2.1.2-1]" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] QoS was not 0, was %d" % self.fh.QoS + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" + + # to allow the server to send back a CONNACK with unsupported protocol version, + # the following two assertions will need to be disabled + self.ProtocolName, valuelen = readUTF(buffer[curlen:], packlen - curlen) + curlen += valuelen + assert self.ProtocolName == "MQTT", "Wrong protocol name %s" % self.ProtocolName + + self.ProtocolVersion = buffer[curlen] + curlen += 1 + assert self.ProtocolVersion == 5, "Wrong protocol version %s" % self.ProtocolVersion + + connectFlags = buffer[curlen] + assert (connectFlags & 0x01) == 0, "[MQTT-3.1.2-3] reserved connect flag must be 0" + self.CleanStart = ((connectFlags >> 1) & 0x01) == 1 + self.WillFlag = ((connectFlags >> 2) & 0x01) == 1 + self.WillQoS = (connectFlags >> 3) & 0x03 + self.WillRETAIN = (connectFlags >> 5) & 0x01 + self.passwordFlag = ((connectFlags >> 6) & 0x01) == 1 + self.usernameFlag = ((connectFlags >> 7) & 0x01) == 1 + curlen += 1 + + if self.WillFlag: + assert self.WillQoS in [0, 1, 2], "[MQTT-3.1.2-12] will qos must not be 3" + else: + assert self.WillQoS == 0, "[MQTT-3.1.2-11] will qos must be 0, if will flag is false" + assert self.WillRETAIN == False, "[MQTT-3.1.2-13] will retain must be false, if will flag is false" + + self.KeepAliveTimer = readInt16(buffer[curlen:]) + curlen += 2 + + curlen += self.properties.unpack(buffer[curlen:])[1] + + logger.info("[MQTT-3.1.3-3] Clientid must be present, and first field") + logger.info("[MQTT-3.1.3-4] Clientid must be Unicode, and between 0 and 65535 bytes long") + self.ClientIdentifier, valuelen = readUTF(buffer[curlen:], packlen - curlen) + curlen += valuelen + + if self.WillFlag: + curlen += self.willProperties.unpack(buffer[curlen:])[1] + self.WillTopic, valuelen = readUTF(buffer[curlen:], packlen - curlen) + curlen += valuelen + self.WillMessage, valuelen = readBytes(buffer[curlen:]) + curlen += valuelen + logger.info("[[MQTT-3.1.2-9] will topic and will message fields must be present") + else: + self.WillTopic = self.WillMessage = None + + if self.usernameFlag: + assert len(buffer) > curlen+2, "Buffer too short to read username length" + self.username, valuelen = readUTF(buffer[curlen:], packlen - curlen) + curlen += valuelen + logger.info("[MQTT-3.1.2-19] username must be in payload if user name flag is 1") + else: + logger.info("[MQTT-3.1.2-18] username must not be in payload if user name flag is 0") + assert self.passwordFlag == False, "[MQTT-3.1.2-22] password flag must be 0 if username flag is 0" + + if self.passwordFlag: + assert len(buffer) > curlen+2, "Buffer too short to read password length" + self.password, valuelen = readBytes(buffer[curlen:]) + curlen += valuelen + logger.info("[MQTT-3.1.2-21] password must be in payload if password flag is 0") + else: + logger.info("[MQTT-3.1.2-20] password must not be in payload if password flag is 0") + + if self.WillFlag and self.usernameFlag and self.passwordFlag: + logger.info("[MQTT-3.1.3-1] clientid, will topic, will message, username and password all present") + + assert curlen == packlen, "Packet is wrong length curlen %d != packlen %d" % (curlen, packlen) + except: + logger.exception("[MQTT-3.1.4-1] server must validate connect packet and close connection without connack if it does not conform") + raise + + def __str__(self): + buf = str(self.fh)+", ProtocolName="+str(self.ProtocolName)+", ProtocolVersion=" +\ + str(self.ProtocolVersion)+", CleanStart="+str(self.CleanStart) +\ + ", WillFlag="+str(self.WillFlag)+", KeepAliveTimer=" +\ + str(self.KeepAliveTimer)+", ClientId="+str(self.ClientIdentifier) +\ + ", usernameFlag="+str(self.usernameFlag)+", passwordFlag="+str(self.passwordFlag) + if self.WillFlag: + buf += ", WillQoS=" + str(self.WillQoS) +\ + ", WillRETAIN=" + str(self.WillRETAIN) +\ + ", WillTopic='"+ self.WillTopic +\ + "', WillMessage='"+str(self.WillMessage)+"'" + if self.username: + buf += ", username="+self.username + if self.password: + buf += ", password="+str(self.password) + buf += ", properties="+str(self.properties) + return buf+")" + + def __eq__(self, packet): + rc = Packets.__eq__(self, packet) and \ + self.ProtocolName == packet.ProtocolName and \ + self.ProtocolVersion == packet.ProtocolVersion and \ + self.CleanStart == packet.CleanStart and \ + self.WillFlag == packet.WillFlag and \ + self.KeepAliveTimer == packet.KeepAliveTimer and \ + self.ClientIdentifier == packet.ClientIdentifier and \ + self.WillFlag == packet.WillFlag + if rc and self.WillFlag: + rc = self.WillQoS == packet.WillQoS and \ + self.WillRETAIN == packet.WillRETAIN and \ + self.WillTopic == packet.WillTopic and \ + self.WillMessage == packet.WillMessage + if rc: + rc = self.properties == packet.properties + return rc + + +class Connacks(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, ReasonCode="Success"): + object.__setattr__(self, "names", + ["fh", "sessionPresent", "reasonCode", "properties"]) + self.fh = FixedHeaders(PacketTypes.CONNACK) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + self.sessionPresent = False + self.reasonCode = ReasonCodes(PacketTypes.CONNACK, ReasonCode) + self.properties = Properties(PacketTypes.CONNACK) + if buffer != None: + self.unpack(buffer) + + def pack(self): + flags = 0x01 if self.sessionPresent else 0x00 + buffer = bytes([flags]) + buffer += self.reasonCode.pack() + buffer += self.properties.pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 4 + assert PacketType(buffer) == PacketTypes.CONNACK + curlen = self.fh.unpack(buffer, maximumPacketSize) + assert buffer[curlen] in [0, 1], "Connect Acknowledge Flags" + self.sessionPresent = (buffer[curlen] == 0x01) + curlen += 1 + curlen += self.reasonCode.unpack(buffer[curlen:]) + curlen += self.properties.unpack(buffer[curlen:])[1] + assert self.fh.DUP == False, "[MQTT-2.1.2-1]" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1]" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" + + def __str__(self): + return str(self.fh)+", Session present="+str((self.sessionPresent & 0x01) == 1)+\ + ", ReturnCode="+str(self.reasonCode)+\ + ", properties="+str(self.properties)+")" + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.reasonCode == packet.reasonCode + + +class Disconnects(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, + reasonCode="Normal disconnection"): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "reasonCode", "properties"]) + self.fh = FixedHeaders(PacketTypes.DISCONNECT) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.reasonCode = ReasonCodes(PacketTypes.DISCONNECT, identifier=reasonCode) + self.properties = Properties(PacketTypes.DISCONNECT) + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = b"" + if self.reasonCode.getName() != "Normal disconnection" or not self.properties.isEmpty(): + buffer += self.reasonCode.pack() + if not self.properties.isEmpty(): + buffer += self.properties.pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + self.properties.clear() + self.reasonCode.set("Normal disconnection") + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.DISCONNECT + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + assert self.fh.DUP == False, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + curlen = 0 + if self.fh.remainingLength > 0: + self.reasonCode.unpack(buffer[curlen:]) + curlen += 1 + if self.fh.remainingLength > 1: + curlen += self.properties.unpack(buffer[curlen:])[1] + assert curlen == self.fh.remainingLength, \ + "DISCONNECT packet is wrong length %d" % self.fh.remainingLength + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", ReasonCode: "+str(self.reasonCode)+", Properties: "+str(self.properties) + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.reasonCode == packet.reasonCode and \ + self.properties == packet.properties + + +class Publishes(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, MsgId=1, TopicName="", Payload=b""): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "topicName", "packetIdentifier", + "properties", "data", "qos2state", "receivedTime"]) + self.fh = FixedHeaders(PacketTypes.PUBLISH) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.topicName = TopicName + self.packetIdentifier = MsgId + self.properties = Properties(PacketTypes.PUBLISH) + # payload + self.data = Payload + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = writeUTF(self.topicName) + if self.fh.QoS != 0: + buffer += writeInt16(self.packetIdentifier) + buffer += self.properties.pack() + buffer += self.data + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.PUBLISH + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert self.fh.QoS in [0, 1, 2], "QoS in Publish must be 0, 1, or 2" + packlen = fhlen + self.fh.remainingLength + assert len(buffer) >= packlen + curlen = fhlen + try: + self.topicName, valuelen = readUTF(buffer[fhlen:], packlen - curlen) + except UnicodeDecodeError: + logger.info("[MQTT-3.3.2-1] topic name in publish must be utf-8") + raise + curlen += valuelen + if self.fh.QoS != 0: + self.packetIdentifier = readInt16(buffer[curlen:]) + logger.info("[MQTT-2.3.1-1] packet indentifier must be in publish if QoS is 1 or 2") + curlen += 2 + assert self.packetIdentifier > 0, "[MQTT-2.3.1-1] packet indentifier must be > 0" + else: + logger.info("[MQTT-2.3.1-5] no packet indentifier in publish if QoS is 0") + self.packetIdentifier = 0 + curlen += self.properties.unpack(buffer[curlen:])[1] + self.data = buffer[curlen:fhlen + self.fh.remainingLength] + if self.fh.QoS == 0: + assert self.fh.DUP == False, "[MQTT-2.1.2-4]" + return fhlen + self.fh.remainingLength + + def __str__(self): + rc = str(self.fh) + if self.fh.QoS != 0: + rc += ", PacketId="+str(self.packetIdentifier) + rc += ", Properties: "+str(self.properties) + rc += ", TopicName="+str(self.topicName)+", Payload="+str(self.data)+")" + return rc + + def __eq__(self, packet): + rc = Packets.__eq__(self, packet) and \ + self.topicName == packet.topicName and \ + self.data == packet.data + if rc and self.fh.QoS != 0: + rc = self.packetIdentifier == packet.packetIdentifier + return rc + + +class Acks(Packets): + + def __init__(self, ackType, buffer, DUP, QoS, RETAIN, packetId): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "packetIdentifier", + "reasonCode", "properties"]) + self.fh = FixedHeaders(ackType) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.packetIdentifier = packetId + self.reasonCode = ReasonCodes(ackType) + self.properties = Properties(ackType) + object.__setattr__(self, "ackType", ackType) + object.__setattr__(self, "ackName", Packets.Names[self.ackType]) + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = writeInt16(self.packetIdentifier) + if self.reasonCode.getName() != "Success" or not self.properties.isEmpty(): + buffer += self.reasonCode.pack() + if not self.properties.isEmpty(): + buffer += self.properties.pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + self.properties.clear() + self.reasonCode.set("Success") + assert len(buffer) >= 2 + assert PacketType(buffer) == self.ackType + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert self.fh.remainingLength in [2, 3, 4], \ + "%s packet is wrong length %d" % (self.ackName, self.fh.remainingLength) + assert len(buffer) >= fhlen + self.fh.remainingLength + self.packetIdentifier = readInt16(buffer[fhlen:]) + curlen = fhlen + 2 + assert self.fh.DUP == False, "[MQTT-2.1.2-1] %s reserved bits must be 0" %\ + self.ackName + if self.ackType == PacketTypes.PUBREL: + assert self.fh.QoS == 1, "[MQTT-3.6.1-1] %s reserved bits must be 0010" %\ + self.ackName + else: + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] %s reserved bits must be 0" %\ + self.ackName + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] %s reserved bits must be 0" %\ + self.ackName + if self.fh.remainingLength > 2: + self.reasonCode.unpack(buffer[curlen:]) + curlen += 1 + if self.fh.remainingLength > 3: + self.properties.unpack(buffer[curlen:]) + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", PacketId="+str(self.packetIdentifier)+")" + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.packetIdentifier == packet.packetIdentifier + + +class Pubacks(Acks): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, PacketId=1): + Acks.__init__(self, PacketTypes.PUBACK, buffer, DUP, QoS, RETAIN, PacketId) + +class Pubrecs(Acks): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, PacketId=1): + Acks.__init__(self, PacketTypes.PUBREC, buffer, DUP, QoS, RETAIN, PacketId) + +class Pubrels(Acks): + + def __init__(self, buffer=None, DUP=False, QoS=1, RETAIN=False, PacketId=1): + Acks.__init__(self, PacketTypes.PUBREL, buffer, DUP, QoS, RETAIN, PacketId) + +class Pubcomps(Acks): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, PacketId=1): + Acks.__init__(self, PacketTypes.PUBCOMP, buffer, DUP, QoS, RETAIN, PacketId) + +class SubscribeOptions(object): + + def __init__(self, QoS=0, noLocal=False, retainAsPublished=False, retainHandling=0): + object.__setattr__(self, "names", + ["QoS", "noLocal", "retainAsPublished", "retainHandling"]) + self.QoS = QoS # bits 0,1 + self.noLocal = noLocal # bit 2 + self.retainAsPublished = retainAsPublished # bit 3 + self.retainHandling = retainHandling # bits 4 and 5: 0, 1 or 2 + + def __setattr__(self, name, value): + if name not in self.names: + raise MQTTException(name + " Attribute name must be one of "+str(self.names)) + object.__setattr__(self, name, value) + + def pack(self): + assert self.QoS in [0, 1, 2] + assert self.retainHandling in [0, 1, 2] + noLocal = 1 if self.noLocal else 0 + retainAsPublished = 1 if self.retainAsPublished else 0 + buffer = bytes([(self.retainHandling << 4) | (retainAsPublished << 3) |\ + (noLocal << 2) | self.QoS]) + return buffer + + def unpack(self, buffer): + b0 = buffer[0] + self.retainHandling = ((b0 >> 4) & 0x03) + self.retainAsPublished = True if ((b0 >> 3) & 0x01) == 1 else False + self.noLocal = True if ((b0 >> 2) & 0x01) == 1 else False + self.QoS = (b0 & 0x03) + assert self.retainHandling in [0, 1, 2] + assert self.QoS in [0, 1, 2] + return 1 + + def __str__(self): + return "{QoS="+str(self.QoS)+", noLocal="+str(self.noLocal)+\ + ", retainAsPublished="+str(self.retainAsPublished)+\ + ", retainHandling="+str(self.retainHandling)+"}" + + +class Subscribes(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=1, RETAIN=False, MsgId=1, Data=[]): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "packetIdentifier", + "properties", "data"]) + self.fh = FixedHeaders(PacketTypes.SUBSCRIBE) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.packetIdentifier = MsgId + self.properties = Properties(PacketTypes.SUBSCRIBE) + # payload - list of topic, subscribe option pairs + self.data = Data[:] + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = writeInt16(self.packetIdentifier) + buffer += self.properties.pack() + for d in self.data: + buffer += writeUTF(d[0]) + d[1].pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + self.properties.clear() + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.SUBSCRIBE + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + logger.info("[MQTT-2.3.1-1] packet indentifier must be in subscribe") + self.packetIdentifier = readInt16(buffer[fhlen:]) + assert self.packetIdentifier > 0, "[MQTT-2.3.1-1] packet indentifier must be > 0" + leftlen = self.fh.remainingLength - 2 + leftlen -= self.properties.unpack(buffer[-leftlen:])[1] + self.data = [] + while leftlen > 0: + topic, topiclen = readUTF(buffer[-leftlen:], leftlen) + leftlen -= topiclen + options = SubscribeOptions() + options.unpack(buffer[-leftlen:]) + leftlen -= 1 + self.data.append((topic, options)) + assert len(self.data) > 0, "[MQTT-3.8.3-1] at least one topic, qos pair must be in subscribe" + assert leftlen == 0 + assert self.fh.DUP == False, "[MQTT-2.1.2-1] DUP must be false in subscribe" + assert self.fh.QoS == 1, "[MQTT-2.1.2-1] QoS must be 1 in subscribe" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] RETAIN must be false in subscribe" + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", PacketId="+str(self.packetIdentifier)+\ + ", Properties: "+str(self.properties)+\ + ", Data="+str( [(x, str(y)) for (x, y) in self.data] ) +")" + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.packetIdentifier == packet.packetIdentifier and \ + self.data == packet.data + + +class UnsubSubacks(Packets): + + def __init__(self, packetType, buffer, DUP, QoS, RETAIN, PacketId, reasonCodes): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "packetIdentifier", + "reasonCodes", "properties"]) + object.__setattr__(self, "packetType", packetType) + self.fh = FixedHeaders(self.packetType) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.packetIdentifier = PacketId + self.properties = Properties(self.packetType) + # payload - list of reason codes corresponding to topics in subscribe + self.reasonCodes = reasonCodes[:] + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = writeInt16(self.packetIdentifier) + buffer += self.properties.pack() + for reasonCode in self.reasonCodes: + buffer += reasonCode.pack() + buffer = self.fh.pack(len(buffer)) + buffer + assert len(buffer) >= 3 # must have property field, even if empty + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 3 + assert PacketType(buffer) == self.packetType + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + self.packetIdentifier = readInt16(buffer[fhlen:]) + leftlen = self.fh.remainingLength - 2 + leftlen -= self.properties.unpack(buffer[-leftlen:])[1] + self.reasonCodes = [] + while leftlen > 0: + if self.packetType == PacketTypes.SUBACK: + reasonCode = ReasonCodes(self.packetType, "Granted QoS 0") + else: + reasonCode = ReasonCodes(self.packetType, "Success") + reasonCode.unpack(buffer[-leftlen:]) + assert reasonCode.value in [0, 1, 2, 0x80], "[MQTT-3.9.3-2] return code in QoS must be 0, 1, 2 or 0x80" + leftlen -= 1 + self.reasonCodes.append(reasonCode) + assert leftlen == 0 + assert self.fh.DUP == False, "[MQTT-2.1.2-1] DUP should be false in suback" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] QoS should be 0 in suback" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] Retain should be false in suback" + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", PacketId="+str(self.packetIdentifier)+\ + ", Properties: "+str(self.properties)+\ + ", reason codes="+str([str(rc) for rc in self.reasonCodes])+")" + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.packetIdentifier == packet.packetIdentifier and \ + self.data == packet.data + + +class Subacks(UnsubSubacks): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, PacketId=1, reasonCodes=[]): + UnsubSubacks.__init__(self, PacketTypes.SUBACK, buffer, DUP, QoS, RETAIN, PacketId, reasonCodes) + + +class Unsubscribes(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=1, RETAIN=False, PacketId=1, TopicFilters=[]): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "packetIdentifier", "properties", "topicFilters"]) + self.fh = FixedHeaders(PacketTypes.UNSUBSCRIBE) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.packetIdentifier = PacketId + self.properties = Properties(PacketTypes.UNSUBSCRIBE) + # payload - list of topics + self.topicFilters = TopicFilters[:] + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = writeInt16(self.packetIdentifier) + buffer += self.properties.pack() + for topicFilter in self.topicFilters: + buffer += writeUTF(topicFilter) + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.UNSUBSCRIBE + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + logger.info("[MQTT-2.3.1-1] packet indentifier must be in unsubscribe") + self.packetIdentifier = readInt16(buffer[fhlen:]) + assert self.packetIdentifier > 0, "[MQTT-2.3.1-1] packet indentifier must be > 0" + leftlen = self.fh.remainingLength - 2 + leftlen -= self.properties.unpack(buffer[-leftlen:])[1] + self.topicFilters = [] + while leftlen > 0: + topic, topiclen = readUTF(buffer[-leftlen:], leftlen) + leftlen -= topiclen + self.topicFilters.append(topic) + assert leftlen == 0 + assert self.fh.DUP == False, "[MQTT-2.1.2-1]" + assert self.fh.QoS == 1, "[MQTT-2.1.2-1]" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" + logger.info("[MQTT-3-10.1-1] fixed header bits are 0,0,1,0") + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", PacketId="+str(self.packetIdentifier)+\ + ", Properties: "+str(self.properties)+\ + ", Data="+str(self.topicFilters)+")" + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.packetIdentifier == packet.packetIdentifier and \ + self.topicFilters == packet.topicFilters + + +class Unsubacks(UnsubSubacks): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, PacketId=1, reasonCodes=[]): + UnsubSubacks.__init__(self, PacketTypes.UNSUBACK, buffer, DUP, QoS, RETAIN, + PacketId, reasonCodes) + + +class Pingreqs(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False): + object.__setattr__(self, "names", ["fh", "DUP", "QoS", "RETAIN"]) + self.fh = FixedHeaders(PacketTypes.PINGREQ) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + if buffer != None: + self.unpack(buffer) + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.PINGREQ + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert self.fh.remainingLength == 0 + assert self.fh.DUP == False, "[MQTT-2.1.2-1]" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1]" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" + return fhlen + + def __str__(self): + return str(self.fh)+")" + + +class Pingresps(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False): + object.__setattr__(self, "names", ["fh", "DUP", "QoS", "RETAIN"]) + self.fh = FixedHeaders(PacketTypes.PINGRESP) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + if buffer != None: + self.unpack(buffer) + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.PINGRESP + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert self.fh.remainingLength == 0 + assert self.fh.DUP == False, "[MQTT-2.1.2-1]" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1]" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1]" + return fhlen + + def __str__(self): + return str(self.fh)+")" + + +class Disconnects(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, + reasonCode="Normal disconnection"): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "reasonCode", "properties"]) + self.fh = FixedHeaders(PacketTypes.DISCONNECT) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.reasonCode = ReasonCodes(PacketTypes.DISCONNECT, aName=reasonCode) + self.properties = Properties(PacketTypes.DISCONNECT) + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = b"" + if self.reasonCode.getName() != "Normal disconnection" or not self.properties.isEmpty(): + buffer += self.reasonCode.pack() + if not self.properties.isEmpty(): + buffer += self.properties.pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + self.properties.clear() + self.reasonCode.set("Normal disconnection") + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.DISCONNECT + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + assert self.fh.DUP == False, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] DISCONNECT reserved bits must be 0" + curlen = fhlen + if self.fh.remainingLength > 0: + self.reasonCode.unpack(buffer[curlen:]) + curlen += 1 + if self.fh.remainingLength > 1: + curlen += self.properties.unpack(buffer[curlen:])[1] + assert curlen == fhlen + self.fh.remainingLength, \ + "DISCONNECT packet is wrong length %d" % self.fh.remainingLength + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", ReasonCode: "+str(self.reasonCode)+", Properties: "+str(self.properties) + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.reasonCode == packet.reasonCode and \ + self.properties == packet.properties + + +class Auths(Packets): + + def __init__(self, buffer=None, DUP=False, QoS=0, RETAIN=False, + reasonCode="Success"): + object.__setattr__(self, "names", + ["fh", "DUP", "QoS", "RETAIN", "reasonCode", "properties"]) + self.fh = FixedHeaders(PacketTypes.AUTH) + self.fh.DUP = DUP + self.fh.QoS = QoS + self.fh.RETAIN = RETAIN + # variable header + self.reasonCode = ReasonCodes(PacketTypes.AUTH, reasonCode) + self.properties = Properties(PacketTypes.AUTH) + if buffer != None: + self.unpack(buffer) + + def pack(self): + buffer = self.reasonCode.pack() + buffer += self.properties.pack() + buffer = self.fh.pack(len(buffer)) + buffer + return buffer + + def unpack(self, buffer, maximumPacketSize): + assert len(buffer) >= 2 + assert PacketType(buffer) == PacketTypes.AUTH + fhlen = self.fh.unpack(buffer, maximumPacketSize) + assert len(buffer) >= fhlen + self.fh.remainingLength + assert self.fh.DUP == False, "[MQTT-2.1.2-1] AUTH reserved bits must be 0" + assert self.fh.QoS == 0, "[MQTT-2.1.2-1] AUTH reserved bits must be 0" + assert self.fh.RETAIN == False, "[MQTT-2.1.2-1] AUTH reserved bits must be 0" + curlen = fhlen + curlen += self.reasonCode.unpack(buffer[curlen:]) + curlen += self.properties.unpack(buffer[curlen:])[1] + assert curlen == fhlen + self.fh.remainingLength, \ + "AUTH packet is wrong length %d %d" % (self.fh.remainingLength, curlen) + return fhlen + self.fh.remainingLength + + def __str__(self): + return str(self.fh)+", ReasonCode: "+str(self.reasonCode)+", Properties: "+str(self.properties) + + def __eq__(self, packet): + return Packets.__eq__(self, packet) and \ + self.reasonCode == packet.reasonCode and \ + self.properties == packet.properties + + +classes = [Connects, Connacks, Publishes, Pubacks, Pubrecs, + Pubrels, Pubcomps, Subscribes, Subacks, Unsubscribes, + Unsubacks, Pingreqs, Pingresps, Disconnects, Auths] + +def unpackPacket(buffer, maximumPacketSize=MAX_PACKET_SIZE): + if PacketType(buffer) != None: + packet = classes[PacketType(buffer)-1]() + packet.unpack(buffer, maximumPacketSize=maximumPacketSize) + else: + packet = None + return packet diff --git a/test/mqttsas.py b/test/mqttsas.py index f32530c6..63195250 100755 --- a/test/mqttsas.py +++ b/test/mqttsas.py @@ -13,6 +13,7 @@ Contributors: Ian Craggs - initial implementation and/or documentation + Ian Craggs - add MQTTV5 support ******************************************************************* """ from __future__ import print_function @@ -20,11 +21,15 @@ from __future__ import print_function import socket, sys, select, traceback, datetime, os try: import socketserver - import MQTTV311 as MQTTV3 # Trace MQTT traffic - Python 3 version + import MQTTV311 # Trace MQTT traffic - Python 3 version + import MQTTV5 except: + traceback.print_exc() import SocketServer as socketserver - import MQTTV3112 as MQTTV3 # Trace MQTT traffic - Python 2 version + import MQTTV3112 as MQTTV311 # Trace MQTT traffic - Python 2 version + import MQTTV5 +MQTT = MQTTV311 logging = True myWindow = None @@ -38,6 +43,7 @@ suspended = [] class MyHandler(socketserver.StreamRequestHandler): def handle(self): + global MQTT if not hasattr(self, "ids"): self.ids = {} if not hasattr(self, "versions"): @@ -55,12 +61,30 @@ class MyHandler(socketserver.StreamRequestHandler): if s in suspended: print("suspended") if s == clients and s not in suspended: - inbuf = MQTTV3.getPacket(clients) # get one packet + inbuf = MQTT.getPacket(clients) # get one packet if inbuf == None: break try: - packet = MQTTV3.unpackPacket(inbuf) - if packet.fh.MessageType == MQTTV3.PUBLISH and \ + # if connect, this could be MQTTV3 or MQTTV5 + if inbuf[0] >> 4 == 1: # connect packet + protocol_string = b'MQTT' + pos = inbuf.find(protocol_string) + if pos != -1: + version = inbuf[pos + len(protocol_string)] + if version == 5: + MQTT = MQTTV5 + else: + MQTT = MQTTV311 + packet = MQTT.unpackPacket(inbuf) + if hasattr(packet.fh, "MessageType"): + packet_type = packet.fh.MessageType + publish_type = MQTT.PUBLISH + connect_type = MQTT.CONNECT + else: + packet_type = packet.fh.PacketType + publish_type = MQTT.PacketTypes.PUBLISH + connect_type = MQTT.PacketTypes.CONNECT + if packet_type == publish_type and \ packet.topicName == "MQTTSAS topic" and \ packet.data == b"TERMINATE": print("Terminating client", self.ids[id(clients)]) @@ -68,26 +92,26 @@ class MyHandler(socketserver.StreamRequestHandler): clients.close() terminated = True break - elif packet.fh.MessageType == MQTTV3.PUBLISH and \ + elif packet_type == publish_type and \ packet.topicName == "MQTTSAS topic" and \ packet.data == b"TERMINATE_SERVER": print("Suspending client ", self.ids[id(clients)]) suspended.append(clients) - elif packet.fh.MessageType == MQTTV3.CONNECT: + elif packet_type == connect_type: self.ids[id(clients)] = packet.ClientIdentifier self.versions[id(clients)] = 3 - print(timestamp() , "C to S", self.ids[id(clients)], repr(packet)) + print(timestamp() , "C to S", self.ids[id(clients)], str(packet)) #print([hex(b) for b in inbuf]) #print(inbuf) except: traceback.print_exc() brokers.send(inbuf) # pass it on elif s == brokers: - inbuf = MQTTV3.getPacket(brokers) # get one packet + inbuf = MQTT.getPacket(brokers) # get one packet if inbuf == None: break try: - print(timestamp(), "S to C", self.ids[id(clients)], repr(MQTTV3.unpackPacket(inbuf))) + print(timestamp(), "S to C", self.ids[id(clients)], str(MQTT.unpackPacket(inbuf))) except: traceback.print_exc() clients.send(inbuf) diff --git a/test/test1.c b/test/test1.c index c5358f36..1cc5f720 100644 --- a/test/test1.c +++ b/test/test1.c @@ -1166,7 +1166,7 @@ int main(int argc, char** argv) fprintf(xml, "\n", (int)(ARRAY_SIZE(tests) - 1)); setenv("MQTT_C_CLIENT_TRACE", "ON", 1); - setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0); + setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1); getopts(argc, argv); diff --git a/test/test15.c b/test/test15.c index c8e97e06..3bae917a 100644 --- a/test/test15.c +++ b/test/test15.c @@ -20,16 +20,10 @@ /** * @file - * Tests for the MQ Telemetry MQTT C client + * Tests for the MQTT C client */ -/* -#if !defined(_RTSHEADER) - #include -#endif -*/ - #include "MQTTClient.h" #include #include @@ -542,19 +536,31 @@ void test2_deliveryComplete(void* context, MQTTClient_deliveryToken dt) ++test2_deliveryCompleted; } -int test2_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m) +int test2_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message) { ++test2_arrivedcount; MyLog(LOGA_DEBUG, "Callback: %d message received on topic %s is %.*s.", - test2_arrivedcount, topicName, m->payloadlen, (char*)(m->payload)); - if (test2_pubmsg.payloadlen != m->payloadlen || - memcmp(m->payload, test2_pubmsg.payload, m->payloadlen) != 0) + test2_arrivedcount, topicName, message->payloadlen, (char*)(message->payload)); + + assert("Message structure version should be 1", message->struct_version == 1, + "message->struct_version was %d", message->struct_version); + if (message->struct_version == 1) + { + const int props_count = 0; + + assert("Properties count should be 0", message->properties.count == props_count, + "Properties count was %d\n", message->properties.count); + logProperties(&message->properties); + } + + if (test2_pubmsg.payloadlen != message->payloadlen || + memcmp(message->payload, test2_pubmsg.payload, message->payloadlen) != 0) { failures++; - MyLog(LOGA_INFO, "Error: wrong data received lengths %d %d\n", test2_pubmsg.payloadlen, m->payloadlen); + MyLog(LOGA_INFO, "Error: wrong data received lengths %d %d\n", test2_pubmsg.payloadlen, message->payloadlen); } MQTTClient_free(topicName); - MQTTClient_freeMessage(&m); + MQTTClient_freeMessage(&message); return 1; } @@ -564,7 +570,7 @@ void test2_sendAndReceive(MQTTClient* c, int qos, char* test_topic) MQTTClient_deliveryToken dt; int i = 0; int iterations = 50; - int rc = 0; + MQTTResponse response = {SUCCESS, NULL}; int wait_seconds = 0; test2_deliveryCompleted = 0; @@ -578,11 +584,11 @@ void test2_sendAndReceive(MQTTClient* c, int qos, char* test_topic) for (i = 1; i <= iterations; ++i) { if (i % 10 == 0) - rc = MQTTClient_publish(c, test_topic, test2_pubmsg.payloadlen, test2_pubmsg.payload, - test2_pubmsg.qos, test2_pubmsg.retained, NULL); + response = MQTTClient_publish5(c, test_topic, test2_pubmsg.payloadlen, test2_pubmsg.payload, + test2_pubmsg.qos, test2_pubmsg.retained, NULL, NULL); else - rc = MQTTClient_publishMessage(c, test_topic, &test2_pubmsg, &dt); - assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + response = MQTTClient_publishMessage5(c, test_topic, &test2_pubmsg, &dt); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); #if defined(WIN32) Sleep(100); @@ -633,6 +639,10 @@ int test2(struct Options options) /* TODO - usused - remove ? MQTTClient_deliveryToken* dt = NULL; */ MQTTClient c; MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperties willProps = MQTTProperties_initializer; + MQTTResponse response = {SUCCESS, NULL}; + MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer; int rc = 0; char* test_topic = "C client test2"; @@ -659,12 +669,12 @@ int test2(struct Options options) assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); MyLog(LOGA_DEBUG, "Connecting"); - rc = MQTTClient_connect(c, &opts); + response = MQTTClient_connect5(c, &opts, &props, &willProps); assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); if (rc != MQTTCLIENT_SUCCESS) goto exit; - rc = MQTTClient_subscribe(c, test_topic, subsqos); + response = MQTTClient_subscribe5(c, test_topic, subsqos, &subopts, &props); assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); test2_sendAndReceive(c, 0, test_topic); @@ -673,9 +683,9 @@ int test2(struct Options options) MyLog(LOGA_DEBUG, "Stopping"); - rc = MQTTClient_unsubscribe(c, test_topic); - assert("Unsubscribe successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); - rc = MQTTClient_disconnect(c, 0); + response = MQTTClient_unsubscribe5(c, test_topic, &props); + assert("Unsubscribe successful", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", rc); + rc = MQTTClient_disconnect5(c, 0, SUCCESS, &props); assert("Disconnect successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); MQTTClient_destroy(&c); @@ -788,12 +798,15 @@ int test4_run(int qos) int mytoken = -99; char buffer[100]; int count = 3; + MQTTProperty property; + MQTTProperties props = MQTTProperties_initializer; + MQTTResponse response = {SUCCESS, NULL}; int i, rc; failures = 0; MyLog(LOGA_INFO, "Starting test 4 - persistence, qos %d", qos); - MQTTClient_create(&c, options.connection, "xrctest1_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); + MQTTClient_create(&c, options.connection, "xrctest15_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); opts.keepAliveInterval = 20; opts.reliable = 0; @@ -804,37 +817,31 @@ int test4_run(int qos) opts.serverURIcount = options.hacount; } - MyLog(LOGA_DEBUG, "Cleanup by connecting clean session\n"); + MyLog(LOGA_DEBUG, "Cleanup by connecting clean start, add session expiry > 0\n"); opts.cleansession = 1; - if ((rc = MQTTClient_connect(c, &opts)) != 0) - { - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + property.identifier = SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; /* in seconds */ + MQTTProperties_add(&props, &property); + response = MQTTClient_connect5(c, &opts, &props, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, + "rc was %d", response.reasonCode); + if (response.reasonCode != MQTTCLIENT_SUCCESS) return -1; - } - opts.cleansession = 0; - MQTTClient_disconnect(c, 0); - - MyLog(LOGA_DEBUG, "Connecting\n"); - if ((rc = MQTTClient_connect(c, &opts)) != 0) - { - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); - return -1; - } /* subscribe so we can get messages back */ - rc = MQTTClient_subscribe(c, topic, subsqos); - assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + response = MQTTClient_subscribe5(c, topic, subsqos, NULL, NULL); + assert("Good rc from subscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); /* send messages so that we can receive the same ones */ for (i = 0; i < count; ++i) { sprintf(buffer, "Message sequence no %d", i); - rc = MQTTClient_publish(c, topic, 10, buffer, qos, 0, NULL); - assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + response = MQTTClient_publish5(c, topic, 10, buffer, qos, 0, NULL, NULL); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); } /* disconnect immediately without receiving the incoming messages */ - MQTTClient_disconnect(c, 0); /* now there should be "orphaned" publications */ + MQTTClient_disconnect5(c, 0, SUCCESS, NULL); /* now there should be "orphaned" publications */ rc = MQTTClient_getPendingDeliveryTokens(c, &tokens); assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); @@ -850,10 +857,10 @@ int test4_run(int qos) assert1("no of tokens should be count", i == count, "no of tokens %d count %d", i, count); mytoken = tokens[0]; } - + MQTTProperties_free(&props); MQTTClient_destroy(&c); /* force re-reading persistence on create */ - MQTTClient_create(&c, options.connection, "xrctest1_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); + MQTTClient_create(&c, options.connection, "xrctest15_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); rc = MQTTClient_getPendingDeliveryTokens(c, &tokens); assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); @@ -869,18 +876,18 @@ int test4_run(int qos) } MyLog(LOGA_DEBUG, "Reconnecting"); - if (MQTTClient_connect(c, &opts) != 0) - { - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + opts.cleansession = 0; + response = MQTTClient_connect5(c, &opts, NULL, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + if (response.reasonCode != MQTTCLIENT_SUCCESS) return -1; - } for (i = 0; i < count; ++i) { int dup = 0; do { - dup = 0; + dup = 0; MQTTClient_receive(c, &topicName, &topicLen, &m, 5000); if (m && m->dup) { @@ -888,7 +895,7 @@ int test4_run(int qos) MyLog(LOGA_DEBUG, "Duplicate message id %d", m->msgid); MQTTClient_freeMessage(&m); MQTTClient_free(topicName); - dup = 1; + dup = 1; } } while (dup == 1); assert("should get a message", m != NULL, "m was %p", m); @@ -907,7 +914,7 @@ int test4_run(int qos) assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); assert("should get no tokens back", tokens == NULL, "tokens was %p", tokens); - MQTTClient_disconnect(c, 0); + rc = MQTTClient_disconnect5(c, 0, SUCCESS, NULL); MQTTClient_destroy(&c); @@ -951,6 +958,9 @@ int test5(struct Options options) MQTTClient_deliveryToken* tokens = NULL; char buffer[100]; int count = 5; + MQTTProperty property; + MQTTProperties props = MQTTProperties_initializer; + MQTTResponse response = {SUCCESS, NULL}; int i, rc; fprintf(xml, "message = test6_will_message; opts.will->qos = 1; @@ -1083,7 +1099,7 @@ int test6(struct Options options) goto exit; /* Connect to the broker */ - rc = MQTTClient_connect(test6_c1, &opts); + response = MQTTClient_connect5(test6_c1, &opts, NULL, NULL); assert("good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); if (rc != MQTTCLIENT_SUCCESS) goto exit; @@ -1100,15 +1116,15 @@ int test6(struct Options options) opts2.keepAliveInterval = 20; opts2.cleansession = 1; MyLog(LOGA_INFO, "Connecting Client_2 ..."); - rc = MQTTClient_connect(test6_c2, &opts2); - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_connect5(test6_c2, &opts2, NULL, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); - rc = MQTTClient_subscribe(test6_c2, test6_will_topic, 2); - assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_subscribe5(test6_c2, test6_will_topic, 2, NULL, NULL); + assert("Good rc from subscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); /* now send the command which will break the connection and cause the will message to be sent */ - rc = MQTTClient_publish(test6_c1, mqttsas_topic, (int)strlen("TERMINATE"), "TERMINATE", 0, 0, NULL); - assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_publish5(test6_c1, mqttsas_topic, (int)strlen("TERMINATE"), "TERMINATE", 0, 0, NULL, NULL); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); MyLog(LOGA_INFO, "Waiting to receive the will message"); count = 0; @@ -1127,8 +1143,8 @@ int test6(struct Options options) assert("connection lost called", test6_connection_lost_called == 1, "connection_lost_called %d\n", test6_connection_lost_called); - rc = MQTTClient_unsubscribe(test6_c2, test6_will_topic); - assert("Good rc from unsubscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + response = MQTTClient_unsubscribe5(test6_c2, test6_will_topic, NULL); + assert("Good rc from unsubscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); rc = MQTTClient_isConnected(test6_c2); assert("Client-2 still connected", rc == 1, "isconnected is %d", rc); @@ -1136,7 +1152,7 @@ int test6(struct Options options) rc = MQTTClient_isConnected(test6_c1); assert("Client-1 not connected", rc == 0, "isconnected is %d", rc); - rc = MQTTClient_disconnect(test6_c2, 100L); + rc = MQTTClient_disconnect5(test6_c2, 100L, SUCCESS, NULL); assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); MQTTClient_destroy(&test6_c1); @@ -1157,6 +1173,7 @@ int test6a(struct Options options) MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer; MQTTClient_connectOptions opts2 = MQTTClient_connectOptions_initializer; int rc, count; + MQTTResponse response = {SUCCESS, NULL}; char* mqttsas_topic = "MQTTSAS topic"; failures = 0; @@ -1166,7 +1183,7 @@ int test6a(struct Options options) opts.keepAliveInterval = 2; opts.cleansession = 1; - opts.MQTTVersion = MQTTVERSION_3_1_1; + opts.MQTTVersion = options.MQTTVersion; opts.will = &wopts; opts.will->payload.data = test6_will_message; opts.will->payload.len = strlen(test6_will_message) + 1; @@ -1191,9 +1208,10 @@ int test6a(struct Options options) goto exit; /* Connect to the broker */ - rc = MQTTClient_connect(test6_c1, &opts); - assert("good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); - if (rc != MQTTCLIENT_SUCCESS) + response = MQTTClient_connect5(test6_c1, &opts, NULL, NULL); + assert("good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, + "rc was %d\n", response.reasonCode); + if (response.reasonCode != MQTTCLIENT_SUCCESS) goto exit; /* Client - 2 (multi-threaded) */ @@ -1208,15 +1226,15 @@ int test6a(struct Options options) opts2.keepAliveInterval = 20; opts2.cleansession = 1; MyLog(LOGA_INFO, "Connecting Client_2 ..."); - rc = MQTTClient_connect(test6_c2, &opts2); - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_connect5(test6_c2, &opts2, NULL, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); - rc = MQTTClient_subscribe(test6_c2, test6_will_topic, 2); - assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_subscribe5(test6_c2, test6_will_topic, 2, NULL, NULL); + assert("Good rc from subscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); /* now send the command which will break the connection and cause the will message to be sent */ - rc = MQTTClient_publish(test6_c1, mqttsas_topic, (int)strlen("TERMINATE"), "TERMINATE", 0, 0, NULL); - assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + response = MQTTClient_publish5(test6_c1, mqttsas_topic, (int)strlen("TERMINATE"), "TERMINATE", 0, 0, NULL, NULL); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); MyLog(LOGA_INFO, "Waiting to receive the will message"); count = 0; @@ -1235,8 +1253,8 @@ int test6a(struct Options options) assert("connection lost called", test6_connection_lost_called == 1, "connection_lost_called %d\n", test6_connection_lost_called); - rc = MQTTClient_unsubscribe(test6_c2, test6_will_topic); - assert("Good rc from unsubscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + response = MQTTClient_unsubscribe5(test6_c2, test6_will_topic, NULL); + assert("Good rc from unsubscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); rc = MQTTClient_isConnected(test6_c2); assert("Client-2 still connected", rc == 1, "isconnected is %d", rc); @@ -1244,7 +1262,7 @@ int test6a(struct Options options) rc = MQTTClient_isConnected(test6_c1); assert("Client-1 not connected", rc == 0, "isconnected is %d", rc); - rc = MQTTClient_disconnect(test6_c2, 100L); + rc = MQTTClient_disconnect5(test6_c2, 100L, SUCCESS, NULL); assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); MQTTClient_destroy(&test6_c1); @@ -1267,8 +1285,7 @@ int main(int argc, char** argv) fprintf(xml, "\n", (int)(ARRAY_SIZE(tests) - 1)); setenv("MQTT_C_CLIENT_TRACE", "ON", 1); - setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0); - //setenv("MQTT_C_CLIENT_TRACE_LEVEL", "PROTOCOL", 0); + setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1); getopts(argc, argv); diff --git a/test/test5.c b/test/test5.c index cb5dbeec..6531d143 100644 --- a/test/test5.c +++ b/test/test5.c @@ -2056,8 +2056,7 @@ int test7(struct Options options) { char* testname = "test7"; int subsqos = 2; - AsyncTestClient tc = - AsyncTestClient_initializer; + AsyncTestClient tc = AsyncTestClient_initializer; MQTTAsync c; MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer; MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; diff --git a/test/test6.c b/test/test6.c index 9c701287..a23e9ff7 100644 --- a/test/test6.c +++ b/test/test6.c @@ -61,7 +61,7 @@ struct int persistence; } opts = { - "tcp://localhost:1885", + "tcp://localhost:1884", NULL, 0, "tcp://localhost:7777", @@ -297,22 +297,24 @@ void control_connectionLost(void* context, char* cause) */ int control_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* m) { - MyLog(LOGA_ALWAYS, "Control message arrived: %.*s %s", + MyLog(LOGA_ALWAYS, "Control message arrived: %.*s wait message: %s", m->payloadlen, m->payload, (wait_message == NULL) ? "None" : wait_message); if (strncmp(m->payload, "stop", 4) == 0) { MyLog(LOGA_ALWAYS, "Stop message arrived, stopping..."); - stopping = 1; + stopping = 1; } else if (wait_message != NULL && strncmp(wait_message, m->payload, strlen(wait_message)) == 0) { + MyLog(LOGA_ALWAYS, "Wait message %s found", wait_message); control_found = 1; wait_message = NULL; } else if (wait_message2 != NULL && strncmp(wait_message2, m->payload, strlen(wait_message2)) == 0) { + MyLog(LOGA_ALWAYS, "Wait message2 %s found", wait_message); control_found = 2; wait_message2 = NULL; } @@ -351,7 +353,7 @@ int control_wait(char* message) sprintf(buf, "waiting for: %s", message); control_send(buf); - MyLog(LOGA_ALWAYS, "waiting for: %s", message); + MyLog(LOGA_ALWAYS, "Waiting for: %s", message); while (control_found == 0 && stopping == 0) { if (++count == 300) @@ -362,6 +364,7 @@ int control_wait(char* message) } MySleep(1000); } + MyLog(LOGA_ALWAYS, "Control message found: %s, control_found %d", message, control_found); return control_found; } @@ -377,7 +380,7 @@ int control_which(char* message1, char* message2) while (control_found == 0) { if (++count == 300) - return 0; /* time out and tell the caller the message was not found */ + break; /* time out and tell the caller the message was not found */ MySleep(1000); } return control_found; @@ -750,7 +753,7 @@ void client_onSubscribe(void* context, MQTTAsync_successData* response) void client_onFailure(void* context, MQTTAsync_failureData* response) { MQTTAsync c = (MQTTAsync)context; - MyLog(LOGA_DEBUG, "In failure callback"); + MyLog(LOGA_INFO, "In failure callback"); client_subscribed = -1; }