MQTT 5.0 connect and connack #417

Makefile

@@ -90,7 +90,7 @@ SYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_C}}
 SAMPLE_FILES_A = paho_c_pub paho_c_sub MQTTAsync_subscribe MQTTAsync_publish
 ASYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_A}}
 
-TEST_FILES_C = test1 test2 sync_client_test test_mqtt4sync
+TEST_FILES_C = test1 test15 test2 sync_client_test test_mqtt4sync
 SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}}
 
 TEST_FILES_CS = test3

src/CMakeLists.txt

@@ -47,8 +47,7 @@ SET(common_src
     SocketBuffer.c
     Heap.c
     LinkedList.c
-    MQTTV5Packet.c
-    MQTTV5Properties.c
+    MQTTProperties.c
     )
 
 IF (WIN32)

src/MQTTAsync.c

@@ -34,6 +34,7 @@
  *    Ian Craggs - auto reconnect timing fix #218
  *    Ian Craggs - fix for issue #190
  *    Ian Craggs - check for NULL SSL options #334
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 /**
@@ -1237,9 +1238,11 @@ static int MQTTAsync_processCommand(void)
 
 			Log(TRACE_MIN, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, command->command.details.conn.MQTTVersion);
 #if defined(OPENSSL)
-			rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->ssl, command->command.details.conn.MQTTVersion);
+			rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->ssl, command->command.details.conn.MQTTVersion,
+					NULL, NULL);
 #else
-			rc = MQTTProtocol_connect(serverURI, command->client->c, command->command.details.conn.MQTTVersion);
+			rc = MQTTProtocol_connect(serverURI, command->client->c, command->command.details.conn.MQTTVersion,
+					NULL, NULL);
 #endif
 			if (command->client->c->connect_state == 0)
 				rc = SOCKET_ERROR;
@@ -2921,7 +2924,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
 				{
 					rc = MQTTCLIENT_SUCCESS;
 					m->c->connect_state = 3;
-					if (MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion) == SOCKET_ERROR)
+					if (MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL) == SOCKET_ERROR)
 					{
 						rc = SOCKET_ERROR;
 						goto exit;
@@ -2940,7 +2943,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
 		{
 #endif
 			m->c->connect_state = 3; /* TCP/SSL connect completed, in which case send the MQTT connect packet */
-			if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion)) == SOCKET_ERROR)
+			if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL)) == SOCKET_ERROR)
 				goto exit;
 #if defined(OPENSSL)
 		}
@@ -2956,7 +2959,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
 		if(!m->c->cleansession && m->c->session == NULL)
 			m->c->session = SSL_get1_session(m->c->net.ssl);
 		m->c->connect_state = 3; /* SSL connect completed, in which case send the MQTT connect packet */
-		if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion)) == SOCKET_ERROR)
+		if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL)) == SOCKET_ERROR)
 			goto exit;
 	}
 #endif

src/MQTTClient.c

@@ -33,6 +33,7 @@
  *    Ian Craggs - binary will message support
  *    Ian Craggs - waitforCompletion fix #240
  *    Ian Craggs - check for NULL SSL options #334
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 /**
@@ -284,11 +285,13 @@ static thread_return_type WINAPI MQTTClient_run(void* n);
 static void MQTTClient_stop(void);
 static void MQTTClient_closeSession(Clients* client);
 static int MQTTClient_cleanSession(Clients* client);
-static int MQTTClient_connectURIVersion(
+static MQTTResponse MQTTClient_connectURIVersion(
 	MQTTClient handle, MQTTClient_connectOptions* options,
 	const char* serverURI, int MQTTVersion,
-	START_TIME_TYPE start, long millisecsTimeout);
-static int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI);
+	START_TIME_TYPE start, long millisecsTimeout,
+	MQTTProperties* connectProperties, MQTTProperties* willProperties);
+static MQTTResponse MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI,
+	MQTTProperties* connectProperties, MQTTProperties* willProperties);
 static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int internal, int stop);
 static int MQTTClient_disconnect_internal(MQTTClient handle, int timeout);
 static void MQTTClient_retry(void);
@@ -848,12 +851,13 @@ void Protocol_processPublication(Publish* publish, Clients* client)
 }
 
 
-static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI, int MQTTVersion,
-	START_TIME_TYPE start, long millisecsTimeout)
+static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI, int MQTTVersion,
+	START_TIME_TYPE start, long millisecsTimeout, MQTTProperties* connectProperties, MQTTProperties* willProperties)
 {
 	MQTTClients* m = handle;
 	int rc = SOCKET_ERROR;
 	int sessionPresent = 0;
+	MQTTResponse resp = {SOCKET_ERROR, NULL};
 
 	FUNC_ENTRY;
 	if (m->ma && !running)
@@ -869,9 +873,9 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt
 
 	Log(TRACE_MIN, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, MQTTVersion);
 #if defined(OPENSSL)
-	rc = MQTTProtocol_connect(serverURI, m->c, m->ssl, MQTTVersion);
+	rc = MQTTProtocol_connect(serverURI, m->c, m->ssl, MQTTVersion, connectProperties, willProperties);
 #else
-	rc = MQTTProtocol_connect(serverURI, m->c, MQTTVersion);
+	rc = MQTTProtocol_connect(serverURI, m->c, MQTTVersion, connectProperties, willProperties);
 #endif
 	if (rc == SOCKET_ERROR)
 		goto exit;
@@ -923,7 +927,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt
 				{
 					rc = MQTTCLIENT_SUCCESS;
 					m->c->connect_state = 3;
-					if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR)
+					if (MQTTPacket_send_connect(m->c, MQTTVersion, connectProperties, willProperties) == SOCKET_ERROR)
 					{
 						rc = SOCKET_ERROR;
 						goto exit;
@@ -942,7 +946,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt
 		{
 #endif
 			m->c->connect_state = 3; /* TCP connect completed, in which case send the MQTT connect packet */
-			if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR)
+			if (MQTTPacket_send_connect(m->c, MQTTVersion, connectProperties, willProperties) == SOCKET_ERROR)
 			{
 				rc = SOCKET_ERROR;
 				goto exit;
@@ -966,7 +970,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt
 		if(!m->c->cleansession && m->c->session == NULL)
 			m->c->session = SSL_get1_session(m->c->net.ssl);
 		m->c->connect_state = 3; /* TCP connect completed, in which case send the MQTT connect packet */
-		if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR)
+		if (MQTTPacket_send_connect(m->c, MQTTVersion, connectProperties, willProperties) == SOCKET_ERROR)
 		{
 			rc = SOCKET_ERROR;
 			goto exit;
@@ -1009,6 +1013,8 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt
 					if (m->c->connected != 1)
 						rc = MQTTCLIENT_DISCONNECTED;
 				}
+				if (m->c->MQTTVersion == MQTTVERSION_5)
+					resp.properties = &connack->properties;
 			}
 			free(connack);
 			m->pack = NULL;
@@ -1026,8 +1032,10 @@ exit:
 	}
 	else
 		MQTTClient_disconnect1(handle, 0, 0, (MQTTVersion == 3)); /* don't want to call connection lost */
-	FUNC_EXIT_RC(rc);
-  return rc;
+
+	resp.reasonCode = rc;
+	FUNC_EXIT_RC(resp.reasonCode);
+	return resp;
 }
 
 static int retryLoopInterval = 5;
@@ -1045,12 +1053,13 @@ static void setRetryLoopInterval(int keepalive)
 }
 
 
-static int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI)
+static MQTTResponse MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties)
 {
 	MQTTClients* m = handle;
 	START_TIME_TYPE start;
 	long millisecsTimeout = 30000L;
-	int rc = SOCKET_ERROR;
+	MQTTResponse rc = {SOCKET_ERROR, NULL};
 	int MQTTVersion = 0;
 
 	FUNC_ENTRY;
@@ -1061,6 +1070,7 @@ static int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* o
 	setRetryLoopInterval(options->keepAliveInterval);
 	m->c->cleansession = options->cleansession;
 	m->c->maxInflightMessages = (options->reliable) ? 1 : 10;
+	m->c->MQTTVersion = options->MQTTVersion;
 
 	if (m->c->will)
 	{
@@ -1167,21 +1177,36 @@ static int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* o
 
 	if (MQTTVersion == MQTTVERSION_DEFAULT)
 	{
-		if ((rc = MQTTClient_connectURIVersion(handle, options, serverURI, 4, start, millisecsTimeout)) != MQTTCLIENT_SUCCESS)
-			rc = MQTTClient_connectURIVersion(handle, options, serverURI, 3, start, millisecsTimeout);
+		rc = MQTTClient_connectURIVersion(handle, options, serverURI, 4, start, millisecsTimeout,
+				connectProperties, willProperties);
+		if (rc.reasonCode != MQTTCLIENT_SUCCESS)
+		{
+			rc = MQTTClient_connectURIVersion(handle, options, serverURI, 3, start, millisecsTimeout,
+					connectProperties, willProperties);
+		}
 	}
 	else
-		rc = MQTTClient_connectURIVersion(handle, options, serverURI, MQTTVersion, start, millisecsTimeout);
+		rc = MQTTClient_connectURIVersion(handle, options, serverURI, MQTTVersion, start, millisecsTimeout,
+				connectProperties, willProperties);
 
-	FUNC_EXIT_RC(rc);
+	FUNC_EXIT_RC(rc.reasonCode);
 	return rc;
 }
 
 
 int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
+{
+	MQTTResponse response = MQTTClient_connect5(handle, options, NULL, NULL);
+
+	return response.reasonCode;
+}
+
+
+MQTTResponse MQTTClient_connect5(MQTTClient handle, MQTTClient_connectOptions* options,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties)
 {
 	MQTTClients* m = handle;
-	int rc = SOCKET_ERROR;
+	MQTTResponse rc = {SOCKET_ERROR, NULL};
 
 	FUNC_ENTRY;
 	Thread_lock_mutex(connect_mutex);
@@ -1189,20 +1214,20 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
 
 	if (options == NULL)
 	{
-		rc = MQTTCLIENT_NULL_PARAMETER;
+		rc.reasonCode = MQTTCLIENT_NULL_PARAMETER;
 		goto exit;
 	}
 
 	if (strncmp(options->struct_id, "MQTC", 4) != 0 || 	options->struct_version < 0 || options->struct_version > 5)
 	{
-		rc = MQTTCLIENT_BAD_STRUCTURE;
+		rc.reasonCode = MQTTCLIENT_BAD_STRUCTURE;
 		goto exit;
 	}
 
 #if defined(OPENSSL)
 	if (m->ssl && options->ssl == NULL)
 	{
-		rc = MQTTCLIENT_NULL_PARAMETER;
+		rc.reasonCode = MQTTCLIENT_NULL_PARAMETER;
 		goto exit;
 	}
 #endif
@@ -1211,7 +1236,7 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
 	{
 		if (strncmp(options->will->struct_id, "MQTW", 4) != 0 || (options->will->struct_version != 0 && options->will->struct_version != 1))
 		{
-			rc = MQTTCLIENT_BAD_STRUCTURE;
+			rc.reasonCode = MQTTCLIENT_BAD_STRUCTURE;
 			goto exit;
 		}
 	}
@@ -1222,7 +1247,7 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
 	{
 		if (strncmp(options->ssl->struct_id, "MQTS", 4) != 0 || options->ssl->struct_version < 0 || options->ssl->struct_version > 2)
 		{
-			rc = MQTTCLIENT_BAD_STRUCTURE;
+			rc.reasonCode = MQTTCLIENT_BAD_STRUCTURE;
 			goto exit;
 		}
 	}
@@ -1231,19 +1256,19 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
 	if ((options->username && !UTF8_validateString(options->username)) ||
 		(options->password && !UTF8_validateString(options->password)))
 	{
-		rc = MQTTCLIENT_BAD_UTF8_STRING;
+		rc.reasonCode = MQTTCLIENT_BAD_UTF8_STRING;
 		goto exit;
 	}
 
 	if (options->MQTTVersion != MQTTVERSION_DEFAULT &&
 			(options->MQTTVersion < MQTTVERSION_3_1 || options->MQTTVersion > MQTTVERSION_5))
 	{
-		rc = MQTTCLIENT_BAD_MQTT_VERSION;
+		rc.reasonCode = MQTTCLIENT_BAD_MQTT_VERSION;
 		goto exit;
 	}
 
 	if (options->struct_version < 2 || options->serverURIcount == 0)
-		rc = MQTTClient_connectURI(handle, options, m->serverURI);
+		rc = MQTTClient_connectURI(handle, options, m->serverURI, connectProperties, willProperties);
 	else
 	{
 		int i;
@@ -1261,7 +1286,8 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
 				m->ssl = 1;
 			}
 #endif
-			if ((rc = MQTTClient_connectURI(handle, options, serverURI)) == MQTTCLIENT_SUCCESS)
+			rc = MQTTClient_connectURI(handle, options, serverURI, connectProperties, willProperties);
+			if (rc.reasonCode == MQTTCLIENT_SUCCESS)
 				break;
 		}
 	}
@@ -1278,7 +1304,7 @@ exit:
 	}
 	Thread_unlock_mutex(mqttclient_mutex);
 	Thread_unlock_mutex(connect_mutex);
-	FUNC_EXIT_RC(rc);
+	FUNC_EXIT_RC(rc.reasonCode);
 	return rc;
 }
 

src/MQTTClient.h

@@ -122,6 +122,8 @@
 /// @endcond
 */
 
+#include "MQTTProperties.h"
+#include "MQTTReasonCodes.h"
 #if !defined(NO_PERSISTENCE)
 #include "MQTTClientPersistence.h"
 #endif
@@ -753,6 +755,16 @@ DLLExport MQTTClient_nameValue* MQTTClient_getVersionInfo(void);
   */
 DLLExport int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options);
 
+
+typedef struct MQTTResponse
+{
+	enum MQTTReasonCodes reasonCode;
+	MQTTProperties* properties; /* optional */
+} MQTTResponse;
+
+DLLExport MQTTResponse MQTTClient_connect5(MQTTClient handle, MQTTClient_connectOptions* options,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties);
+
 /**
   * This function attempts to disconnect the client from the MQTT
   * server. In order to allow the client time to complete handling of messages

src/MQTTPacket.c

 /*******************************************************************************
- * Copyright (c) 2009, 2014 IBM Corp.
+ * Copyright (c) 2009, 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
@@ -14,6 +14,7 @@
  *    Ian Craggs - initial API and implementation and/or initial documentation
  *    Ian Craggs, Allan Stockdill-Mander - SSL updates
  *    Ian Craggs - MQTT 3.1.1 support
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 /**
@@ -143,7 +144,7 @@ void* MQTTPacket_Factory(networkHandles* net, int* error)
 		else
 		{
 			if ((pack = (*new_packets[ptype])(header.byte, data, remaining_length)) == NULL)
-				*error = BAD_MQTT_PACKET;
+				*error = SOCKET_ERROR; // was BAD_MQTT_PACKET;
 #if !defined(NO_PERSISTENCE)
 			else if (header.bits.type == PUBLISH && header.bits.qos == 2)
 			{
@@ -753,3 +754,142 @@ void MQTTPacket_free_packet(MQTTPacket* pack)
 		free(pack);
 	FUNC_EXIT;
 }
+
+
+/**
+ * Writes an integer as 4 bytes to an output buffer.
+ * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
+ * @param anInt the integer to write
+ */
+void writeInt4(char** pptr, int anInt)
+{
+  **pptr = (char)(anInt / 16777216);
+  (*pptr)++;
+  anInt %= 16777216;
+  **pptr = (char)(anInt / 65536);
+  (*pptr)++;
+  anInt %= 65536;
+	**pptr = (char)(anInt / 256);
+	(*pptr)++;
+	**pptr = (char)(anInt % 256);
+	(*pptr)++;
+}
+
+
+/**
+ * Calculates an integer from two bytes read from the input buffer
+ * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
+ * @return the integer value calculated
+ */
+int readInt4(char** pptr)
+{
+	unsigned char* ptr = (unsigned char*)*pptr;
+	int value = 16777216*(*ptr) + 65536*(*(ptr+1)) + 256*(*(ptr+2)) + (*(ptr+3));
+	*pptr += 4;
+	return value;
+}
+
+
+void writeMQTTLenString(char** pptr, MQTTLenString lenstring)
+{
+  writeInt(pptr, lenstring.len);
+  memcpy(*pptr, lenstring.data, lenstring.len);
+  *pptr += lenstring.len;
+}
+
+
+int MQTTLenStringRead(MQTTLenString* lenstring, char** pptr, char* enddata)
+{
+	int len = 0;
+
+	/* the first two bytes are the length of the string */
+	if (enddata - (*pptr) > 1) /* enough length to read the integer? */
+	{
+		lenstring->len = readInt(pptr); /* increments pptr to point past length */
+		if (&(*pptr)[lenstring->len] <= enddata)
+		{
+			lenstring->data = (char*)*pptr;
+			*pptr += lenstring->len;
+			len = 2 + lenstring->len;
+		}
+	}
+	return len;
+}
+
+/*
+if (prop->value.integer4 >= 0 && prop->value.integer4 <= 127)
+  len = 1;
+else if (prop->value.integer4 >= 128 && prop->value.integer4 <= 16383)
+  len = 2;
+else if (prop->value.integer4 >= 16384 && prop->value.integer4 < 2097151)
+  len = 3;
+else if (prop->value.integer4 >= 2097152 && prop->value.integer4 < 268435455)
+  len = 4;
+*/
+int MQTTPacket_VBIlen(int rem_len)
+{
+	int rc = 0;
+
+	if (rem_len < 128)
+		rc = 1;
+	else if (rem_len < 16384)
+		rc = 2;
+	else if (rem_len < 2097152)
+		rc = 3;
+	else
+		rc = 4;
+  return rc;
+}
+
+
+/**
+ * Decodes the message length according to the MQTT algorithm
+ * @param getcharfn pointer to function to read the next character from the data source
+ * @param value the decoded length returned
+ * @return the number of bytes read from the socket
+ */
+int MQTTPacket_VBIdecode(int (*getcharfn)(char*, int), int* value)
+{
+	char c;
+	int multiplier = 1;
+	int len = 0;
+#define MAX_NO_OF_REMAINING_LENGTH_BYTES 4
+
+	*value = 0;
+	do
+	{
+		int rc = MQTTPACKET_READ_ERROR;
+
+		if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+		{
+			rc = MQTTPACKET_READ_ERROR;	/* bad data */
+			goto exit;
+		}
+		rc = (*getcharfn)(&c, 1);
+		if (rc != 1)
+			goto exit;
+		*value += (c & 127) * multiplier;
+		multiplier *= 128;
+	} while ((c & 128) != 0);
+exit:
+	return len;
+}
+
+
+static char* bufptr;
+
+int bufchar(char* c, int count)
+{
+	int i;
+
+	for (i = 0; i < count; ++i)
+		*c = *bufptr++;
+	return count;
+}
+
+
+int MQTTPacket_decodeBuf(char* buf, int* value)
+{
+	bufptr = buf;
+	return MQTTPacket_VBIdecode(bufchar, value);
+}

src/MQTTPacket.h

@@ -28,16 +28,19 @@
 #include "LinkedList.h"
 #include "Clients.h"
 
-/*BE
-include "Socket"
-include "LinkedList"
-include "Clients"
-BE*/
-
 typedef unsigned int bool;
 typedef void* (*pf)(unsigned char, char*, size_t);
 
-#define BAD_MQTT_PACKET -4
+#include "MQTTProperties.h"
+
+enum errors
+{
+	MQTTPACKET_BAD = -4,
+	MQTTPACKET_BUFFER_TOO_SHORT = -2,
+	MQTTPACKET_READ_ERROR = -1,
+	MQTTPACKET_READ_COMPLETE
+};
+
 
 enum msgTypes
 {
@@ -146,7 +149,8 @@ typedef struct
 		} bits;
 #endif
 	} flags;	 /**< connack flags byte */
-	char rc; /**< connack return code */
+	char rc; /**< connack reason code */
+	MQTTProperties properties; /**< MQTT 5.0 properties.  Not used for MQTT < 5.0 */
 } Connack;
 
 
@@ -256,7 +260,13 @@ int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID
 
 void MQTTPacket_free_packet(MQTTPacket* pack);
 
+void writeInt4(char** pptr, int anInt);
+int readInt4(char** pptr);
+void writeMQTTLenString(char** pptr, MQTTLenString lenstring);
+int MQTTLenStringRead(MQTTLenString* lenstring, char** pptr, char* enddata);
+int MQTTPacket_VBIlen(int rem_len);
+int MQTTPacket_decodeBuf(char* buf, int* value);
+
 #include "MQTTPacketOut.h"
-#include "MQTTV5Packet.h"
 
 #endif /* MQTTPACKET_H */

src/MQTTPacketOut.c

@@ -16,6 +16,7 @@
  *    Ian Craggs - MQTT 3.1.1 support
  *    Rong Xiang, Ian Craggs - C++ compatibility
  *    Ian Craggs - binary password and will payload
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 /**
@@ -37,19 +38,20 @@
 
 
 /**
- * Send an MQTT CONNECT packet down a socket.
+ * Send an MQTT CONNECT packet down a socket for V5 or later
  * @param client a structure from which to get all the required values
  * @param MQTTVersion the MQTT version to connect with
+ * @param connectProperties MQTT V5 properties for the connect packet
+ * @param willProperties MQTT V5 properties for the will message, if any
  * @return the completion code (e.g. TCPSOCKET_COMPLETE)
  */
-int MQTTPacket_send_connect(Clients* client, int MQTTVersion)
+int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
+	               MQTTProperties* connectProperties, MQTTProperties* willProperties)
 {
 	char *buf, *ptr;
 	Connect packet;
 	int rc = -1, len;
 
-	MQTTProperties connectProperties = MQTTProperties_initializer;
-
 	FUNC_ENTRY;
 	packet.header.byte = 0;
 	packet.header.bits.type = CONNECT;
@@ -63,10 +65,10 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion)
 		len += client->passwordlen+2;
 	if (MQTTVersion >= 5)
 	{
-      //if (connectProperties)
-	    len += MQTTProperties_len(&connectProperties);
-	  /*if (client->will && willProperties)
-		  len += MQTTProperties_len(willProperties);*/
+		if (connectProperties)
+			len += MQTTProperties_len(connectProperties);
+		if (client->will && willProperties)
+			len += MQTTProperties_len(willProperties);
 	}
 
 	ptr = buf = malloc(len);
@@ -91,7 +93,6 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion)
 		packet.flags.bits.willQoS = client->will->qos;
 		packet.flags.bits.willRetain = client->will->retained;
 	}
-
 	if (client->username)
 		packet.flags.bits.username = 1;
 	if (client->password)
@@ -100,10 +101,12 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion)
 	writeChar(&ptr, packet.flags.all);
 	writeInt(&ptr, client->keepAliveInterval);
 	if (MQTTVersion == 5)
-	  MQTTProperties_write(&ptr, &connectProperties);
+		MQTTProperties_write(&ptr, connectProperties);
 	writeUTF(&ptr, client->clientID);
 	if (client->will)
 	{
+		if (MQTTVersion == 5)
+			MQTTProperties_write(&ptr, willProperties);
 		writeUTF(&ptr, client->will->topic);
 		writeData(&ptr, client->will->payload, client->will->payloadlen);
 	}
@@ -136,8 +139,17 @@ void* MQTTPacket_connack(unsigned char aHeader, char* data, size_t datalen)
 
 	FUNC_ENTRY;
 	pack->header.byte = aHeader;
-	pack->flags.all = readChar(&curdata);
-	pack->rc = readChar(&curdata);
+	pack->flags.all = readChar(&curdata); /* connect flags */
+	pack->rc = readChar(&curdata); /* reason code */
+	if (datalen > 2)
+	{
+		MQTTProperties props = MQTTProperties_initializer;
+		pack->properties = props;
+		pack->properties.max_count = 10;
+		pack->properties.array = malloc(sizeof(MQTTProperty) * pack->properties.max_count);
+		if (MQTTProperties_read(&pack->properties, &curdata, curdata + datalen) != 1)
+			pack = NULL; /* signal protocol error */
+	}
 	FUNC_EXIT;
 	return pack;
 }

src/MQTTPacketOut.h

@@ -14,6 +14,7 @@
  *    Ian Craggs - initial API and implementation and/or initial documentation
  *    Ian Craggs, Allan Stockdill-Mander - SSL updates
  *    Ian Craggs - MQTT 3.1.1 support
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 #if !defined(MQTTPACKETOUT_H)
@@ -21,7 +22,8 @@
 
 #include "MQTTPacket.h"
 
-int MQTTPacket_send_connect(Clients* client, int MQTTVersion);
+int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties);
 void* MQTTPacket_connack(unsigned char aHeader, char* data, size_t datalen);
 
 int MQTTPacket_send_pingreq(networkHandles* net, const char* clientID);

src/MQTTV5Properties.c → src/MQTTProperties.c

@@ -14,8 +14,10 @@
  *    Ian Craggs - initial API and implementation and/or initial documentation
  *******************************************************************************/
 
-#include "MQTTV5Properties.h"
+#include "MQTTProperties.h"
+
 #include "MQTTPacket.h"
+#include "Heap.h"
 
 #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
 
@@ -244,8 +246,13 @@ int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata)
   {
     *pptr += MQTTPacket_decodeBuf(*pptr, &remlength);
     properties->length = remlength;
-    while (properties->count < properties->max_count && remlength > 0)
+    while (remlength > 0)
+    {
+    	  if (properties->count == properties->max_count)
     	  {
+    		  properties->max_count += 10;
+    		  properties->array = realloc(properties->array, sizeof(MQTTProperty) * properties->max_count);
+    	  }
       remlength -= MQTTProperty_read(&properties->array[properties->count], pptr, enddata);
       properties->count++;
     }
@@ -255,3 +262,74 @@ int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata)
 
   return rc;
 }
+
+struct {
+	enum PropertyNames value;
+	const char* name;
+} nameToString[] =
+{
+  {PAYLOAD_FORMAT_INDICATOR, "PAYLOAD_FORMAT_INDICATOR"},
+  {MESSAGE_EXPIRY_INTERVAL, "MESSAGE_EXPIRY_INTERVAL"},
+  {CONTENT_TYPE, "CONTENT_TYPE"},
+  {RESPONSE_TOPIC, "RESPONSE_TOPIC"},
+  {CORRELATION_DATA, "CORRELATION_DATA"},
+  {SUBSCRIPTION_IDENTIFIER, "SUBSCRIPTION_IDENTIFIER"},
+  {SESSION_EXPIRY_INTERVAL, "SESSION_EXPIRY_INTERVAL"},
+  {ASSIGNED_CLIENT_IDENTIFER, "ASSIGNED_CLIENT_IDENTIFER"},
+  {SERVER_KEEP_ALIVE, "SERVER_KEEP_ALIVE"},
+  {AUTHENTICATION_METHOD, "AUTHENTICATION_METHOD"},
+  {AUTHENTICATION_DATA, "AUTHENTICATION_DATA"},
+  {REQUEST_PROBLEM_INFORMATION, "REQUEST_PROBLEM_INFORMATION"},
+  {WILL_DELAY_INTERVAL, "WILL_DELAY_INTERVAL"},
+  {REQUEST_RESPONSE_INFORMATION, "REQUEST_RESPONSE_INFORMATION"},
+  {RESPONSE_INFORMATION, "RESPONSE_INFORMATION"},
+  {SERVER_REFERENCE, "SERVER_REFERENCE"},
+  {REASON_STRING, "REASON_STRING"},
+  {RECEIVE_MAXIMUM, "RECEIVE_MAXIMUM"},
+  {TOPIC_ALIAS_MAXIMUM, "TOPIC_ALIAS_MAXIMUM"},
+  {TOPIC_ALIAS, "TOPIC_ALIAS"},
+  {MAXIMUM_QOS, "MAXIMUM_QOS"},
+  {RETAIN_AVAILABLE, "RETAIN_AVAILABLE"},
+  {USER_PROPERTY, "USER_PROPERTY"},
+  {MAXIMUM_PACKET_SIZE, "MAXIMUM_PACKET_SIZE"},
+  {WILDCARD_SUBSCRIPTION_AVAILABLE, "WILDCARD_SUBSCRIPTION_AVAILABLE"},
+  {SUBSCRIPTION_IDENTIFIER_AVAILABLE, "SUBSCRIPTION_IDENTIFIER_AVAILABLE"},
+  {SHARED_SUBSCRIPTION_AVAILABLE, "SHARED_SUBSCRIPTION_AVAILABLE"}
+};
+
+const char* MQTTPropertyName(enum PropertyNames value)
+{
+  int i = 0;
+  const char* result = NULL;
+
+  for (i = 0; i < ARRAY_SIZE(nameToString); ++i)
+  {
+    if (nameToString[i].value == value)
+    {
+    	  result = nameToString[i].name;
+    	  break;
+    }
+  }
+  return result;
+}
+
+
+DLLExport void MQTTProperties_free(MQTTProperties* props)
+{
+  int i = 0;
+
+  for (i = 0; i < props->count; ++i)
+  {
+    int id = props->array[i].identifier;
+
+    switch (MQTTProperty_getType(id))
+    {
+    case BINARY_DATA:
+    	case UTF_8_ENCODED_STRING:
+    	  break;
+    	case UTF_8_STRING_PAIR:
+	  break;
+    }
+  }
+  free(props->array);
+}

src/MQTTV5Properties.h → src/MQTTProperties.h

@@ -53,6 +53,16 @@ enum PropertyNames {
   SHARED_SUBSCRIPTION_AVAILABLE = 42
 };
 
+#if defined(WIN32) || defined(WIN64)
+  #define DLLImport __declspec(dllimport)
+  #define DLLExport __declspec(dllexport)
+#else
+  #define DLLImport extern
+  #define DLLExport __attribute__ ((visibility ("default")))
+#endif
+
+DLLExport const char* MQTTPropertyName(enum PropertyNames);
+
 enum PropertyTypes {
   BYTE,
   TWO_BYTE_INTEGER,
@@ -63,6 +73,8 @@ enum PropertyTypes {
   UTF_8_STRING_PAIR
 };
 
+DLLExport int MQTTProperty_getType(int identifier);
+
 
 typedef struct
 {
@@ -94,10 +106,12 @@ int MQTTProperties_len(MQTTProperties* props);
  * @param prop
  * @return whether the write succeeded or not, number of bytes written or < 0
  */
-int MQTTProperties_add(MQTTProperties* props, MQTTProperty* prop);
+DLLExport int MQTTProperties_add(MQTTProperties* props, MQTTProperty* prop);
 
 int MQTTProperties_write(char** pptr, MQTTProperties* properties);
 
 int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata);
 
+DLLExport void MQTTProperties_free(MQTTProperties* properties);
+
 #endif /* MQTTPROPERTIES_H */

src/MQTTProtocolOut.c

@@ -20,6 +20,7 @@
  *    Ian Craggs - SNI support
  *    Ian Craggs - fix for issue #164
  *    Ian Craggs - fix for issue #179
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 /**
@@ -94,9 +95,11 @@ char* MQTTProtocol_addressPort(const char* uri, int* port)
  * @return return code
  */
 #if defined(OPENSSL)
-int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int ssl, int MQTTVersion)
+int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int ssl, int MQTTVersion,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties)
 #else
-int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersion)
+int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersion,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties)
 #endif
 {
 	int rc, port;
@@ -129,7 +132,7 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersi
 		if (rc == 0)
 		{
 			/* Now send the MQTT connect packet */
-			if ((rc = MQTTPacket_send_connect(aClient, MQTTVersion)) == 0)
+			if ((rc = MQTTPacket_send_connect(aClient, MQTTVersion, connectProperties, willProperties)) == 0)
 				aClient->connect_state = 3; /* MQTT Connect sent - wait for CONNACK */ 
 			else
 				aClient->connect_state = 0;

src/MQTTProtocolOut.h

 /*******************************************************************************
- * Copyright (c) 2009, 2017 IBM Corp.
+ * Copyright (c) 2009, 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
@@ -15,6 +15,7 @@
  *    Ian Craggs, Allan Stockdill-Mander - SSL updates
  *    Ian Craggs - MQTT 3.1.1 support
  *    Ian Craggs - SNI support
+ *    Ian Craggs - MQTT 5.0 support
  *******************************************************************************/
 
 #if !defined(MQTTPROTOCOLOUT_H)
@@ -33,9 +34,11 @@
 char* MQTTProtocol_addressPort(const char* uri, int* port);
 void MQTTProtocol_reconnect(const char* ip_address, Clients* client);
 #if defined(OPENSSL)
-int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int ssl, int MQTTVersion);
+int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int ssl, int MQTTVersion,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties);
 #else
-int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int MQTTVersion);
+int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int MQTTVersion,
+		MQTTProperties* connectProperties, MQTTProperties* willProperties);
 #endif
 int MQTTProtocol_handlePingresps(void* pack, int sock);
 int MQTTProtocol_subscribe(Clients* client, List* topics, List* qoss, int msgID);

src/MQTTV5ReasonCodes.h → src/MQTTReasonCodes.h

@@ -14,7 +14,7 @@
  *    Ian Craggs - initial API and implementation and/or initial documentation
  *******************************************************************************/
 
-enum ReasonCodes {
+enum MQTTReasonCodes {
   SUCCESS = 0,
   NORMAL_DISCONNECTION = 0,
   GRANTED_QOS_0 = 0,

src/MQTTV5Packet.c

-/*******************************************************************************
- * Copyright (c) 2017, 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 API and implementation and/or initial documentation
- *******************************************************************************/
-
-#include "MQTTV5Packet.h"
-#include "MQTTPacket.h"
-
-#include <string.h>
-
-/**
- * Writes an integer as 4 bytes to an output buffer.
- * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
- * @param anInt the integer to write
- */
-void writeInt4(char** pptr, int anInt)
-{
-  **pptr = (char)(anInt / 16777216);
-  (*pptr)++;
-  anInt %= 16777216;
-  **pptr = (char)(anInt / 65536);
-  (*pptr)++;
-  anInt %= 65536;
-	**pptr = (char)(anInt / 256);
-	(*pptr)++;
-	**pptr = (char)(anInt % 256);
-	(*pptr)++;
-}
-
-
-/**
- * Calculates an integer from two bytes read from the input buffer
- * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
- * @return the integer value calculated
- */
-int readInt4(char** pptr)
-{
-	char* ptr = *pptr;
-	int value = 16777216*(*ptr) + 65536*(*(ptr+1)) + 256*(*(ptr+2)) + (*(ptr+3));
-	*pptr += 4;
-	return value;
-}
-
-
-void writeMQTTLenString(char** pptr, MQTTLenString lenstring)
-{
-  writeInt(pptr, lenstring.len);
-  memcpy(*pptr, lenstring.data, lenstring.len);
-  *pptr += lenstring.len;
-}
-
-
-int MQTTLenStringRead(MQTTLenString* lenstring, char** pptr, char* enddata)
-{
-	int len = 0;
-
-	/* the first two bytes are the length of the string */
-	if (enddata - (*pptr) > 1) /* enough length to read the integer? */
-	{
-		lenstring->len = readInt(pptr); /* increments pptr to point past length */
-		if (&(*pptr)[lenstring->len] <= enddata)
-		{
-			lenstring->data = (char*)*pptr;
-			*pptr += lenstring->len;
-			len = 2 + lenstring->len;
-		}
-	}
-	return len;
-}
-
-/*
-if (prop->value.integer4 >= 0 && prop->value.integer4 <= 127)
-  len = 1;
-else if (prop->value.integer4 >= 128 && prop->value.integer4 <= 16383)
-  len = 2;
-else if (prop->value.integer4 >= 16384 && prop->value.integer4 < 2097151)
-  len = 3;
-else if (prop->value.integer4 >= 2097152 && prop->value.integer4 < 268435455)
-  len = 4;
-*/
-int MQTTPacket_VBIlen(int rem_len)
-{
-	int rc = 0;
-
-	if (rem_len < 128)
-		rc = 1;
-	else if (rem_len < 16384)
-		rc = 2;
-	else if (rem_len < 2097152)
-		rc = 3;
-	else
-		rc = 4;
-  return rc;
-}
-
-
-/**
- * Decodes the message length according to the MQTT algorithm
- * @param getcharfn pointer to function to read the next character from the data source
- * @param value the decoded length returned
- * @return the number of bytes read from the socket
- */
-int MQTTPacket_VBIdecode(int (*getcharfn)(char*, int), int* value)
-{
-	char c;
-	int multiplier = 1;
-	int len = 0;
-#define MAX_NO_OF_REMAINING_LENGTH_BYTES 4
-
-	*value = 0;
-	do
-	{
-		int rc = MQTTPACKET_READ_ERROR;
-
-		if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
-		{
-			rc = MQTTPACKET_READ_ERROR;	/* bad data */
-			goto exit;
-		}
-		rc = (*getcharfn)(&c, 1);
-		if (rc != 1)
-			goto exit;
-		*value += (c & 127) * multiplier;
-		multiplier *= 128;
-	} while ((c & 128) != 0);
-exit:
-	return len;
-}
-
-
-static char* bufptr;
-
-int bufchar(char* c, int count)
-{
-	int i;
-
-	for (i = 0; i < count; ++i)
-		*c = *bufptr++;
-	return count;
-}
-
-
-int MQTTPacket_decodeBuf(char* buf, int* value)
-{
-	bufptr = buf;
-	return MQTTPacket_VBIdecode(bufchar, value);
-}

src/MQTTV5Packet.h

-/*******************************************************************************
- * Copyright (c) 2017, 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 API and implementation and/or initial documentation
- *******************************************************************************/
-
-#if !defined(MQTTV5PACKET_H)
-#define MQTTV5PACKET_H
-
-#include "MQTTV5Properties.h"
-
-enum errors
-{
-	MQTTPACKET_BUFFER_TOO_SHORT = -2,
-	MQTTPACKET_READ_ERROR = -1,
-	MQTTPACKET_READ_COMPLETE
-};
-
-void writeInt4(char** pptr, int anInt);
-int readInt4(char** pptr);
-void writeMQTTLenString(char** pptr, MQTTLenString lenstring);
-int MQTTLenStringRead(MQTTLenString* lenstring, char** pptr, char* enddata);
-int MQTTPacket_VBIlen(int rem_len);
-int MQTTPacket_decodeBuf(char* buf, int* value);
-
-#endif /* MQTTV5PACKET_H */

test/CMakeLists.txt

@@ -95,6 +95,25 @@ SET_TESTS_PROPERTIES(
 	PROPERTIES TIMEOUT 540
 )
 
+ADD_EXECUTABLE(
+	test15
+	test15.c
+)
+
+TARGET_LINK_LIBRARIES(
+	test15
+	paho-mqtt3c
+)
+
+ADD_TEST(
+	NAME test15-1-single-thread-client
+	COMMAND "test15" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER}
+)
+
+SET_TESTS_PROPERTIES(
+	test15-1-single-thread-client
+	PROPERTIES TIMEOUT 540
+)
 
 ADD_EXECUTABLE(
 	test2

test/test15.c

+/*******************************************************************************
+ * Copyright (c) 2009, 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 API and implementation and/or initial documentation
+ *    Ian Craggs - MQTT 3.1.1 support
+ *    Ian Craggs - change will message test back to using proxy
+ *    Ian Craggs - MQTT 5.0 support
+ *******************************************************************************/
+
+
+/**
+ * @file
+ * Tests for the MQ Telemetry MQTT C client
+ */
+
+
+/*
+#if !defined(_RTSHEADER)
+	#include <rts.h>
+#endif
+*/
+
+#include "MQTTClient.h"
+#include <string.h>
+#include <stdlib.h>
+
+#if !defined(_WINDOWS)
+  #include <sys/time.h>
+  #include <sys/socket.h>
+  #include <unistd.h>
+  #include <errno.h>
+#else
+  #include <windows.h>
+  #define setenv(a, b, c) _putenv_s(a, b)
+#endif
+
+#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
+
+void usage(void)
+{
+	printf("help!!\n");
+	exit(EXIT_FAILURE);
+}
+
+struct Options
+{
+	char* connection;         /**< connection to system under test. */
+	char** haconnections;
+	char* proxy_connection;
+	int hacount;
+	int verbose;
+	int test_no;
+	int MQTTVersion;
+	int iterations;
+} options =
+{
+	"tcp://iot.eclipse.org:1883",
+	NULL,
+	"tcp://localhost:1883",
+	0,
+	0,
+	0,
+	MQTTVERSION_5,
+	1,
+};
+
+void getopts(int argc, char** argv)
+{
+	int count = 1;
+
+	while (count < argc)
+	{
+		if (strcmp(argv[count], "--test_no") == 0)
+		{
+			if (++count < argc)
+				options.test_no = atoi(argv[count]);
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--connection") == 0)
+		{
+			if (++count < argc)
+			{
+				options.connection = argv[count];
+				printf("\nSetting connection to %s\n", options.connection);
+			}
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--haconnections") == 0)
+		{
+			if (++count < argc)
+			{
+				char* tok = strtok(argv[count], " ");
+				options.hacount = 0;
+				options.haconnections = malloc(sizeof(char*) * 5);
+				while (tok)
+				{
+					options.haconnections[options.hacount] = malloc(strlen(tok) + 1);
+					strcpy(options.haconnections[options.hacount], tok);
+					options.hacount++;
+					tok = strtok(NULL, " ");
+				}
+			}
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--proxy_connection") == 0)
+		{
+			if (++count < argc)
+				options.proxy_connection = argv[count];
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--MQTTversion") == 0)
+		{
+			if (++count < argc)
+			{
+				options.MQTTVersion = atoi(argv[count]);
+				printf("setting MQTT version to %d\n", options.MQTTVersion);
+			}
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--iterations") == 0)
+		{
+			if (++count < argc)
+				options.iterations = atoi(argv[count]);
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--verbose") == 0)
+		{
+			options.verbose = 1;
+			printf("\nSetting verbose on\n");
+		}
+		count++;
+	}
+}
+
+
+#define LOGA_DEBUG 0
+#define LOGA_INFO 1
+#include <stdarg.h>
+#include <time.h>
+#include <sys/timeb.h>
+void MyLog(int LOGA_level, char* format, ...)
+{
+	static char msg_buf[256];
+	va_list args;
+	struct timeb ts;
+
+	struct tm *timeinfo;
+
+	if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
+	  return;
+
+	ftime(&ts);
+	timeinfo = localtime(&ts.time);
+	strftime(msg_buf, 80, "%Y%m%d %H%M%S", timeinfo);
+
+	sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
+
+	va_start(args, format);
+	vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
+	va_end(args);
+
+	printf("%s\n", msg_buf);
+	fflush(stdout);
+}
+
+
+#if defined(WIN32) || defined(_WINDOWS)
+#define mqsleep(A) Sleep(1000*A)
+#define START_TIME_TYPE DWORD
+static DWORD start_time = 0;
+START_TIME_TYPE start_clock(void)
+{
+	return GetTickCount();
+}
+#elif defined(AIX)
+#define mqsleep sleep
+#define START_TIME_TYPE struct timespec
+START_TIME_TYPE start_clock(void)
+{
+	static struct timespec start;
+	clock_gettime(CLOCK_REALTIME, &start);
+	return start;
+}
+#else
+#define mqsleep sleep
+#define START_TIME_TYPE struct timeval
+/* TODO - unused - remove? static struct timeval start_time; */
+START_TIME_TYPE start_clock(void)
+{
+	struct timeval start_time;
+	gettimeofday(&start_time, NULL);
+	return start_time;
+}
+#endif
+
+
+#if defined(WIN32)
+long elapsed(START_TIME_TYPE start_time)
+{
+	return GetTickCount() - start_time;
+}
+#elif defined(AIX)
+#define assert(a)
+long elapsed(struct timespec start)
+{
+	struct timespec now, res;
+
+	clock_gettime(CLOCK_REALTIME, &now);
+	ntimersub(now, start, res);
+	return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
+}
+#else
+long elapsed(START_TIME_TYPE start_time)
+{
+	struct timeval now, res;
+
+	gettimeofday(&now, NULL);
+	timersub(&now, &start_time, &res);
+	return (res.tv_sec)*1000 + (res.tv_usec)/1000;
+}
+#endif
+
+
+#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
+#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
+
+int tests = 0;
+int failures = 0;
+FILE* xml;
+START_TIME_TYPE global_start_time;
+char output[3000];
+char* cur_output = output;
+
+
+void write_test_result(void)
+{
+	long duration = elapsed(global_start_time);
+
+	fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000);
+	if (cur_output != output)
+	{
+		fprintf(xml, "%s", output);
+		cur_output = output;
+	}
+	fprintf(xml, "</testcase>\n");
+}
+
+
+void myassert(char* filename, int lineno, char* description, int value, char* format, ...)
+{
+	++tests;
+	if (!value)
+	{
+		va_list args;
+
+		++failures;
+		MyLog(LOGA_INFO, "Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
+
+		va_start(args, format);
+		vprintf(format, args);
+		va_end(args);
+
+		cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
+                        description, filename, lineno);
+	}
+	else
+		MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
+}
+
+
+/*********************************************************************
+
+Test1: single-threaded client
+
+*********************************************************************/
+void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
+{
+	MQTTClient_deliveryToken dt;
+	MQTTClient_message pubmsg = MQTTClient_message_initializer;
+	MQTTClient_message* m = NULL;
+	char* topicName = NULL;
+	int topicLen;
+	int i = 0;
+	int iterations = 50;
+	int rc;
+
+	MyLog(LOGA_DEBUG, "%d messages at QoS %d", iterations, qos);
+	pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+	pubmsg.payloadlen = 11;
+	pubmsg.qos = qos;
+	pubmsg.retained = 0;
+
+	for (i = 0; i< iterations; ++i)
+	{
+		if (i % 10 == 0)
+			rc = MQTTClient_publish(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &dt);
+		else
+			rc = MQTTClient_publishMessage(c, test_topic, &pubmsg, &dt);
+		assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+		if (qos > 0)
+		{
+			rc = MQTTClient_waitForCompletion(c, dt, 5000L);
+			assert("Good rc from waitforCompletion", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+		}
+
+		rc = MQTTClient_receive(c, &topicName, &topicLen, &m, 5000);
+		assert("Good rc from receive", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+		if (topicName)
+		{
+			MyLog(LOGA_DEBUG, "Message received on topic %s is %.*s", topicName, m->payloadlen, (char*)(m->payload));
+			if (pubmsg.payloadlen != m->payloadlen ||
+					memcmp(m->payload, pubmsg.payload, m->payloadlen) != 0)
+			{
+				failures++;
+				MyLog(LOGA_INFO, "Error: wrong data - received lengths %d %d", pubmsg.payloadlen, m->payloadlen);
+				break;
+			}
+			MQTTClient_free(topicName);
+			MQTTClient_freeMessage(&m);
+		}
+		else
+			printf("No message received within timeout period\n");
+	}
+
+	/* receive any outstanding messages */
+	MQTTClient_receive(c, &topicName, &topicLen, &m, 2000);
+	while (topicName)
+	{
+		printf("Message received on topic %s is %.*s.\n", topicName, m->payloadlen, (char*)(m->payload));
+		MQTTClient_free(topicName);
+		MQTTClient_freeMessage(&m);
+		MQTTClient_receive(c, &topicName, &topicLen, &m, 2000);
+	}
+}
+
+void logProperties(MQTTProperties *props)
+{
+	int i = 0;
+
+	for (i = 0; i < props->count; ++i)
+	{
+		int id = props->array[i].identifier;
+		const char* name = MQTTPropertyName(id);
+		char* intformat = "Property name %s value %d";
+
+		switch (MQTTProperty_getType(id))
+		{
+		case BYTE:
+		  MyLog(LOGA_INFO, intformat, name, props->array[i].value.byte);
+		  break;
+		case TWO_BYTE_INTEGER:
+		  MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer2);
+		  break;
+		case FOUR_BYTE_INTEGER:
+		  MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
+		  break;
+		case VARIABLE_BYTE_INTEGER:
+		  MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
+		  break;
+		case BINARY_DATA:
+		case UTF_8_ENCODED_STRING:
+		  MyLog(LOGA_INFO, "Property name %s value %*.s", name,
+				  props->array[i].value.data.len, props->array[i].value.data.data);
+		  break;
+		case UTF_8_STRING_PAIR:
+		  MyLog(LOGA_INFO, "Property name %s key %*.s value %*.s", name,
+			  props->array[i].value.data.len, props->array[i].value.data.data,
+		  	  props->array[i].value.value.len, props->array[i].value.value.data);
+		  break;
+		}
+	}
+}
+
+int test1(struct Options options)
+{
+	int subsqos = 2;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer;
+	MQTTProperties props = MQTTProperties_initializer;
+	MQTTProperties willProps = MQTTProperties_initializer;
+	MQTTResponse response;
+	int rc = 0;
+	char* test_topic = "C client test1";
+
+	fprintf(xml, "<testcase classname=\"test1\" name=\"single threaded client using receive\"");
+	global_start_time = start_clock();
+	failures = 0;
+	MyLog(LOGA_INFO, "Starting test 1 - single threaded client using receive");
+
+	rc = MQTTClient_create(&c, options.connection, "single_threaded_test",
+			MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+	assert("good rc from create",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	if (rc != MQTTCLIENT_SUCCESS)
+	{
+		MQTTClient_destroy(&c);
+		goto exit;
+	}
+
+	opts.keepAliveInterval = 20;
+	opts.cleansession = 1;
+	opts.username = "testuser";
+	opts.password = "testpassword";
+	opts.MQTTVersion = options.MQTTVersion;
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	opts.will = &wopts;
+	opts.will->message = "will message";
+	opts.will->qos = 1;
+	opts.will->retained = 0;
+	opts.will->topicName = "will topic";
+	opts.will = NULL;
+
+	MyLog(LOGA_DEBUG, "Connecting");
+	response = MQTTClient_connect5(c, &opts, &props, &willProps);
+	assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
+	if (rc != MQTTCLIENT_SUCCESS)
+		goto exit;
+
+	if (response.properties)
+	{
+		logProperties(response.properties);
+		MQTTProperties_free(response.properties);
+	}
+
+	//rc = MQTTClient_subscribe(c, test_topic, subsqos);
+	//assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	//test1_sendAndReceive(c, 0, test_topic);
+	//test1_sendAndReceive(c, 1, test_topic);
+	//test1_sendAndReceive(c, 2, test_topic);
+
+	MyLog(LOGA_DEBUG, "Stopping\n");
+
+	//rc = MQTTClient_unsubscribe(c, test_topic);
+	//assert("Unsubscribe successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+	rc = MQTTClient_disconnect(c, 0);
+	assert("Disconnect successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	/* Just to make sure we can connect again */
+	/*rc = MQTTClient_connect5(c, &opts, &props, &willProps);
+	assert("Connect successful",  rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+	rc = MQTTClient_disconnect(c, 0);
+	assert("Disconnect successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+	*/
+	MQTTClient_destroy(&c);
+
+exit:
+	MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.",
+			(failures == 0) ? "passed" : "failed", tests, failures);
+	write_test_result();
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test2: multi-threaded client using callbacks
+
+*********************************************************************/
+volatile int test2_arrivedcount = 0;
+int test2_deliveryCompleted = 0;
+MQTTClient_message test2_pubmsg = MQTTClient_message_initializer;
+
+void test2_deliveryComplete(void* context, MQTTClient_deliveryToken dt)
+{
+	++test2_deliveryCompleted;
+}
+
+int test2_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m)
+{
+	++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)
+	{
+		failures++;
+		MyLog(LOGA_INFO, "Error: wrong data received lengths %d %d\n", test2_pubmsg.payloadlen, m->payloadlen);
+	}
+	MQTTClient_free(topicName);
+	MQTTClient_freeMessage(&m);
+	return 1;
+}
+
+
+void test2_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
+{
+	MQTTClient_deliveryToken dt;
+	int i = 0;
+	int iterations = 50;
+	int rc = 0;
+	int wait_seconds = 0;
+
+	test2_deliveryCompleted = 0;
+
+	MyLog(LOGA_INFO, "%d messages at QoS %d", iterations, qos);
+	test2_pubmsg.payload = "a much longer message that we can shorten to the extent that we need to";
+	test2_pubmsg.payloadlen = 27;
+	test2_pubmsg.qos = qos;
+	test2_pubmsg.retained = 0;
+
+	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);
+		else
+			rc = MQTTClient_publishMessage(c, test_topic, &test2_pubmsg, &dt);
+		assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+		#if defined(WIN32)
+			Sleep(100);
+		#else
+			usleep(100000L);
+		#endif
+
+		wait_seconds = 10;
+		while ((test2_arrivedcount < i) && (wait_seconds-- > 0))
+		{
+			MyLog(LOGA_DEBUG, "Arrived %d count %d", test2_arrivedcount, i);
+			#if defined(WIN32)
+				Sleep(1000);
+			#else
+				usleep(1000000L);
+			#endif
+		}
+		assert("Message Arrived", wait_seconds > 0,
+				"Time out waiting for message %d\n", i );
+	}
+	if (qos > 0)
+	{
+		/* MQ Telemetry can send a message to a subscriber before the server has
+		   completed the QoS 2 handshake with the publisher. For QoS 1 and 2,
+		   allow time for the final delivery complete callback before checking
+		   that all expected callbacks have been made */
+		wait_seconds = 10;
+		while ((test2_deliveryCompleted < iterations) && (wait_seconds-- > 0))
+		{
+			MyLog(LOGA_DEBUG, "Delivery Completed %d count %d", test2_deliveryCompleted, i);
+			#if defined(WIN32)
+				Sleep(1000);
+			#else
+				usleep(1000000L);
+			#endif
+		}
+		assert("All Deliveries Complete", wait_seconds > 0,
+			   "Number of deliveryCompleted callbacks was %d\n",
+			   test2_deliveryCompleted);
+	}
+}
+
+
+int test2(struct Options options)
+{
+	char* testname = "test2";
+	int subsqos = 2;
+	/* TODO - usused - remove ? MQTTClient_deliveryToken* dt = NULL; */
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	int rc = 0;
+	char* test_topic = "C client test2";
+
+	fprintf(xml, "<testcase classname=\"test1\" name=\"multi-threaded client using callbacks\"");
+	MyLog(LOGA_INFO, "Starting test 2 - multi-threaded client using callbacks");
+	global_start_time = start_clock();
+	failures = 0;
+
+	MQTTClient_create(&c, options.connection, "multi_threaded_sample", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+
+	opts.keepAliveInterval = 20;
+	opts.cleansession = 1;
+	opts.MQTTVersion = options.MQTTVersion;
+	opts.username = "testuser";
+	opts.binarypwd.data = "testpassword";
+	opts.binarypwd.len = strlen(opts.binarypwd.data);
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	rc = MQTTClient_setCallbacks(c, NULL, NULL, test2_messageArrived, test2_deliveryComplete);
+	assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	MyLog(LOGA_DEBUG, "Connecting");
+	rc = MQTTClient_connect(c, &opts);
+	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);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	test2_sendAndReceive(c, 0, test_topic);
+	test2_sendAndReceive(c, 1, test_topic);
+	test2_sendAndReceive(c, 2, test_topic);
+
+	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);
+	assert("Disconnect successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	MQTTClient_destroy(&c);
+
+exit:
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+	write_test_result();
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test 3: connack return codes
+
+for AMQTDD, needs an amqtdd.cfg of:
+
+	allow_anonymous false
+  password_file passwords
+
+and a passwords file of:
+
+	Admin:Admin
+
+*********************************************************************/
+int test3(struct Options options)
+{
+	char* testname = "test3";
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer;
+
+	fprintf(xml, "<testcase classname=\"test1\" name=\"connack return codes\"");
+	global_start_time = start_clock();
+	failures = 0;
+	MyLog(LOGA_INFO, "Starting test 3 - connack return codes");
+
+#if 0
+	/* clientid too long (RC = 2) */
+	rc = MQTTClient_create(&c, options.connection, "client_ID_too_long_for_MQTT_protocol_version_3",
+		MQTTCLIENT_PERSISTENCE_NONE, NULL);
+	assert("good rc from create",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_connect(c, &opts);
+	assert("identifier rejected", rc == 2, "rc was %d\n", rc);
+	MQTTClient_destroy(&c);
+#endif
+	/* broker unavailable (RC = 3)  - TDD when allow_anonymous not set*/
+	rc = MQTTClient_create(&c, options.connection, "The C Client", MQTTCLIENT_PERSISTENCE_NONE, NULL);
+	assert("good rc from create",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+#if 0
+	rc = MQTTClient_connect(c, &opts);
+	assert("broker unavailable", rc == 3, "rc was %d\n", rc);
+
+	/* authentication failure (RC = 4) */
+	opts.username = "Admin";
+	opts.password = "fred";
+	rc = MQTTClient_connect(c, &opts);
+	assert("Bad user name or password", rc == 4, "rc was %d\n", rc);
+#endif
+
+	/* authorization failure (RC = 5) */
+	opts.username = "Admin";
+	opts.password = "Admin";
+	/*opts.will = &wopts;    "Admin" not authorized to publish to Will topic by default
+	opts.will->message = "will message";
+	opts.will->qos = 1;
+	opts.will->retained = 0;
+	opts.will->topicName = "will topic";*/
+	rc = MQTTClient_connect(c, &opts);
+	//assert("Not authorized", rc == 5, "rc was %d\n", rc);
+
+#if 0
+	/* successful connection (RC = 0) */
+	opts.username = "Admin";
+	opts.password = "Admin";
+  opts.will = NULL;
+	rc = MQTTClient_connect(c, &opts);
+	assert("successful connection", rc == MQTTCLIENT_SUCCESS,  "rc was %d\n", rc);
+	MQTTClient_disconnect(c, 0);
+	MQTTClient_destroy(&c);
+#endif
+
+/* TODO - unused - remove ? exit: */
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+	write_test_result();
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test 4: client persistence 1
+
+
+*********************************************************************/
+int test4_run(int qos)
+{
+	char* testname = "test 4";
+	char* topic = "Persistence test 1";
+	int subsqos = 2;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_message* m = NULL;
+	char* topicName = NULL;
+	int topicLen;
+	MQTTClient_deliveryToken* tokens = NULL;
+	int mytoken = -99;
+	char buffer[100];
+	int count = 3;
+	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);
+
+	opts.keepAliveInterval = 20;
+	opts.reliable = 0;
+	opts.MQTTVersion = options.MQTTVersion;
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	MyLog(LOGA_DEBUG, "Cleanup by connecting clean session\n");
+	opts.cleansession = 1;
+	if ((rc = MQTTClient_connect(c, &opts)) != 0)
+	{
+  		assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+		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);
+
+	/* 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);
+	}
+
+	/* disconnect immediately without receiving the incoming messages */
+	MQTTClient_disconnect(c, 0); /* now there should be "orphaned" publications */
+
+	rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
+ 	assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
+	if (tokens)
+	{
+		int i = 0;
+
+		while (tokens[i] != -1)
+			MyLog(LOGA_DEBUG, "Pending delivery token %d", tokens[i++]);
+		MQTTClient_free(tokens);
+		assert1("no of tokens should be count", i == count, "no of tokens %d count %d", i, count);
+		mytoken = tokens[0];
+	}
+
+	MQTTClient_destroy(&c); /* force re-reading persistence on create */
+
+	MQTTClient_create(&c, options.connection, "xrctest1_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+
+	rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
+	assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
+	if (tokens)
+	{
+		int i = 0;
+		while (tokens[i] != -1)
+			MyLog(LOGA_DEBUG, "Pending delivery token %d", tokens[i++]);
+		MQTTClient_free(tokens);
+		assert1("no of tokens should be count", i == count, "no of tokens %d count %d", i, count);
+	}
+
+	MyLog(LOGA_DEBUG, "Reconnecting");
+	if (MQTTClient_connect(c, &opts) != 0)
+	{
+		assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+		return -1;
+	}
+
+	for (i = 0; i < count; ++i)
+	{
+  		int dup = 0;
+		do
+		{
+    		dup = 0;
+			MQTTClient_receive(c, &topicName, &topicLen, &m, 5000);
+			if (m && m->dup)
+			{
+				assert("No duplicates should be received for qos 2", qos == 1, "qos is %d", qos);
+				MyLog(LOGA_DEBUG, "Duplicate message id %d", m->msgid);
+				MQTTClient_freeMessage(&m);
+				MQTTClient_free(topicName);
+  	    		dup = 1;
+			}
+		} while (dup == 1);
+		assert("should get a message", m != NULL, "m was %p", m);
+		if (m)
+		{
+			MyLog(LOGA_DEBUG, "Received message id %d", m->msgid);
+			assert("topicName is correct", strcmp(topicName, topic) == 0, "topicName is %s", topicName);
+			MQTTClient_freeMessage(&m);
+			MQTTClient_free(topicName);
+		}
+	}
+
+	MQTTClient_yield();  /* allow any unfinished protocol exchanges to finish */
+
+	rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
+	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);
+
+	MQTTClient_destroy(&c);
+
+/* TODO - unused -remove? exit: */
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+
+	return failures;
+}
+
+
+int test4(struct Options options)
+{
+	int rc = 0;
+	fprintf(xml, "<testcase classname=\"test1\" name=\"persistence\"");
+	global_start_time = start_clock();
+	rc = test4_run(1) + test4_run(2);
+	fprintf(xml, " time=\"%ld\" >\n", elapsed(global_start_time) / 1000);
+	if (cur_output != output)
+	{
+		fprintf(xml, "%s", output);
+		cur_output = output;
+	}
+	fprintf(xml, "</testcase>\n");
+	return rc;
+}
+
+
+/*********************************************************************
+
+Test 5: disconnect with quiesce timeout should allow exchanges to complete
+
+*********************************************************************/
+int test5(struct Options options)
+{
+	char* testname = "test 5";
+	char* topic = "Persistence test 2";
+	int subsqos = 2;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_deliveryToken* tokens = NULL;
+	char buffer[100];
+	int count = 5;
+	int i, rc;
+
+	fprintf(xml, "<testcase classname=\"test1\" name=\"disconnect with quiesce timeout should allow exchanges to complete\"");
+	global_start_time = start_clock();
+	failures = 0;
+ 	MyLog(LOGA_INFO, "Starting test 5 - disconnect with quiesce timeout should allow exchanges to complete");
+
+	MQTTClient_create(&c, options.connection, "xrctest1_test_5", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+
+	opts.keepAliveInterval = 20;
+	opts.cleansession = 0;
+	opts.reliable = 0;
+	opts.MQTTVersion = options.MQTTVersion;
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	MyLog(LOGA_DEBUG, "Connecting");
+	if ((rc = MQTTClient_connect(c, &opts)) != 0)
+	{
+		assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+		MQTTClient_destroy(&c);
+		goto exit;
+	}
+
+	rc = MQTTClient_subscribe(c, topic, subsqos);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	for (i = 0; i < count; ++i)
+	{
+		sprintf(buffer, "Message sequence no %d", i);
+		rc = MQTTClient_publish(c, topic, 10, buffer, 1, 0, NULL);
+		assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+	}
+
+	MQTTClient_disconnect(c, 1000); /* now there should be no "orphaned" publications */
+	MyLog(LOGA_DEBUG, "Disconnected");
+
+	rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
+ 	assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	assert("should get no tokens back", tokens == NULL, "tokens was %p", tokens);
+
+	MQTTClient_destroy(&c);
+
+exit:
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+	write_test_result();
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test 6: connectionLost and will message
+
+*********************************************************************/
+MQTTClient test6_c1, test6_c2;
+volatile int test6_will_message_arrived = 0;
+volatile int test6_connection_lost_called = 0;
+
+void test6_connectionLost(void* context, char* cause)
+{
+	MQTTClient c = (MQTTClient)context;
+	printf("%s -> Callback: connection lost\n", (c == test6_c1) ? "Client-1" : "Client-2");
+	test6_connection_lost_called = 1;
+}
+
+void test6_deliveryComplete(void* context, MQTTClient_deliveryToken token)
+{
+	printf("Client-2 -> Callback: publish complete for token %d\n", token);
+}
+
+char* test6_will_topic = "C Test 2: will topic";
+char* test6_will_message = "will message from Client-1";
+
+int test6_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m)
+{
+	MQTTClient c = (MQTTClient)context;
+	printf("%s -> Callback: message received on topic '%s' is '%.*s'.\n",
+			 (c == test6_c1) ? "Client-1" : "Client-2", topicName, m->payloadlen, (char*)(m->payload));
+	if (c == test6_c2 && strcmp(topicName, test6_will_topic) == 0 && memcmp(m->payload, test6_will_message, m->payloadlen) == 0)
+		test6_will_message_arrived = 1;
+	MQTTClient_free(topicName);
+	MQTTClient_freeMessage(&m);
+	return 1;
+}
+
+
+int test6(struct Options options)
+{
+	char* testname = "test6";
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_willOptions wopts =  MQTTClient_willOptions_initializer;
+	MQTTClient_connectOptions opts2 = MQTTClient_connectOptions_initializer;
+	int rc, count;
+	char* mqttsas_topic = "MQTTSAS topic";
+
+	failures = 0;
+	MyLog(LOGA_INFO, "Starting test 6 - connectionLost and will messages");
+	fprintf(xml, "<testcase classname=\"test1\" name=\"connectionLost and will messages\"");
+	global_start_time = start_clock();
+
+	opts.keepAliveInterval = 2;
+	opts.cleansession = 1;
+	opts.MQTTVersion = MQTTVERSION_3_1_1;
+	opts.will = &wopts;
+	opts.will->message = test6_will_message;
+	opts.will->qos = 1;
+	opts.will->retained = 0;
+	opts.will->topicName = test6_will_topic;
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	/* Client-1 with Will options */
+	rc = MQTTClient_create(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+	assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	if (rc != MQTTCLIENT_SUCCESS)
+		goto exit;
+
+	rc = MQTTClient_setCallbacks(test6_c1, (void*)test6_c1, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
+	assert("good rc from setCallbacks",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	if (rc != MQTTCLIENT_SUCCESS)
+		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)
+		goto exit;
+
+	/* Client - 2 (multi-threaded) */
+	rc = MQTTClient_create(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+	assert("good rc from create",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* Set the callback functions for the client */
+	rc = MQTTClient_setCallbacks(test6_c2, (void*)test6_c2, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
+	assert("good rc from setCallbacks",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* Connect to the broker */
+	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);
+
+	rc = MQTTClient_subscribe(test6_c2, test6_will_topic, 2);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* 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);
+
+	MyLog(LOGA_INFO, "Waiting to receive the will message");
+	count = 0;
+	while (++count < 40)
+	{
+		#if defined(WIN32)
+			Sleep(1000L);
+		#else
+			sleep(1);
+		#endif
+		if (test6_will_message_arrived == 1 && test6_connection_lost_called == 1)
+			break;
+	}
+	assert("will message arrived", test6_will_message_arrived == 1,
+							"will_message_arrived was %d\n", test6_will_message_arrived);
+	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);
+
+	rc = MQTTClient_isConnected(test6_c2);
+	assert("Client-2 still connected", rc == 1, "isconnected is %d", rc);
+
+	rc = MQTTClient_isConnected(test6_c1);
+	assert("Client-1 not connected", rc == 0, "isconnected is %d", rc);
+
+	rc = MQTTClient_disconnect(test6_c2, 100L);
+	assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	MQTTClient_destroy(&test6_c1);
+	MQTTClient_destroy(&test6_c2);
+
+exit:
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.\n",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+	write_test_result();
+	return failures;
+}
+
+
+int test6a(struct Options options)
+{
+	char* testname = "test6a";
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
+	MQTTClient_willOptions wopts =  MQTTClient_willOptions_initializer;
+	MQTTClient_connectOptions opts2 = MQTTClient_connectOptions_initializer;
+	int rc, count;
+	char* mqttsas_topic = "MQTTSAS topic";
+
+	failures = 0;
+	MyLog(LOGA_INFO, "Starting test 6 - connectionLost and binary will messages");
+	fprintf(xml, "<testcase classname=\"test1\" name=\"connectionLost and binary will messages\"");
+	global_start_time = start_clock();
+
+	opts.keepAliveInterval = 2;
+	opts.cleansession = 1;
+	opts.MQTTVersion = MQTTVERSION_3_1_1;
+	opts.will = &wopts;
+	opts.will->payload.data = test6_will_message;
+	opts.will->payload.len = strlen(test6_will_message) + 1;
+	opts.will->qos = 1;
+	opts.will->retained = 0;
+	opts.will->topicName = test6_will_topic;
+	if (options.haconnections != NULL)
+	{
+		opts.serverURIs = options.haconnections;
+		opts.serverURIcount = options.hacount;
+	}
+
+	/* Client-1 with Will options */
+	rc = MQTTClient_create(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+	assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	if (rc != MQTTCLIENT_SUCCESS)
+		goto exit;
+
+	rc = MQTTClient_setCallbacks(test6_c1, (void*)test6_c1, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
+	assert("good rc from setCallbacks",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	if (rc != MQTTCLIENT_SUCCESS)
+		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)
+		goto exit;
+
+	/* Client - 2 (multi-threaded) */
+	rc = MQTTClient_create(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+	assert("good rc from create",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* Set the callback functions for the client */
+	rc = MQTTClient_setCallbacks(test6_c2, (void*)test6_c2, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
+	assert("good rc from setCallbacks",  rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* Connect to the broker */
+	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);
+
+	rc = MQTTClient_subscribe(test6_c2, test6_will_topic, 2);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* 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);
+
+	MyLog(LOGA_INFO, "Waiting to receive the will message");
+	count = 0;
+	while (++count < 40)
+	{
+		#if defined(WIN32)
+			Sleep(1000L);
+		#else
+			sleep(1);
+		#endif
+		if (test6_will_message_arrived == 1 && test6_connection_lost_called == 1)
+			break;
+	}
+	assert("will message arrived", test6_will_message_arrived == 1,
+							"will_message_arrived was %d\n", test6_will_message_arrived);
+	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);
+
+	rc = MQTTClient_isConnected(test6_c2);
+	assert("Client-2 still connected", rc == 1, "isconnected is %d", rc);
+
+	rc = MQTTClient_isConnected(test6_c1);
+	assert("Client-1 not connected", rc == 0, "isconnected is %d", rc);
+
+	rc = MQTTClient_disconnect(test6_c2, 100L);
+	assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
+
+	MQTTClient_destroy(&test6_c1);
+	MQTTClient_destroy(&test6_c2);
+
+exit:
+	MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.\n",
+			(failures == 0) ? "passed" : "failed", testname, tests, failures);
+	write_test_result();
+	return failures;
+}
+
+int main(int argc, char** argv)
+{
+	int rc = 0;
+ 	int (*tests[])() = {NULL, test1, test2, test3, test4, test5, test6, test6a};
+	int i;
+
+	xml = fopen("TEST-test1.xml", "w");
+	fprintf(xml, "<testsuite name=\"test1\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1));
+
+	setenv("MQTT_C_CLIENT_TRACE", "ON", 1);
+	setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0);
+
+	getopts(argc, argv);
+
+	for (i = 0; i < options.iterations; ++i)
+	{
+	 	if (options.test_no == 0)
+		{ /* run all the tests */
+ 		   	for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
+				rc += tests[options.test_no](options); /* return number of failures.  0 = test succeeded */
+		}
+		else
+ 		   	rc = tests[options.test_no](options); /* run just the selected test */
+	}
+
+ 	if (rc == 0)
+		MyLog(LOGA_INFO, "verdict pass");
+	else
+		MyLog(LOGA_INFO, "verdict fail");
+
+	fprintf(xml, "</testsuite>\n");
+	fclose(xml);
+	return rc;
+}