Merge d423dab407 into 5b0c7ff396

Use maxInflight messages to control incoming window for MQTT 3.1.1

.github/workflows/build_linux.yml

@@ -17,7 +17,7 @@ jobs:
           mkdir build.paho
           cd build.paho
           echo "pwd $PWD"
-          cmake -DPAHO_BUILD_STATIC=FALSE -DPAHO_BUILD_SHARED=TRUE -DCMAKE_BUILD_TYPE=Debug -DPAHO_WITH_SSL=TRUE -DOPENSSL_ROOT_DIR= -DPAHO_BUILD_DOCUMENTATION=FALSE -DPAHO_BUILD_SAMPLES=TRUE -DPAHO_HIGH_PERFORMANCE=TRUE ..
+          cmake -DPAHO_BUILD_STATIC=TRUE -DPAHO_BUILD_SHARED=TRUE -DCMAKE_BUILD_TYPE=Debug -DPAHO_WITH_SSL=TRUE -DOPENSSL_ROOT_DIR= -DPAHO_BUILD_DOCUMENTATION=FALSE -DPAHO_BUILD_SAMPLES=TRUE -DPAHO_HIGH_PERFORMANCE=TRUE ..
           cmake --build .
       - name: Start test broker
         run: |

src/CMakeLists.txt

@@ -80,6 +80,7 @@ elseif(UNIX)
     set(LIBS_SYSTEM compat pthread)
   elseif(CMAKE_SYSTEM_NAME MATCHES "QNX")
     set(LIBS_SYSTEM c socket)
+    add_definitions(-D_QNX_SOURCE)
   else()
     set(LIBS_SYSTEM c pthread)
   endif()

src/Clients.h

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2024 IBM Corp. and Ian Craggs
+ * Copyright (c) 2009, 2026 IBM Corp. and Ian Craggs
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v2.0
@@ -135,7 +135,17 @@ typedef struct Clients
 	int keepAliveInterval;          /**< the MQTT keep alive interval */
 	int savedKeepAliveInterval;     /**< saved keep alive interval, in case reset by server keep alive */
 	int retryInterval;              /**< the MQTT retry interval for QoS > 0 */
-	int maxInflightMessages;        /**< the max number of inflight outbound messages we allow */
+	int maxInflightMessages;        /**< MQTT 3.1.1: the max number of inflight QoS > 0 messages in either direction.
+	                                      MQTT 5.0: sent to the server as the CONNECT RECEIVE_MAXIMUM property, limiting
+	                                      the number of inflight incoming QoS > 0 messages we accept from the server */
+	int serverReceiveMaximum;       /**< MQTT 5.0 only: the RECEIVE_MAXIMUM property received in the CONNACK
+	                                      (65535 if not sent by the server), limiting the number of inflight
+	                                      outbound QoS > 0 messages we can send to the server */
+	int incomingQoS1Count;          /**< count of QoS 1 PUBLISH messages received but not yet acknowledged
+	                                      (PUBACK not yet sent, e.g. queued because of pending socket writes).
+	                                      Added to inboundMsgs->count to check that the server isn't sending
+	                                      us more concurrent QoS > 0 publishes than maxInflightMessages allows
+	                                      (MQTT 5.0: the RECEIVE_MAXIMUM we advertised in the CONNECT packet) */
 	willMessages* will;             /**< the MQTT will message, if any */
 	List* inboundMsgs;              /**< inbound in flight messages */
 	List* outboundMsgs;				/**< outbound in flight messages */

src/MQTTAsync.c

@@ -505,7 +505,7 @@ void MQTTAsync_destroy(MQTTAsync* handle)
 	if (m == NULL)
 		goto exit;
 
-	MQTTAsync_closeSession(m->c, MQTTREASONCODE_SUCCESS, NULL);
+	MQTTAsync_closeSession(m->c, MQTTREASONCODE_SUCCESS, NULL, 0);
 
 	MQTTAsync_NULLPublishResponses(m);
 	MQTTAsync_freeResponses(m);
@@ -699,6 +699,8 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
 	setRetryLoopInterval(options->keepAliveInterval);
 	m->c->cleansession = options->cleansession;
 	m->c->maxInflightMessages = options->maxInflight;
+	m->c->serverReceiveMaximum = 65535; /* default, until/unless overridden by the CONNACK RECEIVE_MAXIMUM property */
+	m->c->incomingQoS1Count = 0;
 	if (options->struct_version >= 3)
 		m->c->MQTTVersion = options->MQTTVersion;
 	else

src/MQTTAsync.h

@@ -1248,7 +1248,14 @@ typedef struct
 	  */
 	int cleansession;
 	/**
-      * This controls how many messages can be in-flight simultaneously.
+      * This controls how many QoS > 0 messages can be in-flight simultaneously.
+      *
+      * For MQTT 5.0, this becomes the ::MQTTPROPERTY_CODE_RECEIVE_MAXIMUM
+      * property and is sent with the CONNECT packet. The Server Receive
+      * Maximum is sent from the server and is independent of this value.
+      *
+      * For MQTT 3.1.1, this controls both outbound and inbound maximum
+      * number of concurrent QoS 1 and 2 messages.
 	  */
 	int maxInflight;
 	/**

src/MQTTAsyncUtils.c

@@ -58,7 +58,7 @@ static int MQTTAsync_processCommand(void);
 static void MQTTAsync_checkTimeouts(void);
 static int MQTTAsync_completeConnection(MQTTAsyncs* m, Connack* connack);
 static void MQTTAsync_stop(void);
-static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props);
+static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props, int sendDisconnect);
 static int clientStructCompare(void* a, void* b);
 static int MQTTAsync_cleanSession(Clients* client);
 static int MQTTAsync_deliverMessage(MQTTAsyncs* m, char* topicName, size_t topicLen, MQTTAsync_message* mm);
@@ -971,7 +971,7 @@ void MQTTAsync_checkDisconnect(MQTTAsync handle, MQTTAsync_command* command)
 	if (m->c->outboundMsgs->count == 0 || MQTTTime_elapsed(command->start_time) >= (ELAPSED_TIME_TYPE)command->details.dis.timeout)
 	{
 		int was_connected = m->c->connected;
-		MQTTAsync_closeSession(m->c, command->details.dis.reasonCode, &command->properties);
+		MQTTAsync_closeSession(m->c, command->details.dis.reasonCode, &command->properties, 0);
 		if (command->details.dis.internal)
 		{
 			if (m->cl && was_connected)
@@ -1250,7 +1250,8 @@ static int MQTTAsync_processCommand(void)
 			}
 			else if (((cmd->command.type == PUBLISH && cmd->command.details.pub.qos > 0) ||
 						cmd->command.type == SUBSCRIBE || cmd->command.type == UNSUBSCRIBE) &&
-				(cmd->client->c->outboundMsgs->count >= cmd->client->c->maxInflightMessages))
+				(cmd->client->c->outboundMsgs->count >= ((cmd->client->c->MQTTVersion >= MQTTVERSION_5) ?
+					cmd->client->c->serverReceiveMaximum : cmd->client->c->maxInflightMessages)))
 			{
 				Log(TRACE_MIN, -1, "Blocking on server receive maximum for client %s",
 						cmd->client->c->clientID); /* flow control */
@@ -1639,19 +1640,23 @@ exit:
 }
 
 
-static void nextOrClose(MQTTAsyncs* m, int rc, char* message)
+static void nextOrClose(MQTTAsyncs* m, int rc, char* message, int sendDisconnect)
 {
 	int was_connected = m->c->connected;
 	int more_to_try = 0;
 	int connectionLost_called = 0;
+	enum MQTTReasonCodes reasonCode = MQTTREASONCODE_SUCCESS;
 	FUNC_ENTRY;
 
+	if (rc > 0) /* MQTT 5 reason codes are always positive */
+		reasonCode = rc;
+
 	more_to_try = MQTTAsync_checkConn(&m->connect, m, was_connected);
 	if (more_to_try)
 	{
 		MQTTAsync_queuedCommand* conn;
 
-		MQTTAsync_closeOnly(m->c, MQTTREASONCODE_SUCCESS, NULL);
+		MQTTAsync_closeOnly(m->c, reasonCode, NULL, sendDisconnect);
 		if (m->cl && was_connected)
 		{
 			Log(TRACE_MIN, -1, "Calling connectionLost for client %s", m->c->clientID);
@@ -1684,7 +1689,7 @@ static void nextOrClose(MQTTAsyncs* m, int rc, char* message)
 
 	if (!more_to_try)
 	{
-		MQTTAsync_closeSession(m->c, MQTTREASONCODE_SUCCESS, NULL);
+		MQTTAsync_closeSession(m->c, reasonCode, NULL, sendDisconnect);
 		if (m->connect.onFailure)
 		{
 			MQTTAsync_failureData data;
@@ -1754,7 +1759,7 @@ static void MQTTAsync_checkTimeouts(void)
 		/* check connect timeout */
 		if (m->c->connect_state != NOT_IN_PROGRESS && MQTTTime_elapsed(m->connect.start_time) > (ELAPSED_TIME_TYPE)(m->connectTimeout * 1000))
 		{
-			nextOrClose(m, MQTTASYNC_FAILURE, "TCP connect timeout");
+			nextOrClose(m, MQTTASYNC_FAILURE, "TCP connect timeout", 0);
 			continue;
 		}
 
@@ -2114,7 +2119,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
 		if (rc == SOCKET_ERROR)
 		{
 			Log(TRACE_MINIMUM, -1, "Error from MQTTAsync_cycle() - removing socket %d", sock);
-			nextOrClose(m, rc, "socket error");
+			nextOrClose(m, rc, "socket error", 0);
 		}
 		else
 		{
@@ -2200,16 +2205,12 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
 						if (m->c->MQTTVersion >= MQTTVERSION_5)
 						{
 							if (MQTTProperties_hasProperty(&connack->properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM))
-							{
-								int recv_max = (int)MQTTProperties_getNumericValue(&connack->properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM);
-								if (m->c->maxInflightMessages > recv_max)
-									m->c->maxInflightMessages = recv_max;
-							}
+								m->c->serverReceiveMaximum = (int)MQTTProperties_getNumericValue(&connack->properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM);
 						}
 					}
 					else
 					{
-					    nextOrClose(m, rc, "CONNACK return code");
+					    nextOrClose(m, rc, "CONNACK return code", 0);
 					}
 					MQTTPacket_freeConnack(connack);
 				}
@@ -2377,7 +2378,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
 					}
 					rc = MQTTProtocol_handleDisconnects(pack, m->c->net.socket);
 					m->c->connected = 0; /* don't send disconnect packet back */
-					nextOrClose(m, discrc, "Received disconnect");
+					nextOrClose(m, discrc, "Received disconnect", 0);
 				}
 				else
 				{
@@ -2453,7 +2454,7 @@ static void MQTTAsync_stop(void)
 }
 
 
-static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props)
+static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props, int sendDisconnect)
 {
 	FUNC_ENTRY;
 	client->good = 0;
@@ -2462,7 +2463,7 @@ static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode
 	if (client->net.socket > 0)
 	{
 		MQTTProtocol_checkPendingWrites();
-		if (client->connected && MQTTAsync_Socket_noPendingWrites(client->net.socket))
+		if (sendDisconnect && client->connected && MQTTAsync_Socket_noPendingWrites(client->net.socket))
 			MQTTPacket_send_disconnect(client, reasonCode, props);
 		MQTTAsync_lock_mutex(socket_mutex);
 		WebSocket_close(&client->net, WebSocket_CLOSE_NORMAL, NULL);
@@ -2484,10 +2485,10 @@ static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode
 }
 
 
-void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props)
+void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props, int sendDisconnect)
 {
 	FUNC_ENTRY;
-	MQTTAsync_closeOnly(client, reasonCode, props);
+	MQTTAsync_closeOnly(client, reasonCode, props, sendDisconnect);
 
 	if (client->cleansession ||
 			(client->MQTTVersion >= MQTTVERSION_5 && client->sessionExpiry == 0))
@@ -2771,9 +2772,9 @@ static int MQTTAsync_disconnect_internal(MQTTAsync handle, int timeout)
 }
 
 
-void MQTTProtocol_closeSession(Clients* c, int sendwill)
+void MQTTProtocol_closeSession(Clients* c, int rc, int sendDisconnect)
 {
-	nextOrClose((MQTTAsync)c->context, MQTTASYNC_DISCONNECTED, "MQTTProtocol_closeSession");
+	nextOrClose((MQTTAsync)c->context, rc, "MQTTProtocol_closeSession", sendDisconnect);
 }
 
 
@@ -3050,7 +3051,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
 
 exit:
 	if ((rc != 0 && rc != TCPSOCKET_INTERRUPTED && (m->c->connect_state != SSL_IN_PROGRESS && m->c->connect_state != WEBSOCKET_IN_PROGRESS)) || (rc == SSL_FATAL))
-		nextOrClose(m, MQTTASYNC_FAILURE, "TCP/TLS connect failure");
+		nextOrClose(m, MQTTASYNC_FAILURE, "TCP/TLS connect failure", 0);
 
 	FUNC_EXIT_RC(rc);
 	return rc;
@@ -3101,7 +3102,7 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc)
 			if (m->c->connect_state == WAIT_FOR_CONNACK && *rc == SOCKET_ERROR)
 			{
 				Log(TRACE_MINIMUM, -1, "CONNECT sent but MQTTPacket_Factory has returned SOCKET_ERROR");
-				nextOrClose(m, MQTTASYNC_FAILURE, "TCP connect completion failure");
+				nextOrClose(m, MQTTASYNC_FAILURE, "TCP connect completion failure", 0);
 			}
 		}
 		if (pack)

src/MQTTAsyncUtils.h

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2024 IBM Corp. and others
+ * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v2.0
@@ -168,7 +168,7 @@ void MQTTAsync_emptyMessageQueue(Clients* client);
 void MQTTAsync_freeResponses(MQTTAsyncs* m);
 void MQTTAsync_freeCommands(MQTTAsyncs* m);
 int MQTTAsync_unpersistCommandsAndMessages(Clients* c);
-void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props);
+void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props, int sendDisconnect);
 int MQTTAsync_disconnect1(MQTTAsync handle, const MQTTAsync_disconnectOptions* options, int internal);
 int MQTTAsync_assignMsgId(MQTTAsyncs* m);
 int MQTTAsync_getNoBufferedMessages(MQTTAsyncs* m);

src/MQTTClient.c

@@ -345,7 +345,8 @@ static int clientSockCompare(void* a, void* b);
 static thread_return_type WINAPI connectionLost_call(void* context);
 static thread_return_type WINAPI MQTTClient_run(void* n);
 static int MQTTClient_stop(void);
-static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props);
+static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props,
+	int sendDisconnect);
 static int MQTTClient_cleanSession(Clients* client);
 static MQTTResponse MQTTClient_connectURIVersion(
 	MQTTClient handle, MQTTClient_connectOptions* options,
@@ -354,8 +355,9 @@ static MQTTResponse MQTTClient_connectURIVersion(
 	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, enum MQTTReasonCodes, MQTTProperties*);
-static int MQTTClient_disconnect_internal(MQTTClient handle, int timeout);
+static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int internal, int stop, enum MQTTReasonCodes,
+	MQTTProperties*, int sendDisconnect);
+static int MQTTClient_disconnect_internal(MQTTClient handle, int timeout, int rc, int sendDisconnect);
 static void MQTTClient_retry(void);
 static MQTTPacket* MQTTClient_cycle(SOCKET* sock, ELAPSED_TIME_TYPE timeout, int* rc);
 static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* rc, int64_t timeout);
@@ -885,7 +887,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
 		if (rc == SOCKET_ERROR)
 		{
 			if (m->c->connected)
-				MQTTClient_disconnect_internal(m, 0);
+				MQTTClient_disconnect_internal(m, 0, rc, 0);
 			else
 			{
 				if (m->c->connect_state == SSL_IN_PROGRESS)
@@ -969,7 +971,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
 							if (dp->properties)
 							{
 								*(dp->properties) = disc->properties;
-								MQTTClient_disconnect1(m, 10, 0, 1, MQTTREASONCODE_SUCCESS, NULL);
+								MQTTClient_disconnect1(m, 10, 0, 1, MQTTREASONCODE_SUCCESS, NULL, 0);
 								Log(TRACE_MIN, -1, "Calling disconnected for client %s", m->c->clientID);
 								Paho_thread_start(call_disconnected, dp);
 							}
@@ -1115,7 +1117,7 @@ int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClient_connect
 }
 
 
-static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props)
+static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props, int sendDisconnect)
 {
 	FUNC_ENTRY;
 	client->good = 0;
@@ -1123,7 +1125,7 @@ static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason
 	client->ping_due = 0;
 	if (client->net.socket > 0)
 	{
-		if (client->connected)
+		if (sendDisconnect && client->connected)
 			MQTTPacket_send_disconnect(client, reason, props);
 		Paho_thread_lock_mutex(socket_mutex);
 		WebSocket_close(&client->net, WebSocket_CLOSE_NORMAL, NULL);
@@ -1514,7 +1516,7 @@ exit:
 		}
 	}
 	else
-		MQTTClient_disconnect1(handle, 0, 0, (MQTTVersion == 3), MQTTREASONCODE_SUCCESS, NULL); /* don't want to call connection lost */
+		MQTTClient_disconnect1(handle, 0, 0, (MQTTVersion == 3), MQTTREASONCODE_SUCCESS, NULL, 0); /* don't want to call connection lost */
 
 	resp.reasonCode = rc;
 	FUNC_EXIT_RC(resp.reasonCode);
@@ -1565,6 +1567,8 @@ static MQTTResponse MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectO
 		if (options->maxInflightMessages > 0)
 			m->c->maxInflightMessages = options->maxInflightMessages;
 	}
+	m->c->serverReceiveMaximum = 65535; /* default, until/unless overridden by the CONNACK RECEIVE_MAXIMUM property */
+	m->c->incomingQoS1Count = 0;
 
 	if (options->struct_version >= 7)
 	{
@@ -1943,11 +1947,7 @@ MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions*
 	if (rc.reasonCode == MQTTREASONCODE_SUCCESS)
 	{
 		if (rc.properties && MQTTProperties_hasProperty(rc.properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM))
-		{
-			int recv_max = (int)MQTTProperties_getNumericValue(rc.properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM);
-			if (m->c->maxInflightMessages > recv_max)
-				m->c->maxInflightMessages = recv_max;
-		}
+			m->c->serverReceiveMaximum = (int)MQTTProperties_getNumericValue(rc.properties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM);
 	}
 
 exit:
@@ -1971,7 +1971,7 @@ exit:
  * mqttclient_mutex must be locked when you call this function, if multi threaded
  */
 static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int call_connection_lost, int stop,
-		enum MQTTReasonCodes reason, MQTTProperties* props)
+		enum MQTTReasonCodes reason, MQTTProperties* props, int sendDisconnect)
 {
 	MQTTClients* m = handle;
 	START_TIME_TYPE start;
@@ -2002,7 +2002,7 @@ static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int call_conne
 		}
 	}
 
-	MQTTClient_closeSession(m->c, reason, props);
+	MQTTClient_closeSession(m->c, reason, props, sendDisconnect);
 
 exit:
 	if (stop)
@@ -2023,18 +2023,18 @@ exit:
 /**
  * mqttclient_mutex must be locked when you call this function, if multi threaded
  */
-static int MQTTClient_disconnect_internal(MQTTClient handle, int timeout)
+static int MQTTClient_disconnect_internal(MQTTClient handle, int timeout, int rc, int sendDisconnect)
 {
-	return MQTTClient_disconnect1(handle, timeout, 1, 1, MQTTREASONCODE_SUCCESS, NULL);
+	return MQTTClient_disconnect1(handle, timeout, 1, 1, MQTTREASONCODE_SUCCESS, NULL, sendDisconnect);
 }
 
 
 /**
  * mqttclient_mutex must be locked when you call this function, if multi threaded
  */
-void MQTTProtocol_closeSession(Clients* c, int sendwill)
+void MQTTProtocol_closeSession(Clients* c, int rc, int sendDisconnect)
 {
-	MQTTClient_disconnect_internal((MQTTClient)c->context, 0);
+	MQTTClient_disconnect_internal((MQTTClient)c->context, 0, rc, sendDisconnect);
 }
 
 
@@ -2043,7 +2043,7 @@ int MQTTClient_disconnect(MQTTClient handle, int timeout)
 	int rc = 0;
 
 	Paho_thread_lock_mutex(mqttclient_mutex);
-	rc = MQTTClient_disconnect1(handle, timeout, 0, 1, MQTTREASONCODE_SUCCESS, NULL);
+	rc = MQTTClient_disconnect1(handle, timeout, 0, 1, MQTTREASONCODE_SUCCESS, NULL, 1);
 	Paho_thread_unlock_mutex(mqttclient_mutex);
 	return rc;
 }
@@ -2054,7 +2054,7 @@ int MQTTClient_disconnect5(MQTTClient handle, int timeout, enum MQTTReasonCodes
 	int rc = 0;
 
 	Paho_thread_lock_mutex(mqttclient_mutex);
-	rc = MQTTClient_disconnect1(handle, timeout, 0, 1, reason, props);
+	rc = MQTTClient_disconnect1(handle, timeout, 0, 1, reason, props, 1);
 	Paho_thread_unlock_mutex(mqttclient_mutex);
 	return rc;
 }
@@ -2197,7 +2197,7 @@ MQTTResponse MQTTClient_subscribeMany5(MQTTClient handle, int count, char* const
 	}
 
 	if (rc == SOCKET_ERROR)
-		MQTTClient_disconnect_internal(handle, 0);
+		MQTTClient_disconnect_internal(handle, 0, rc, 0);
 	else if (rc == TCPSOCKET_COMPLETE)
 		rc = MQTTCLIENT_SUCCESS;
 
@@ -2347,7 +2347,7 @@ MQTTResponse MQTTClient_unsubscribeMany5(MQTTClient handle, int count, char* con
 	}
 
 	if (rc == SOCKET_ERROR)
-		MQTTClient_disconnect_internal(handle, 0);
+		MQTTClient_disconnect_internal(handle, 0, rc, 0);
 
 exit:
 	if (rc < 0)
@@ -2414,8 +2414,10 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p
 	if (rc != MQTTCLIENT_SUCCESS)
 		goto exit;
 
-	/* If outbound queue is full, block until it is not */
-	while (m->c->outboundMsgs->count >= m->c->maxInflightMessages ||
+	/* If outbound queue is full, block until it is not.
+	   MQTT 5.0: the server's CONNACK RECEIVE_MAXIMUM limits our outbound QoS > 0 messages.
+	   MQTT < 5.0: maxInflightMessages limits outbound (and, implicitly, inbound) QoS > 0 messages. */
+	while (m->c->outboundMsgs->count >= ((m->c->MQTTVersion >= MQTTVERSION_5) ? m->c->serverReceiveMaximum : m->c->maxInflightMessages) ||
          Socket_noPendingWrites(m->c->net.socket) == 0) /* wait until the socket is free of large packets being written */
 	{
 		if (blocked == 0)
@@ -2517,7 +2519,7 @@ exit_and_free:
 
 	if (rc == SOCKET_ERROR)
 	{
-		MQTTClient_disconnect_internal(handle, 0);
+		MQTTClient_disconnect_internal(handle, 0, rc, 0);
 		/* Return success for qos > 0 as the send will be retried automatically */
 		rc = (qos > 0) ? MQTTCLIENT_SUCCESS : MQTTCLIENT_FAILURE;
 	}
@@ -2870,7 +2872,7 @@ int MQTTClient_receive(MQTTClient handle, char** topicName, int* topicLen, MQTTC
 		rc = MQTTClient_deliverMessage(rc, m, topicName, topicLen, message);
 
 	if (rc == SOCKET_ERROR)
-		MQTTClient_disconnect_internal(handle, 0);
+		MQTTClient_disconnect_internal(handle, 0, rc, 0);
 
 exit:
 	FUNC_EXIT_RC(rc);
@@ -2902,7 +2904,7 @@ void MQTTClient_yield(void)
 		{
 			MQTTClients* m = (MQTTClient)(handles->current->content);
 			if (m->c->connect_state != DISCONNECTING)
-				MQTTClient_disconnect_internal(m, 0);
+				MQTTClient_disconnect_internal(m, 0, rc, 0);
 		}
 		Paho_thread_unlock_mutex(mqttclient_mutex);
 		elapsed = MQTTTime_elapsed(start);

src/MQTTPacketOut.c

@@ -51,11 +51,31 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
 	char *buf, *ptr;
 	Connect packet;
 	int rc = SOCKET_ERROR, len;
+	MQTTProperties localConnectProperties = MQTTProperties_initializer;
 
 	FUNC_ENTRY;
 	packet.header.byte = 0;
 	packet.header.bits.type = CONNECT;
 
+	if (MQTTVersion >= MQTTVERSION_5)
+	{
+		/* Advertise our own RECEIVE_MAXIMUM (limiting inbound QoS > 0 publishes from the server),
+		 * unless the caller has already set one explicitly. Work on a local copy so that repeated
+		 * calls (e.g. on reconnect) don't keep appending the property to the caller's properties. */
+		localConnectProperties = MQTTProperties_copy(connectProperties);
+		if (client->maxInflightMessages != 65535 && /* the default is 65535 if not present */
+			!MQTTProperties_hasProperty(&localConnectProperties, MQTTPROPERTY_CODE_RECEIVE_MAXIMUM))
+		{
+			MQTTProperty prop;
+
+			prop.identifier = MQTTPROPERTY_CODE_RECEIVE_MAXIMUM;
+			prop.value.integer2 = (unsigned short)client->maxInflightMessages;
+			if (MQTTProperties_add(&localConnectProperties, &prop) != 0)
+				goto exit_nofree;
+		}
+		connectProperties = &localConnectProperties;
+	}
+
 	len = ((MQTTVersion == MQTTVERSION_3_1) ? 12 : 10) + (int)strlen(client->clientID)+2;
 	if (client->will)
 		len += (int)strlen(client->will->topic)+2 + client->will->payloadlen+2;
@@ -126,6 +146,8 @@ exit:
 	if (rc != TCPSOCKET_INTERRUPTED)
 		free(buf);
 exit_nofree:
+	if (MQTTVersion >= MQTTVERSION_5)
+		MQTTProperties_free(&localConnectProperties);
 	FUNC_EXIT_RC(rc);
 	return rc;
 }

src/MQTTProtocolClient.c

@@ -331,6 +331,11 @@ int MQTTProtocol_handlePublishes(void* pack, SOCKET sock)
 	FUNC_ENTRY;
 	client = (Clients*)(ListFindItem(bstate->clients, &sock, clientSocketCompare)->content);
 	clientid = client->clientID;
+	if (client->MQTTVersion != publish->MQTTVersion)
+	{
+		rc = SOCKET_ERROR;
+		goto exit;
+	}
 
 	/* Format and print publish data to trace */
 	{
@@ -351,14 +356,47 @@ int MQTTProtocol_handlePublishes(void* pack, SOCKET sock)
 
 	socketHasPendingWrites = !Socket_noPendingWrites(sock);
 
+	/* if the message id already exists in the input queue, it's not a new message */
+	if (ListFindItem(client->inboundMsgs, &(publish->msgId), messageIDCompare) == NULL)
+	{
+		/* the server must not send us more concurrent QoS > 0 publishes than we can handle:
+		 * MQTT 5.0: the RECEIVE_MAXIMUM we advertised in our CONNECT packet (client->maxInflightMessages).
+		 * MQTT < 5.0: client->maxInflightMessages is used in both directions. */
+		int inflight = client->inboundMsgs->count + client->incomingQoS1Count + 1; /* +1 for this new message */
+
+		if (inflight > client->maxInflightMessages)
+		{
+			client->good = 0;
+			if (publish->MQTTVersion >= MQTTVERSION_5)
+			{
+				Log(TRACE_PROTOCOL, -1, "Receive maximum (%d) exceeded by server for client %s, disconnecting",
+						client->maxInflightMessages, clientid);
+				MQTTProtocol_closeSession(client, MQTTREASONCODE_RECEIVE_MAXIMUM_EXCEEDED, 1);
+			}
+			else
+			{
+				Log(LOG_ERROR, -1, "Max inflight messages (%d) exceeded by server for client %s, disconnecting",
+						client->maxInflightMessages, clientid);
+				MQTTProtocol_closeSession(client,SOCKET_ERROR,0);
+			}
+			rc = SOCKET_ERROR;
+			goto exit;
+		}
+	}
+
 	if (publish->header.bits.qos == 1)
 	{
 		Protocol_processPublication(publish, client, 1);
 
+		++client->incomingQoS1Count;
+
 		if (socketHasPendingWrites)
 			rc = MQTTProtocol_queueAck(client, PUBACK, publish->msgId);
 		else
+		{
 			rc = MQTTPacket_send_puback(publish->MQTTVersion, publish->msgId, &client->net, client->clientID);
+			--client->incomingQoS1Count;
+		}
 	}
 	else if (publish->header.bits.qos == 2)
 	{
@@ -730,7 +768,7 @@ void MQTTProtocol_keepalive(START_TIME_TYPE now)
 				MQTTTime_difftime(now, client->net.lastReceived) >= (DIFF_TIME_TYPE)(client->keepAliveInterval * 1500))
 			{
 				Log(TRACE_PROTOCOL, -1, "PINGRESP not received in keepalive interval for client %s on socket %d, disconnecting", client->clientID, client->net.socket);
-				MQTTProtocol_closeSession(client, 1);
+				MQTTProtocol_closeSession(client, MQTTREASONCODE_KEEP_ALIVE_TIMEOUT, 1);
 			}
 		}
 		else if (client->ping_due == 1 &&
@@ -742,7 +780,7 @@ void MQTTProtocol_keepalive(START_TIME_TYPE now)
 			{
 				/* ping still outstanding after keep alive interval, so close session */
 				Log(TRACE_PROTOCOL, -1, "PINGREQ still outstanding for client %s on socket %d, disconnecting", client->clientID, client->net.socket);
-				MQTTProtocol_closeSession(client, 1);
+				MQTTProtocol_closeSession(client, MQTTREASONCODE_KEEP_ALIVE_TIMEOUT, 1);
 			}
 		}
 		else if (MQTTTime_difftime(now, client->net.lastSent) >= (DIFF_TIME_TYPE)(client->keepAliveInterval * 1000))
@@ -753,7 +791,7 @@ void MQTTProtocol_keepalive(START_TIME_TYPE now)
 				if (MQTTPacket_send_pingreq(&client->net, client->clientID) != TCPSOCKET_COMPLETE)
 				{
 					Log(TRACE_PROTOCOL, -1, "Error sending PINGREQ for client %s on socket %d, disconnecting", client->clientID, client->net.socket);
-					MQTTProtocol_closeSession(client, 1);
+					MQTTProtocol_closeSession(client, SOCKET_ERROR,0);
 				}
 				else
 				{
@@ -779,7 +817,7 @@ void MQTTProtocol_keepalive(START_TIME_TYPE now)
 				if (MQTTPacket_send_pingreq(&client->net, client->clientID) != TCPSOCKET_COMPLETE)
 				{
 					Log(TRACE_PROTOCOL, -1, "Error sending PINGREQ for client %s on socket %d, disconnecting", client->clientID, client->net.socket);
-					MQTTProtocol_closeSession(client, 1);
+					MQTTProtocol_closeSession(client, SOCKET_ERROR, 0);
 				}
 				else
 				{
@@ -844,7 +882,7 @@ static void MQTTProtocol_retries(START_TIME_TYPE now, Clients* client, int regar
 					client->good = 0;
 					Log(TRACE_PROTOCOL, 29, NULL, client->clientID, client->net.socket,
 												Socket_getpeer(client->net.socket));
-					MQTTProtocol_closeSession(client, 1);
+					MQTTProtocol_closeSession(client, rc, 0);
 					client = NULL;
 				}
 				else
@@ -862,7 +900,7 @@ static void MQTTProtocol_retries(START_TIME_TYPE now, Clients* client, int regar
 					client->good = 0;
 					Log(TRACE_PROTOCOL, 29, NULL, client->clientID, client->net.socket,
 							Socket_getpeer(client->net.socket));
-					MQTTProtocol_closeSession(client, 1);
+					MQTTProtocol_closeSession(client, SOCKET_ERROR, 0);
 					client = NULL;
 				}
 				else
@@ -925,7 +963,7 @@ void MQTTProtocol_retry(START_TIME_TYPE now, int doRetry, int regardless)
 			continue;
 		if (client->good == 0)
 		{
-			MQTTProtocol_closeSession(client, 1);
+			MQTTProtocol_closeSession(client, SOCKET_ERROR, 0);
 			continue;
 		}
 		if (Socket_noPendingWrites(client->net.socket) == 0)
@@ -1061,6 +1099,8 @@ void MQTTProtocol_writeAvailable(SOCKET socket)
 		{
 			case PUBACK:
 				rc = MQTTPacket_send_puback(client->MQTTVersion, ackReq->messageId, &client->net, client->clientID);
+				if (client->incomingQoS1Count > 0)
+					--client->incomingQoS1Count;
 				break;
 			case PUBREC:
 				rc = MQTTPacket_send_pubrec(client->MQTTVersion, ackReq->messageId, &client->net, client->clientID);

src/MQTTProtocolClient.h

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2022 IBM Corp.
+ * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v2.0
@@ -45,7 +45,7 @@ int MQTTProtocol_handlePubrecs(void* pack, SOCKET sock, Publications** pubToRemo
 int MQTTProtocol_handlePubrels(void* pack, SOCKET sock);
 int MQTTProtocol_handlePubcomps(void* pack, SOCKET sock, Publications** pubToRemove);
 
-void MQTTProtocol_closeSession(Clients* c, int sendwill);
+void MQTTProtocol_closeSession(Clients* c, int rc, int sendDisconnect);
 void MQTTProtocol_keepalive(START_TIME_TYPE);
 void MQTTProtocol_retry(START_TIME_TYPE, int, int);
 void MQTTProtocol_freeClient(Clients* client);

src/SHA1.c

@@ -16,6 +16,13 @@
 
 #include "SHA1.h"
 
+// In the source files using endian functions (like SHA1.c):
+#ifdef __QNXNTO__
+#include <gulliver.h>
+#define be32toh(x) ENDIAN_BE32(x)
+#define htobe32(x) ENDIAN_BE32(x)
+#endif
+
 #if !defined(OPENSSL)
 #if defined(_WIN32)
 #pragma comment(lib, "crypt32.lib")

src/Socket.c

@@ -10,6 +10,13 @@
  * and the Eclipse Distribution License is available at
  *   http://www.eclipse.org/org/documents/edl-v10.php.
  *
+ * AI Disclosure: This file was partly AI-generated. The AI-generated
+ * portions are made available under CC0-1.0 and not subject to the
+ * project's licence. The human contributor has reviewed and verified
+ * that the code is correct.
+ *
+ * SPDX-License-Identifier: EPL-2.0 and CC0-1.0
+ *
  * Contributors:
  *    Ian Craggs - initial implementation and documentation
  *    Ian Craggs - async client updates

test/CMakeLists.txt

@@ -2114,3 +2114,101 @@ if(PAHO_BUILD_SHARED)
     PROPERTIES TIMEOUT 540
   )
 endif()
+
+
+# White-box unit tests for internal modules (Base64, Tree, Proxy parsing, MQTTProperties, utf-8)
+# that don't need a broker connection, so link directly against the static library to reach
+# internal symbols not exposed by the public MQTTClient/MQTTAsync headers.
+if(PAHO_BUILD_STATIC)
+  add_executable(test_unit_coverage test_unit_coverage.c)
+  target_link_libraries(test_unit_coverage paho-mqtt3a-static)
+
+  add_test(
+    NAME test_unit_coverage
+    COMMAND "test_unit_coverage"
+  )
+endif()
+
+
+# Black-box tests, using only the public MQTTClient API, that exercise specific
+# MQTT 3.1.1/5.0 conformance-statement log lines in the paho.mqtt.testing broker
+# that aren't otherwise reached by the rest of the test suite.
+if(PAHO_BUILD_STATIC)
+  add_executable(test_conformance-static test_conformance.c)
+  target_link_libraries(test_conformance-static paho-mqtt3c-static)
+
+  add_test(NAME test_conformance-1-subscribe_empty-static
+    COMMAND "test_conformance-static" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-2-keepalive_ping-static
+    COMMAND "test_conformance-static" "--test_no" "2" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-3-server_keepalive_override-static
+    COMMAND "test_conformance-static" "--test_no" "3" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-4-zero_length_clientid-static
+    COMMAND "test_conformance-static" "--test_no" "4" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-5-duplicate_clientid-static
+    COMMAND "test_conformance-static" "--test_no" "5" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-6-zero_byte_retained_delete-static
+    COMMAND "test_conformance-static" "--test_no" "6" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-7-unsubscribe_no_match-static
+    COMMAND "test_conformance-static" "--test_no" "7" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-8-retained_flag_live_publish-static
+    COMMAND "test_conformance-static" "--test_no" "8" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-9-password_without_username-static
+    COMMAND "test_conformance-static" "--test_no" "9" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-10-user_property_order-static
+    COMMAND "test_conformance-static" "--test_no" "10" "--connection" ${MQTT_TEST_BROKER})
+
+  set_tests_properties(
+    test_conformance-1-subscribe_empty-static
+    test_conformance-2-keepalive_ping-static
+    test_conformance-3-server_keepalive_override-static
+    test_conformance-4-zero_length_clientid-static
+    test_conformance-5-duplicate_clientid-static
+    test_conformance-6-zero_byte_retained_delete-static
+    test_conformance-7-unsubscribe_no_match-static
+    test_conformance-8-retained_flag_live_publish-static
+    test_conformance-9-password_without_username-static
+    test_conformance-10-user_property_order-static
+    PROPERTIES TIMEOUT 60
+  )
+endif()
+
+if(PAHO_BUILD_SHARED)
+  add_executable(test_conformance test_conformance.c)
+  target_link_libraries(test_conformance paho-mqtt3c)
+
+  add_test(NAME test_conformance-1-subscribe_empty
+    COMMAND "test_conformance" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-2-keepalive_ping
+    COMMAND "test_conformance" "--test_no" "2" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-3-server_keepalive_override
+    COMMAND "test_conformance" "--test_no" "3" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-4-zero_length_clientid
+    COMMAND "test_conformance" "--test_no" "4" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-5-duplicate_clientid
+    COMMAND "test_conformance" "--test_no" "5" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-6-zero_byte_retained_delete
+    COMMAND "test_conformance" "--test_no" "6" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-7-unsubscribe_no_match
+    COMMAND "test_conformance" "--test_no" "7" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-8-retained_flag_live_publish
+    COMMAND "test_conformance" "--test_no" "8" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-9-password_without_username
+    COMMAND "test_conformance" "--test_no" "9" "--connection" ${MQTT_TEST_BROKER})
+  add_test(NAME test_conformance-10-user_property_order
+    COMMAND "test_conformance" "--test_no" "10" "--connection" ${MQTT_TEST_BROKER})
+
+  set_tests_properties(
+    test_conformance-1-subscribe_empty
+    test_conformance-2-keepalive_ping
+    test_conformance-3-server_keepalive_override
+    test_conformance-4-zero_length_clientid
+    test_conformance-5-duplicate_clientid
+    test_conformance-6-zero_byte_retained_delete
+    test_conformance-7-unsubscribe_no_match
+    test_conformance-8-retained_flag_live_publish
+    test_conformance-9-password_without_username
+    test_conformance-10-user_property_order
+    PROPERTIES TIMEOUT 60
+  )
+endif()

test/test_conformance.c

@@ -0,0 +1,792 @@
+/*******************************************************************************
+ * Copyright (c) 2026 Ian Craggs
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    https://www.eclipse.org/legal/epl-2.0/
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * AI Disclosure: This file was partly AI-generated. The AI-generated
+ * portions are made available under CC0-1.0 and not subject to the
+ * project's licence. The human contributor has reviewed and verified
+ * that the code is correct.
+ *
+ * SPDX-License-Identifier: EPL-2.0 and CC0-1.0
+ *
+ * Contributors:
+ *   Ian Craggs - initial implementation and documentation
+ *******************************************************************************/
+
+/**
+ * @file
+ * Black-box tests, using only the public MQTTClient API, whose purpose is to
+ * make the paho.mqtt.testing broker (github.com/eclipse-paho/paho.mqtt.testing)
+ * exercise specific MQTT 3.1.1 / 5.0 conformance-statement log lines in its own
+ * source that the rest of the C client test suite doesn't otherwise reach:
+ *
+ *  - MQTT-3.8.3-1 / MQTT5-3.8.3-1   empty subscribe topic list
+ *  - MQTT-3.12.4-1 / MQTT5-3.1.2-20 keepalive PINGREQ/PINGRESP
+ *  - MQTT5-3.1.2-21                 server overriding the requested keepalive
+ *  - MQTT5-3.1.3-6 / MQTT5-3.1.3-7  zero-length client id assigned by server
+ *  - MQTT5-3.1.4-3                  second client with the same id disconnects the first
+ *  - MQTT-3.3.1-11                  zero-byte retained publish deletes a retained message
+ *  - MQTT-3.10.4-5                  unsubscribe from a non-subscribed topic still gets an unsuback
+ *  - MQTT-2.1.2-6 / MQTT-2.1.2-10   retained flag on store, cleared flag on live delivery
+ *  - MQTT-3.1.2-22 / MQTT5-3.1.2-22 password set without username
+ *  - MQTT-3.1.3-10                  ordering of multiple user properties on a publish
+ */
+
+#include "MQTTClient.h"
+#include <string.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <stdarg.h>
+
+#if !defined(_WINDOWS)
+  #include <sys/time.h>
+  #include <unistd.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;
+	int verbose;
+	int test_no;
+} options =
+{
+	"tcp://localhost:1883",
+	0,
+	0,
+};
+
+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];
+			else
+				usage();
+		}
+		else if (strcmp(argv[count], "--verbose") == 0)
+			options.verbose = 1;
+		count++;
+	}
+}
+
+#define LOGA_DEBUG 0
+#define LOGA_INFO 1
+
+void MyLog(int LOGA_level, char* format, ...)
+{
+	va_list args;
+
+	if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
+		return;
+
+	va_start(args, format);
+	vprintf(format, args);
+	printf("\n");
+	va_end(args);
+}
+
+int tests = 0;
+int failures = 0;
+
+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", filename, lineno, description);
+		va_start(args, format);
+		vprintf(format, args);
+		va_end(args);
+	}
+	else
+		MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
+}
+
+#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
+
+#if defined(_WIN32)
+#define mysleep(ms) Sleep(ms)
+#else
+#define mysleep(ms) usleep((ms) * 1000L)
+#endif
+
+
+/*********************************************************************
+
+Test1: subscribe with an empty topic list is rejected by the broker
+       [MQTT-3.8.3-1] / [MQTT5-3.8.3-1]
+
+*********************************************************************/
+int test_subscribe_empty(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	int MQTTVersions[] = { MQTTVERSION_3_1_1, MQTTVERSION_5 };
+	size_t i;
+
+	MyLog(LOGA_INFO, "Starting test - subscribe with empty topic list");
+	failures = 0;
+
+	for (i = 0; i < ARRAY_SIZE(MQTTVersions); ++i)
+	{
+		createOpts.MQTTVersion = MQTTVersions[i];
+		rc = MQTTClient_createWithOptions(&c, options.connection, "conformance_sub_empty", MQTTCLIENT_PERSISTENCE_NONE,
+				NULL, &createOpts);
+		assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+		opts.keepAliveInterval = 20;
+		opts.MQTTVersion = MQTTVersions[i];
+		opts.cleansession = (MQTTVersions[i] >= MQTTVERSION_5) ? 0 : 1;
+		opts.cleanstart = (MQTTVersions[i] >= MQTTVERSION_5) ? 1 : 0;
+
+		if (MQTTVersions[i] >= MQTTVERSION_5)
+		{
+			MQTTResponse response = MQTTClient_connect5(c, &opts, NULL, NULL);
+			rc = response.reasonCode;
+		}
+		else
+			rc = MQTTClient_connect(c, &opts);
+		assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+		/* deliberately empty: no topics/qos at all, count == 0 */
+		if (MQTTVersions[i] >= MQTTVERSION_5)
+		{
+			MQTTResponse response = MQTTClient_subscribeMany5(c, 0, NULL, NULL, NULL, NULL);
+			rc = response.reasonCode;
+		}
+		else
+			rc = MQTTClient_subscribeMany(c, 0, NULL, NULL);
+		MyLog(LOGA_INFO, "subscribeMany with 0 topics returned %d for MQTT version %d", rc, MQTTVersions[i]);
+
+		/* the broker considers this a protocol violation and drops the connection without a suback,
+		 * so any subsequent operation on this connection should reflect that */
+		mysleep(500);
+		assert("Connection dropped after empty subscribe", MQTTClient_isConnected(c) == 0, "isConnected was %d\n",
+				MQTTClient_isConnected(c));
+
+		MQTTClient_disconnect(c, 0);
+		MQTTClient_destroy(&c);
+	}
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test2: an idle connection with a short keepalive causes the client to send
+       PINGREQ, and the broker to respond with PINGRESP
+       [MQTT-3.12.4-1] / [MQTT5-3.1.2-20]
+
+*********************************************************************/
+int test_keepalive_ping(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	int MQTTVersions[] = { MQTTVERSION_3_1_1, MQTTVERSION_5 };
+	size_t i;
+
+	MyLog(LOGA_INFO, "Starting test - idle connection triggers keepalive ping");
+	failures = 0;
+
+	for (i = 0; i < ARRAY_SIZE(MQTTVersions); ++i)
+	{
+		rc = MQTTClient_create(&c, options.connection, "conformance_keepalive", MQTTCLIENT_PERSISTENCE_NONE, NULL);
+		assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+		opts.keepAliveInterval = 2;
+		opts.MQTTVersion = MQTTVersions[i];
+		opts.cleansession = (MQTTVersions[i] >= MQTTVERSION_5) ? 0 : 1;
+		opts.cleanstart = (MQTTVersions[i] >= MQTTVERSION_5) ? 1 : 0;
+
+		rc = MQTTClient_connect(c, &opts);
+		assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+		/* stay idle for a few keepalive intervals: this is a synchronous-mode client (no
+		 * setCallbacks call), so per the library's documented usage it must call yield()
+		 * or receive() periodically for the internal keepalive ping to actually be sent */
+		{
+			int count;
+			for (count = 0; count < 60; ++count)
+			{
+				MQTTClient_yield();
+				mysleep(100);
+			}
+		}
+
+		assert("Still connected after idle period covering several keepalive intervals",
+				MQTTClient_isConnected(c) == 1, "isConnected was %d\n", MQTTClient_isConnected(c));
+
+		rc = MQTTClient_disconnect(c, 1000);
+		assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+		MQTTClient_destroy(&c);
+	}
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test3: requesting a keepalive longer than the server's configured maximum
+       causes the server to return a shorter SERVER_KEEP_ALIVE property
+       [MQTT5-3.1.2-21]
+
+*********************************************************************/
+int test_server_keepalive_override(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+
+	MyLog(LOGA_INFO, "Starting test - server keepalive override");
+	failures = 0;
+
+	createOpts.MQTTVersion = MQTTVERSION_5;
+	rc = MQTTClient_createWithOptions(&c, options.connection, "conformance_server_ka", MQTTCLIENT_PERSISTENCE_NONE,
+			NULL, &createOpts);
+	assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* the test broker's configured serverKeepAlive is 60 seconds; requesting more than that
+	 * should cause it to return SERVER_KEEP_ALIVE in the CONNACK properties */
+	opts.keepAliveInterval = 120;
+	opts.cleanstart = 1;
+	opts.MQTTVersion = MQTTVERSION_5;
+
+	response = MQTTClient_connect5(c, &opts, NULL, NULL);
+	assert("Good rc from connect", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+
+	if (response.properties != NULL)
+	{
+		int hasProp = MQTTProperties_hasProperty(response.properties, MQTTPROPERTY_CODE_SERVER_KEEP_ALIVE);
+		assert("Server returned a keepalive override", hasProp == 1, "hasProperty was %d\n", hasProp);
+		if (hasProp)
+		{
+			int64_t serverKA = MQTTProperties_getNumericValue(response.properties, MQTTPROPERTY_CODE_SERVER_KEEP_ALIVE);
+			assert("Server keepalive is shorter than requested", serverKA > 0 && serverKA < 120,
+					"serverKA was %lld\n", (long long)serverKA);
+		}
+	}
+	else
+		assert("CONNACK properties should be present", 0, "properties was NULL\n", 0);
+
+	MQTTClient_disconnect(c, 1000);
+	MQTTClient_destroy(&c);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test4: connecting with an empty client id (MQTT 5.0, clean start) causes the
+       server to assign one and return it in ASSIGNED_CLIENT_IDENTIFIER
+       [MQTT5-3.1.3-6] / [MQTT5-3.1.3-7]
+
+*********************************************************************/
+int test_zero_length_clientid(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+
+	MyLog(LOGA_INFO, "Starting test - zero length client id");
+	failures = 0;
+
+	/* MQTTCLIENT_PERSISTENCE_NONE bypasses the C library's own client-side rejection
+	 * of an empty client id, which only applies to the default (file) persistence */
+	createOpts.MQTTVersion = MQTTVERSION_5;
+	rc = MQTTClient_createWithOptions(&c, options.connection, "", MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
+	assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	opts.keepAliveInterval = 20;
+	opts.cleanstart = 1;
+	opts.MQTTVersion = MQTTVERSION_5;
+
+	response = MQTTClient_connect5(c, &opts, NULL, NULL);
+	assert("Good rc from connect", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+
+	if (response.properties != NULL)
+	{
+		MQTTProperty* prop = MQTTProperties_getProperty(response.properties, MQTTPROPERTY_CODE_ASSIGNED_CLIENT_IDENTIFIER);
+		assert("Server assigned a client identifier", prop != NULL, "prop was %p\n", (void*)prop);
+		if (prop != NULL)
+			assert("Assigned client identifier is non-empty", prop->value.data.len > 0, "len was %d\n", prop->value.data.len);
+	}
+	else
+		assert("CONNACK properties should be present", 0, "properties was NULL\n", 0);
+
+	MQTTClient_disconnect(c, 1000);
+	MQTTClient_destroy(&c);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test5: a second client connecting with the same client id as an existing,
+       still-connected client causes the server to disconnect the first
+       [MQTT5-3.1.4-3]
+
+*********************************************************************/
+static int duplicate_clientid_a_disconnected = 0;
+
+void duplicate_clientid_disconnected(void* context, MQTTProperties* properties, enum MQTTReasonCodes reasonCode)
+{
+	MyLog(LOGA_DEBUG, "Callback: disconnected, reason code %d", reasonCode);
+	duplicate_clientid_a_disconnected = 1;
+}
+
+/* a is put into asynchronous mode (via setCallbacks) so that its background thread
+ * actually processes the broker's unsolicited DISCONNECT when b takes over its session;
+ * this messageArrived callback is otherwise unused here */
+int duplicate_clientid_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message)
+{
+	MQTTClient_free(topicName);
+	MQTTClient_freeMessage(&message);
+	return 1;
+}
+
+int test_duplicate_clientid(struct Options options)
+{
+	int rc;
+	MQTTClient a, b;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+	const char* clientId = "conformance_dup_clientid";
+
+	MyLog(LOGA_INFO, "Starting test - duplicate client id disconnects the old client");
+	failures = 0;
+
+	createOpts.MQTTVersion = MQTTVERSION_5;
+	rc = MQTTClient_createWithOptions(&a, options.connection, clientId, MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
+	assert("Good rc from create (a)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_createWithOptions(&b, options.connection, clientId, MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
+	assert("Good rc from create (b)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	opts.keepAliveInterval = 20;
+	opts.cleanstart = 1;
+	opts.MQTTVersion = MQTTVERSION_5;
+
+	duplicate_clientid_a_disconnected = 0;
+	rc = MQTTClient_setCallbacks(a, NULL, NULL, duplicate_clientid_messageArrived, NULL);
+	assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_setDisconnected(a, NULL, duplicate_clientid_disconnected);
+	assert("Good rc from setDisconnected", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	response = MQTTClient_connect5(a, &opts, NULL, NULL);
+	assert("Good rc from connect (a)", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+	assert("a is connected", MQTTClient_isConnected(a) == 1, "isConnected was %d\n", MQTTClient_isConnected(a));
+
+	response = MQTTClient_connect5(b, &opts, NULL, NULL);
+	assert("Good rc from connect (b)", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+
+	{
+		int count = 0;
+		while (duplicate_clientid_a_disconnected == 0 && ++count < 30)
+			mysleep(100);
+	}
+
+	assert("a's disconnected callback fired when b connected with the same client id",
+			duplicate_clientid_a_disconnected == 1, "a_disconnected was %d\n", duplicate_clientid_a_disconnected);
+	assert("b is still connected", MQTTClient_isConnected(b) == 1, "isConnected(b) was %d\n", MQTTClient_isConnected(b));
+
+	MQTTClient_disconnect(b, 1000);
+	MQTTClient_destroy(&a);
+	MQTTClient_destroy(&b);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test6: publishing a zero-byte retained message deletes an existing retained
+       message on that topic
+       [MQTT-3.3.1-11]
+
+*********************************************************************/
+static int zero_byte_retained_messages_arrived = 0;
+
+int zero_byte_retained_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message)
+{
+	MyLog(LOGA_DEBUG, "Callback: message received on topic %s, %d bytes, retained %d",
+			topicName, message->payloadlen, message->retained);
+	++zero_byte_retained_messages_arrived;
+	MQTTClient_free(topicName);
+	MQTTClient_freeMessage(&message);
+	return 1;
+}
+
+int test_zero_byte_retained_delete(struct Options options)
+{
+	int rc;
+	MQTTClient pub, sub;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_message pubmsg = MQTTClient_message_initializer;
+	const char* topic = "conformance/zero_byte_retained";
+	int count;
+
+	MyLog(LOGA_INFO, "Starting test - zero byte retained message deletes retained message");
+	failures = 0;
+	zero_byte_retained_messages_arrived = 0;
+
+	rc = MQTTClient_create(&pub, options.connection, "conformance_zbr_pub", MQTTCLIENT_PERSISTENCE_NONE, NULL);
+	assert("Good rc from create (pub)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	opts.keepAliveInterval = 20;
+	opts.MQTTVersion = MQTTVERSION_3_1_1;
+	opts.cleansession = 1;
+	opts.cleanstart = 0;
+	rc = MQTTClient_connect(pub, &opts);
+	assert("Good rc from connect (pub)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* store a real retained message */
+	pubmsg.payload = "a retained message";
+	pubmsg.payloadlen = (int)strlen(pubmsg.payload);
+	pubmsg.qos = 1;
+	pubmsg.retained = 1;
+	rc = MQTTClient_publishMessage(pub, topic, &pubmsg, NULL);
+	assert("Good rc from publish of retained message", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	mysleep(500);
+
+	/* now delete it with a zero-byte retained publish */
+	pubmsg.payload = NULL;
+	pubmsg.payloadlen = 0;
+	rc = MQTTClient_publishMessage(pub, topic, &pubmsg, NULL);
+	assert("Good rc from publish of zero-byte retained message", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	mysleep(500);
+
+	/* a fresh subscriber should now get nothing retained on this topic */
+	rc = MQTTClient_create(&sub, options.connection, "conformance_zbr_sub", MQTTCLIENT_PERSISTENCE_NONE, NULL);
+	assert("Good rc from create (sub)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_setCallbacks(sub, NULL, NULL, zero_byte_retained_messageArrived, NULL);
+	assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_connect(sub, &opts);
+	assert("Good rc from connect (sub)", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_subscribe(sub, topic, 1);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	count = 0;
+	while (zero_byte_retained_messages_arrived == 0 && ++count < 20)
+		mysleep(100);
+
+	assert("No retained message should have been delivered after deletion",
+			zero_byte_retained_messages_arrived == 0, "messages arrived %d\n", zero_byte_retained_messages_arrived);
+
+	MQTTClient_disconnect(pub, 1000);
+	MQTTClient_disconnect(sub, 1000);
+	MQTTClient_destroy(&pub);
+	MQTTClient_destroy(&sub);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test7: unsubscribing from a topic the client was never subscribed to still
+       results in a normal (non-error) unsuback
+       [MQTT-3.10.4-5]
+
+*********************************************************************/
+int test_unsubscribe_no_match(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+	MQTTResponse unsub_response = MQTTResponse_initializer;
+
+	MyLog(LOGA_INFO, "Starting test - unsubscribe with no matching subscription");
+	failures = 0;
+
+	createOpts.MQTTVersion = MQTTVERSION_5;
+	rc = MQTTClient_createWithOptions(&c, options.connection, "conformance_unsub_nomatch", MQTTCLIENT_PERSISTENCE_NONE,
+			NULL, &createOpts);
+	assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	opts.keepAliveInterval = 20;
+	opts.cleanstart = 1;
+	opts.MQTTVersion = MQTTVERSION_5;
+	response = MQTTClient_connect5(c, &opts, NULL, NULL);
+	assert("Good rc from connect", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+
+	/* [MQTT-3.10.4-5] the server must respond to an unsubscribe of a topic with no matching
+	 * subscription with an UNSUBACK, using reason code 0x11 (No subscription existed) */
+	unsub_response = MQTTClient_unsubscribe5(c, "conformance/never_subscribed_to_this", NULL);
+	assert("Reason code is no subscription existed", unsub_response.reasonCode == MQTTREASONCODE_NO_SUBSCRIPTION_FOUND,
+			"reasonCode was %d\n", unsub_response.reasonCode);
+
+	MQTTClient_disconnect(c, 1000);
+	MQTTClient_destroy(&c);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test8: a message published with the retained flag set, while a subscriber is
+       already subscribed to that topic, is stored as the new retained message
+       and delivered without the retained flag set on the live forward
+       [MQTT-2.1.2-6] / [MQTT-2.1.2-10]
+
+*********************************************************************/
+static int retained_flag_messages_arrived = 0;
+
+int retained_flag_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message)
+{
+	MyLog(LOGA_DEBUG, "Callback: message received on topic %s, retained %d", topicName, message->retained);
+	++retained_flag_messages_arrived;
+	MQTTClient_free(topicName);
+	MQTTClient_freeMessage(&message);
+	return 1;
+}
+
+int test_retained_flag_live_publish(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_message pubmsg = MQTTClient_message_initializer;
+	const char* topic = "conformance/retained_flag_live";
+	int count;
+
+	MyLog(LOGA_INFO, "Starting test - retain flag on a live publish to an existing subscriber");
+	failures = 0;
+	retained_flag_messages_arrived = 0;
+
+	rc = MQTTClient_create(&c, options.connection, "conformance_retain_live", MQTTCLIENT_PERSISTENCE_NONE, NULL);
+	assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	rc = MQTTClient_setCallbacks(c, NULL, NULL, retained_flag_messageArrived, NULL);
+	assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+	opts.keepAliveInterval = 20;
+	opts.MQTTVersion = MQTTVERSION_3_1_1;
+	opts.cleansession = 1;
+	opts.cleanstart = 0;
+	rc = MQTTClient_connect(c, &opts);
+	assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* subscribe first ... */
+	rc = MQTTClient_subscribe(c, topic, 1);
+	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	/* ... then publish retained, while still subscribed: a live forward, not a retained replay */
+	pubmsg.payload = "live retained publish";
+	pubmsg.payloadlen = (int)strlen(pubmsg.payload);
+	pubmsg.qos = 1;
+	pubmsg.retained = 1;
+	rc = MQTTClient_publishMessage(c, topic, &pubmsg, NULL);
+	assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	count = 0;
+	while (retained_flag_messages_arrived == 0 && ++count < 50)
+		mysleep(100);
+
+	assert("The live retained publish was delivered", retained_flag_messages_arrived == 1,
+			"messages arrived %d\n", retained_flag_messages_arrived);
+
+	MQTTClient_disconnect(c, 1000);
+	MQTTClient_destroy(&c);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test9: a CONNECT with a password but no username is not client-side validated
+       by this library, so it reaches the broker as a non-compliant packet
+       [MQTT-3.1.2-22] / [MQTT5-3.1.2-22]
+
+*********************************************************************/
+int test_password_without_username(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+	int MQTTVersions[] = { MQTTVERSION_3_1_1, MQTTVERSION_5 };
+	size_t i;
+
+	MyLog(LOGA_INFO, "Starting test - password without username");
+	failures = 0;
+
+	for (i = 0; i < ARRAY_SIZE(MQTTVersions); ++i)
+	{
+		createOpts.MQTTVersion = MQTTVersions[i];
+		rc = MQTTClient_createWithOptions(&c, options.connection, "conformance_pw_no_user", MQTTCLIENT_PERSISTENCE_NONE,
+				NULL, &createOpts);
+		assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+		opts.keepAliveInterval = 20;
+		opts.MQTTVersion = MQTTVersions[i];
+		opts.cleansession = (MQTTVersions[i] >= MQTTVERSION_5) ? 0 : 1;
+		opts.cleanstart = (MQTTVersions[i] >= MQTTVERSION_5) ? 1 : 0;
+		opts.username = NULL;
+		opts.password = "somepassword";
+
+		if (MQTTVersions[i] >= MQTTVERSION_5)
+			response = MQTTClient_connect5(c, &opts, NULL, NULL);
+		else
+		{
+			rc = MQTTClient_connect(c, &opts);
+			response.reasonCode = (enum MQTTReasonCodes)rc;
+		}
+		/* the broker treats this as a malformed CONNECT and drops the connection without
+		 * ever sending a CONNACK, so the client should report the connect as unsuccessful */
+		MyLog(LOGA_INFO, "connect with password but no username returned reasonCode %d for MQTT version %d",
+				response.reasonCode, MQTTVersions[i]);
+		assert("Connect did not succeed", response.reasonCode != MQTTREASONCODE_SUCCESS,
+				"reasonCode was %d\n", response.reasonCode);
+
+		MQTTClient_disconnect(c, 0);
+		MQTTClient_destroy(&c);
+	}
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+/*********************************************************************
+
+Test10: a publish with more than one user property causes the broker to
+        check that it must maintain the order of user properties
+        [MQTT-3.1.3-10]
+
+*********************************************************************/
+int test_user_property_order(struct Options options)
+{
+	int rc;
+	MQTTClient c;
+	MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
+	MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
+	MQTTResponse response = MQTTResponse_initializer;
+	MQTTProperties props = MQTTProperties_initializer;
+	MQTTProperty property;
+
+	MyLog(LOGA_INFO, "Starting test - user property order on publish");
+	failures = 0;
+
+	createOpts.MQTTVersion = MQTTVERSION_5;
+	rc = MQTTClient_createWithOptions(&c, options.connection, "conformance_user_props", MQTTCLIENT_PERSISTENCE_NONE,
+			NULL, &createOpts);
+	assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
+
+	opts.keepAliveInterval = 20;
+	opts.cleanstart = 1;
+	opts.MQTTVersion = MQTTVERSION_5;
+	response = MQTTClient_connect5(c, &opts, NULL, NULL);
+	assert("Good rc from connect", response.reasonCode == MQTTREASONCODE_SUCCESS, "rc was %d\n", response.reasonCode);
+
+	property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
+	property.value.data.data = "key1";
+	property.value.data.len = (int)strlen("key1");
+	property.value.value.data = "value1";
+	property.value.value.len = (int)strlen("value1");
+	MQTTProperties_add(&props, &property);
+
+	property.value.data.data = "key2";
+	property.value.value.data = "value2";
+	MQTTProperties_add(&props, &property);
+
+	response = MQTTClient_publish5(c, "conformance/user_property_order", (int)strlen("payload"), "payload", 0, 0, &props, NULL);
+	assert("Good rc from publish with multiple user properties", response.reasonCode == MQTTREASONCODE_SUCCESS,
+			"rc was %d\n", response.reasonCode);
+
+	MQTTProperties_free(&props);
+	MQTTClient_disconnect(c, 1000);
+	MQTTClient_destroy(&c);
+
+	MyLog(LOGA_INFO, "TEST: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", tests, failures);
+	return failures;
+}
+
+
+int main(int argc, char** argv)
+{
+	int rc = 0, i;
+	int (*testfns[])(struct Options) = { NULL,
+		test_subscribe_empty,
+		test_keepalive_ping,
+		test_server_keepalive_override,
+		test_zero_length_clientid,
+		test_duplicate_clientid,
+		test_zero_byte_retained_delete,
+		test_unsubscribe_no_match,
+		test_retained_flag_live_publish,
+		test_password_without_username,
+		test_user_property_order,
+	};
+
+	getopts(argc, argv);
+
+	if (options.test_no == 0)
+	{
+		for (options.test_no = 1; options.test_no < (int)ARRAY_SIZE(testfns); ++options.test_no)
+			rc += testfns[options.test_no](options);
+	}
+	else
+		rc = testfns[options.test_no](options);
+
+	if (rc == 0)
+		MyLog(LOGA_INFO, "verdict pass");
+	else
+		MyLog(LOGA_INFO, "verdict fail");
+
+	return rc;
+}

test/test_unit_coverage.c

@@ -0,0 +1,605 @@
+/*******************************************************************************
+ * Copyright (c) 2026 IBM Corp. and Ian Craggs
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    https://www.eclipse.org/legal/epl-2.0/
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * AI Disclosure: This file was partly AI-generated. The AI-generated
+ * portions are made available under CC0-1.0 and not subject to the
+ * project's licence. The human contributor has reviewed and verified
+ * that the code is correct.
+ *
+ * SPDX-License-Identifier: EPL-2.0 and CC0-1.0
+ *
+ * Contributors:
+ *.  Ian Craggs - initial implementation and documentation
+ *******************************************************************************/
+
+/**
+ * @file
+ * White-box unit tests for internal modules that don't need a live broker
+ * connection: Base64, Tree, Proxy (parsing only), MQTTProperties and utf-8.
+ * These call internal functions directly (not the public MQTTClient/MQTTAsync
+ * API), targeting error/edge paths that the broker-driven integration tests
+ * don't reach.
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+
+#include "MQTTAsync.h"
+#include "Base64.h"
+#include "Tree.h"
+#include "Proxy.h"
+#include "MQTTProperties.h"
+#include "MQTTPacket.h"
+#include "utf-8.h"
+#include "Heap.h" /* redefines malloc/free as mymalloc/myfree, matching how Proxy.c allocates */
+
+/* internal functions with external linkage that aren't declared in the public headers */
+Node* TreeFindContentIndex(Tree* aTree, void* key, int index);
+Node* TreeNextElementIndex(Tree* aTree, Node* curnode, int index);
+void* TreeRemoveIndex(Tree* aTree, void* content, int index);
+int MQTTProperty_write(char** pptr, MQTTProperty* prop);
+int MQTTProperty_read(MQTTProperty* prop, char** pptr, char* enddata);
+
+static unsigned int fails = 0u;
+
+#define TEST_EXPECT(name, x) \
+	if (!(x)) { fprintf(stderr, "failed test: %s (%s:%d)\n", name, __FILE__, __LINE__); ++fails; }
+
+
+/* ---------------------------------------------------------------------- */
+/* Base64 */
+/* ---------------------------------------------------------------------- */
+
+static void test_base64(void)
+{
+	char out[512];
+	b64_size_t r;
+
+	/* round trip, including edge lengths that leave 1 or 2 bytes of padding */
+	const char* inputs[] = { "", "p", "pa", "pah", "paho", "paho with websockets." };
+	int i;
+
+	for (i = 0; i < (int)(sizeof(inputs) / sizeof(inputs[0])); ++i)
+	{
+		char encoded[512];
+		char decoded[512];
+		b64_size_t inlen = (b64_size_t)strlen(inputs[i]);
+		b64_size_t enclen;
+
+		enclen = Base64_encode(encoded, sizeof(encoded), (const b64_data_t*)inputs[i], inlen);
+		TEST_EXPECT("base64 encode length matches Base64_encodeLength",
+				enclen == Base64_encodeLength((const b64_data_t*)inputs[i], inlen));
+
+		r = Base64_decode((b64_data_t*)decoded, sizeof(decoded), encoded, enclen);
+		TEST_EXPECT("base64 decode length matches Base64_decodeLength",
+				r == Base64_decodeLength(encoded, enclen));
+		TEST_EXPECT("base64 round trip content matches",
+				r == inlen && memcmp(decoded, inputs[i], inlen) == 0);
+	}
+
+#if !defined(_WIN32)
+	/* decode: input length not a multiple of 4 - too short to decode anything */
+	r = Base64_decode((b64_data_t*)out, sizeof(out), "cGE", 3);
+	TEST_EXPECT("base64 decode with in_len < 4 returns 0", r == 0);
+
+	/* decode: output buffer too small for the first byte only ("cGE=" -> "pa") */
+	r = Base64_decode((b64_data_t*)out, 1, "cGE=", 4);
+	TEST_EXPECT("base64 decode stops when out_len exhausted after 1st byte", r == 1);
+
+	/* decode: output buffer holds exactly 2 of the 3 decoded bytes ("cGFo" -> "pah") */
+	r = Base64_decode((b64_data_t*)out, 2, "cGFo", 4);
+	TEST_EXPECT("base64 decode stops when out_len exhausted after 2nd byte", r == 2);
+
+	/* decode: invalid character in the 3rd position of a quad */
+	r = Base64_decode((b64_data_t*)out, sizeof(out), "AB!D", 4);
+	TEST_EXPECT("base64 decode with invalid 3rd char stops at 1 byte", r == 1);
+
+	/* decode: invalid character in the 4th position of a quad */
+	r = Base64_decode((b64_data_t*)out, sizeof(out), "ABC!", 4);
+	TEST_EXPECT("base64 decode with invalid 4th char stops at 2 bytes", r == 2);
+#endif
+
+	/* decodeLength / encodeLength edge cases (NULL / short inputs) */
+	TEST_EXPECT("base64 decodeLength NULL input", Base64_decodeLength(NULL, 4) == 3);
+	TEST_EXPECT("base64 decodeLength zero length", Base64_decodeLength("cGE=", 0) == 0);
+	TEST_EXPECT("base64 decodeLength length 1, no padding check on 2nd-last byte",
+			Base64_decodeLength("c", 1) == 0);
+	TEST_EXPECT("base64 decodeLength one padding char", Base64_decodeLength("cGE=", 4) == 2);
+	TEST_EXPECT("base64 decodeLength two padding chars", Base64_decodeLength("cA==", 4) == 1);
+	TEST_EXPECT("base64 decodeLength no padding", Base64_decodeLength("cGFo", 4) == 3);
+}
+
+
+/* ---------------------------------------------------------------------- */
+/* Tree */
+/* ---------------------------------------------------------------------- */
+
+static void test_tree(void)
+{
+	Tree* t;
+	Tree stackTree;
+	Node* n;
+	int values[] = { 5, 3, 8, 1, 4, 7, 9, 2, 6, 0 };
+	int i, prev, count;
+	const char* strs[] = { "banana", "apple", "cherry" };
+
+	/* direct compare function checks (note: these return -1 when the first arg is
+	 * the *greater* one - an inverted convention relative to strcmp, but self-consistent
+	 * for how the tree uses it) */
+	{
+		int a = 1, b = 2;
+		TEST_EXPECT("TreeIntCompare with a < b", TreeIntCompare(&a, &b, 0) == 1);
+		TEST_EXPECT("TreeIntCompare with a > b", TreeIntCompare(&b, &a, 0) == -1);
+		TEST_EXPECT("TreeIntCompare equal", TreeIntCompare(&a, &a, 0) == 0);
+	}
+	{
+		int a = 1, b = 2;
+		TEST_EXPECT("TreePtrCompare with a > b", TreePtrCompare(&a, &b, 0) == (&a > &b ? -1 : 1));
+		TEST_EXPECT("TreePtrCompare equal", TreePtrCompare(&a, &a, 0) == 0);
+	}
+	TEST_EXPECT("TreeStringCompare less", TreeStringCompare("apple", "banana", 0) < 0);
+	TEST_EXPECT("TreeStringCompare equal", TreeStringCompare("apple", "apple", 0) == 0);
+	TEST_EXPECT("TreeStringCompare greater", TreeStringCompare("banana", "apple", 0) > 0);
+
+	/* dynamically allocated tree, single index, iteration in sorted order */
+	t = TreeInitialize(TreeIntCompare);
+	TEST_EXPECT("TreeInitialize returns non-NULL", t != NULL);
+
+	for (i = 0; i < (int)(sizeof(values) / sizeof(values[0])); ++i)
+		TreeAdd(t, &values[i], sizeof(int));
+	TEST_EXPECT("Tree count after adds", t->count == (int)(sizeof(values) / sizeof(values[0])));
+
+	n = TreeFind(t, &values[0]);
+	TEST_EXPECT("TreeFind locates an added value", n != NULL && *(int*)(n->content) == values[0]);
+
+	n = NULL;
+	prev = -1;
+	count = 0;
+	while ((n = TreeNextElement(t, n)) != NULL)
+	{
+		int v = *(int*)(n->content);
+		TEST_EXPECT("TreeNextElement returns ascending order", v > prev);
+		prev = v;
+		++count;
+	}
+	TEST_EXPECT("TreeNextElement visited every node", count == (int)(sizeof(values) / sizeof(values[0])));
+
+	{
+		int key = 4;
+		void* removed = TreeRemove(t, &key);
+		TEST_EXPECT("TreeRemove finds and removes an existing key", removed != NULL);
+		n = TreeFind(t, &key);
+		TEST_EXPECT("TreeFind no longer finds removed key", n == NULL);
+	}
+
+	TreeFree(t);
+
+	/* stack-allocated tree (no malloc), second index, both indexes ordering the
+	 * same string content (content type must be consistent across all indexes
+	 * added via TreeAddIndex, since TreeAdd inserts into every index) */
+	TreeInitializeNoMalloc(&stackTree, TreeStringCompare);
+	TreeAddIndex(&stackTree, TreeStringCompare);
+	TEST_EXPECT("TreeAddIndex increments indexes", stackTree.indexes == 2);
+
+	for (i = 0; i < (int)(sizeof(strs) / sizeof(strs[0])); ++i)
+		TreeAdd(&stackTree, (void*)strs[i], strlen(strs[i]) + 1);
+
+	n = TreeFindIndex(&stackTree, (void*)"apple", 1);
+	TEST_EXPECT("TreeFindIndex on second index finds content",
+			n != NULL && strcmp((char*)n->content, "apple") == 0);
+
+	n = TreeFindContentIndex(&stackTree, (void*)"banana", 1);
+	TEST_EXPECT("TreeFindContentIndex on second index finds content",
+			n != NULL && strcmp((char*)n->content, "banana") == 0);
+
+	n = NULL;
+	count = 0;
+	while ((n = TreeNextElementIndex(&stackTree, n, 1)) != NULL)
+		++count;
+	TEST_EXPECT("TreeNextElementIndex walks all nodes on second index",
+			count == (int)(sizeof(strs) / sizeof(strs[0])));
+
+	{
+		void* removed = TreeRemoveIndex(&stackTree, (void*)"cherry", 1);
+		TEST_EXPECT("TreeRemoveIndex removes via second index", removed != NULL);
+	}
+
+	/* allow_duplicates: when set, adding a key that already compares equal to an
+	 * existing one is silently skipped (unlike the default, which replaces the
+	 * existing node's content in place). Use a fresh, single-index int tree. */
+	{
+		Tree dupTree;
+		int dupkey1 = 42, dupkey2 = 42;
+		void* addResult;
+		Node* found;
+
+		TreeInitializeNoMalloc(&dupTree, TreeIntCompare);
+		dupTree.allow_duplicates = 1;
+
+		TreeAdd(&dupTree, &dupkey1, sizeof(int));
+		addResult = TreeAdd(&dupTree, &dupkey2, sizeof(int));
+
+		TEST_EXPECT("allow_duplicates skips inserting a second equal key",
+				dupTree.count == 1 && addResult == NULL);
+
+		found = TreeFind(&dupTree, &dupkey2);
+		TEST_EXPECT("allow_duplicates leaves the original entry's content in place",
+				found != NULL && found->content == &dupkey1);
+	}
+}
+
+
+/* ---------------------------------------------------------------------- */
+/* Proxy (pure parsing functions only - no live connection) */
+/* ---------------------------------------------------------------------- */
+
+static void test_proxy(void)
+{
+	char no_proxy_buf[256];
+
+	strcpy(no_proxy_buf, "example.com");
+	TEST_EXPECT("Proxy_noProxy exact match returns 0 (don't use proxy)",
+			Proxy_noProxy("example.com", no_proxy_buf) == 0);
+
+	strcpy(no_proxy_buf, "example.com");
+	TEST_EXPECT("Proxy_noProxy suffix match returns 0",
+			Proxy_noProxy("sub.example.com", no_proxy_buf) == 0);
+
+	strcpy(no_proxy_buf, "example.com");
+	TEST_EXPECT("Proxy_noProxy non-matching host returns 1 (use proxy)",
+			Proxy_noProxy("notexample.com", no_proxy_buf) == 1);
+
+	strcpy(no_proxy_buf, "*");
+	TEST_EXPECT("Proxy_noProxy wildcard matches anything",
+			Proxy_noProxy("anything.example.org", no_proxy_buf) == 0);
+
+	strcpy(no_proxy_buf, "foo.com,example.com");
+	TEST_EXPECT("Proxy_noProxy matches second entry in comma list",
+			Proxy_noProxy("example.com", no_proxy_buf) == 0);
+
+	strcpy(no_proxy_buf, "foo.com,bar.com");
+	TEST_EXPECT("Proxy_noProxy with no match at all returns 1",
+			Proxy_noProxy("example.com", no_proxy_buf) == 1);
+
+	/* Proxy_setHTTPProxy: no auth */
+	{
+		char source[] = "http://myproxy.example.com:8080/";
+		char* dest = NULL;
+		char* auth_dest = NULL;
+		int rc = Proxy_setHTTPProxy(NULL, source, &dest, &auth_dest, "http://");
+
+		TEST_EXPECT("Proxy_setHTTPProxy without auth succeeds", rc == 0);
+		TEST_EXPECT("Proxy_setHTTPProxy strips the prefix",
+				dest != NULL && strcmp(dest, "myproxy.example.com:8080/") == 0);
+		TEST_EXPECT("Proxy_setHTTPProxy leaves auth NULL when none is given", auth_dest == NULL);
+	}
+
+	/* Proxy_setHTTPProxy: with user:pass auth, verify the encoded auth matches
+	 * an independently computed Base64 encoding of "user:pass" */
+	{
+		char source[] = "http://user:pass@myproxy.example.com:8080/";
+		char* dest = NULL;
+		char* auth_dest = NULL;
+		char expected[64];
+		b64_size_t explen;
+		int rc;
+
+		explen = Base64_encode(expected, sizeof(expected), (const b64_data_t*)"user:pass", 9);
+		expected[explen] = '\0';
+
+		rc = Proxy_setHTTPProxy(NULL, source, &dest, &auth_dest, "http://");
+
+		TEST_EXPECT("Proxy_setHTTPProxy with auth succeeds", rc == 0);
+		TEST_EXPECT("Proxy_setHTTPProxy dest starts after the '@'",
+				dest != NULL && strcmp(dest, "myproxy.example.com:8080/") == 0);
+		TEST_EXPECT("Proxy_setHTTPProxy encodes auth as expected",
+				auth_dest != NULL && strcmp(auth_dest, expected) == 0);
+		free(auth_dest);
+	}
+
+	/* Proxy_setHTTPProxy: NULL source is a safe no-op */
+	{
+		char* dest = (char*)0x1; /* sentinel: should remain untouched */
+		char* auth_dest = NULL;
+		int rc = Proxy_setHTTPProxy(NULL, NULL, &dest, &auth_dest, "http://");
+
+		TEST_EXPECT("Proxy_setHTTPProxy with NULL source returns success", rc == 0);
+		TEST_EXPECT("Proxy_setHTTPProxy with NULL source doesn't touch dest",
+				dest == (char*)0x1);
+	}
+}
+
+
+/* ---------------------------------------------------------------------- */
+/* utf-8 */
+/* ---------------------------------------------------------------------- */
+
+static void test_utf8(void)
+{
+	/* 1-byte ASCII */
+	TEST_EXPECT("utf8 1-byte ascii valid", UTF8_validate(1, "A") == 1);
+
+	/* 2-byte: (c) copyright sign, 0xC2 0xA9 */
+	TEST_EXPECT("utf8 2-byte valid", UTF8_validate(2, "\xC2\xA9") == 1);
+
+	/* 3-byte, each of the four sub-ranges in valid_ranges */
+	TEST_EXPECT("utf8 3-byte valid (E0 sub-range)", UTF8_validate(3, "\xE0\xA0\x80") == 1);
+	TEST_EXPECT("utf8 3-byte valid (E1-EC sub-range)", UTF8_validate(3, "\xE1\x80\x80") == 1);
+	TEST_EXPECT("utf8 3-byte valid (ED sub-range, excludes surrogates)",
+			UTF8_validate(3, "\xED\x80\x80") == 1);
+	TEST_EXPECT("utf8 3-byte valid (EE-EF sub-range)", UTF8_validate(3, "\xEE\x80\x80") == 1);
+
+	/* 4-byte, each of the three sub-ranges */
+	TEST_EXPECT("utf8 4-byte valid (F0 sub-range)", UTF8_validate(4, "\xF0\x90\x80\x80") == 1);
+	TEST_EXPECT("utf8 4-byte valid (F1-F3 sub-range)", UTF8_validate(4, "\xF1\x80\x80\x80") == 1);
+	TEST_EXPECT("utf8 4-byte valid (F4 sub-range)", UTF8_validate(4, "\xF4\x80\x80\x80") == 1);
+
+	/* truncated multi-byte sequences: not enough bytes for the declared length */
+	TEST_EXPECT("utf8 truncated 3-byte sequence is invalid", UTF8_validate(2, "\xE0\xA0") == 0);
+	TEST_EXPECT("utf8 truncated 4-byte sequence is invalid", UTF8_validate(3, "\xF0\x90\x80") == 0);
+
+	/* invalid continuation byte within a valid lead byte's range */
+	TEST_EXPECT("utf8 invalid 2-byte continuation is invalid", UTF8_validate(2, "\xC2\x20") == 0);
+	TEST_EXPECT("utf8 invalid 3-byte continuation is invalid", UTF8_validate(3, "\xE0\x20\x80") == 0);
+
+	/* whole-string validation, embedding a valid multi-byte character */
+	TEST_EXPECT("UTF8_validateString accepts a valid multi-byte string",
+			UTF8_validateString("caf\xC3\xA9") == 1);
+	TEST_EXPECT("UTF8_validateString rejects an invalid byte sequence",
+			UTF8_validateString("caf\xFF\xFF") == 0);
+}
+
+
+/* ---------------------------------------------------------------------- */
+/* MQTTProperties */
+/* ---------------------------------------------------------------------- */
+
+static void test_mqtt_properties(void)
+{
+	MQTTProperties props = MQTTProperties_initializer;
+	MQTTProperty prop;
+	int rc;
+
+	/* invalid property identifier is rejected */
+	memset(&prop, 0, sizeof(prop));
+	prop.identifier = (enum MQTTPropertyCodes)9999;
+	rc = MQTTProperties_add(&props, &prop);
+	TEST_EXPECT("MQTTProperties_add rejects an unknown property id", rc == MQTT_INVALID_PROPERTY_ID);
+
+	/* BYTE-type property: write then read back */
+	{
+		char buf[16];
+		char* wptr = buf;
+		char* rptr = buf;
+		MQTTProperty byteProp;
+		MQTTProperty readProp;
+
+		memset(&byteProp, 0, sizeof(byteProp));
+		byteProp.identifier = MQTTPROPERTY_CODE_MAXIMUM_QOS;
+		byteProp.value.byte = 2;
+
+		MQTTProperty_write(&wptr, &byteProp);
+		rc = MQTTProperty_read(&readProp, &rptr, buf + sizeof(buf));
+		TEST_EXPECT("MQTTProperty_read of a BYTE property succeeds", rc > 0);
+		TEST_EXPECT("MQTTProperty_read of a BYTE property returns the value written",
+				readProp.identifier == MQTTPROPERTY_CODE_MAXIMUM_QOS && readProp.value.byte == 2);
+	}
+
+	/* truncated buffers: not enough data for each property type */
+	{
+		char buf[8];
+		char* pptr;
+		MQTTProperty readProp;
+
+		/* BYTE: identifier present, but no value byte */
+		pptr = buf;
+		buf[0] = (char)MQTTPROPERTY_CODE_MAXIMUM_QOS;
+		rc = MQTTProperty_read(&readProp, &pptr, buf + 1);
+		TEST_EXPECT("MQTTProperty_read truncated BYTE value fails", rc == -1);
+
+		/* TWO_BYTE_INTEGER: identifier present, only 1 of 2 value bytes */
+		pptr = buf;
+		buf[0] = (char)MQTTPROPERTY_CODE_TOPIC_ALIAS;
+		buf[1] = 0;
+		rc = MQTTProperty_read(&readProp, &pptr, buf + 2);
+		TEST_EXPECT("MQTTProperty_read truncated TWO_BYTE_INTEGER fails", rc == -1);
+
+		/* FOUR_BYTE_INTEGER: identifier present, only 2 of 4 value bytes */
+		pptr = buf;
+		buf[0] = (char)MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
+		buf[1] = 0;
+		buf[2] = 0;
+		rc = MQTTProperty_read(&readProp, &pptr, buf + 3);
+		TEST_EXPECT("MQTTProperty_read truncated FOUR_BYTE_INTEGER fails", rc == -1);
+
+		/* UTF-8 string: identifier present, length prefix says 5 bytes but none follow */
+		pptr = buf;
+		buf[0] = (char)MQTTPROPERTY_CODE_CONTENT_TYPE;
+		buf[1] = 0;
+		buf[2] = 5;
+		rc = MQTTProperty_read(&readProp, &pptr, buf + 3);
+		TEST_EXPECT("MQTTProperty_read truncated UTF-8 string fails", rc == -1);
+
+		/* just the 1-byte identifier, nothing else in the buffer at all */
+		pptr = buf;
+		buf[0] = (char)MQTTPROPERTY_CODE_MAXIMUM_QOS;
+		rc = MQTTProperty_read(&readProp, &pptr, buf);
+		TEST_EXPECT("MQTTProperty_read with zero remaining data fails", rc == -1);
+	}
+
+	/* MQTTProperties_read: declared remaining length longer than the buffer supplied */
+	{
+		char buf[8];
+		char* pptr = buf;
+		MQTTProperties readProps = MQTTProperties_initializer;
+
+		buf[0] = 100; /* VBI: claims 100 bytes of properties follow, buffer is much shorter */
+		rc = MQTTProperties_read(&readProps, &pptr, buf + 1);
+		TEST_EXPECT("MQTTProperties_read with an over-long declared length fails",
+				rc == MQTTPACKET_BUFFER_TOO_SHORT);
+	}
+
+	/* MQTTProperties_read: more than 10 properties, to exercise array growth (realloc) */
+	{
+		char buf[256];
+		char* wptr;
+		char* pptr;
+		MQTTProperties writeProps = MQTTProperties_initializer;
+		MQTTProperties readProps = MQTTProperties_initializer;
+		int i;
+		int numProps = 15;
+
+		for (i = 0; i < numProps; ++i)
+		{
+			MQTTProperty p;
+			memset(&p, 0, sizeof(p));
+			p.identifier = MQTTPROPERTY_CODE_SUBSCRIPTION_IDENTIFIER;
+			p.value.integer4 = (unsigned int)(i + 1);
+			rc = MQTTProperties_add(&writeProps, &p);
+			TEST_EXPECT("MQTTProperties_add succeeds while building the growth test list", rc == 0);
+		}
+
+		wptr = buf;
+		rc = MQTTProperties_write(&wptr, &writeProps);
+		TEST_EXPECT("MQTTProperties_write of >10 properties succeeds", rc > 0);
+
+		/* MQTTProperties_read decodes its own leading VBI length prefix, so pptr
+		 * must point at the very start of the buffer written by MQTTProperties_write */
+		pptr = buf;
+		rc = MQTTProperties_read(&readProps, &pptr, buf + sizeof(buf));
+		TEST_EXPECT("MQTTProperties_read of >10 properties succeeds", rc == 1);
+		TEST_EXPECT("MQTTProperties_read grows the array to hold all properties",
+				readProps.count == numProps && readProps.max_count >= numProps);
+
+		MQTTProperties_free(&writeProps);
+		MQTTProperties_free(&readProps);
+	}
+
+	/* MQTTProperties_read: a corrupt property (unknown identifier) mid-buffer -> MQTTPACKET_BAD */
+	{
+		char buf[8];
+		char* pptr = buf;
+		MQTTProperties readProps = MQTTProperties_initializer;
+
+		buf[0] = 1;    /* VBI: 1 byte of properties follows */
+		buf[1] = (char)0x7F; /* an identifier that isn't in the known property table */
+		rc = MQTTProperties_read(&readProps, &pptr, buf + 2);
+		TEST_EXPECT("MQTTProperties_read with an unknown property id fails", rc == MQTTPACKET_BAD);
+	}
+
+	/* MQTTProperties_copy: source list has a manually-injected invalid identifier,
+	 * which should be skipped/logged rather than crashing the copy */
+	{
+		MQTTProperties badProps = MQTTProperties_initializer;
+		MQTTProperties copied;
+		MQTTProperty p;
+
+		memset(&p, 0, sizeof(p));
+		p.identifier = MQTTPROPERTY_CODE_MAXIMUM_QOS;
+		p.value.byte = 1;
+		MQTTProperties_add(&badProps, &p);
+
+		/* bypass validation: hand-corrupt the stored identifier after adding it */
+		badProps.array[0].identifier = (enum MQTTPropertyCodes)9999;
+
+		copied = MQTTProperties_copy(&badProps);
+		TEST_EXPECT("MQTTProperties_copy with a corrupt source entry doesn't crash and copies nothing",
+				copied.count == 0);
+
+		badProps.array[0].identifier = MQTTPROPERTY_CODE_MAXIMUM_QOS; /* restore so free() is safe */
+		MQTTProperties_free(&badProps);
+		MQTTProperties_free(&copied);
+	}
+
+	/* MQTTProperties_getNumericValueAt: BYTE type, and the "unsupported type" default */
+	{
+		MQTTProperties numProps = MQTTProperties_initializer;
+		MQTTProperty p;
+		int64_t v;
+
+		memset(&p, 0, sizeof(p));
+		p.identifier = MQTTPROPERTY_CODE_MAXIMUM_QOS; /* BYTE type */
+		p.value.byte = 7;
+		MQTTProperties_add(&numProps, &p);
+
+		v = MQTTProperties_getNumericValue(&numProps, MQTTPROPERTY_CODE_MAXIMUM_QOS);
+		TEST_EXPECT("MQTTProperties_getNumericValue reads a BYTE property", v == 7);
+
+		memset(&p, 0, sizeof(p));
+		p.identifier = MQTTPROPERTY_CODE_CONTENT_TYPE; /* UTF-8 string: unsupported for numeric read */
+		p.value.data.data = (char*)"text/plain";
+		p.value.data.len = (int)strlen("text/plain");
+		MQTTProperties_add(&numProps, &p);
+
+		v = MQTTProperties_getNumericValue(&numProps, MQTTPROPERTY_CODE_CONTENT_TYPE);
+		TEST_EXPECT("MQTTProperties_getNumericValue on a non-numeric type returns the default", v == -999999);
+
+		MQTTProperties_free(&numProps);
+	}
+
+	/* MQTTProperties_getPropertyAt: index > 0 to find the second occurrence */
+	{
+		MQTTProperties userProps = MQTTProperties_initializer;
+		MQTTProperty p;
+		MQTTProperty* found;
+
+		memset(&p, 0, sizeof(p));
+		p.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
+		p.value.data.data = (char*)"k1";
+		p.value.data.len = 2;
+		p.value.value.data = (char*)"v1";
+		p.value.value.len = 2;
+		MQTTProperties_add(&userProps, &p);
+
+		p.value.data.data = (char*)"k2";
+		p.value.value.data = (char*)"v2";
+		MQTTProperties_add(&userProps, &p);
+
+		found = MQTTProperties_getPropertyAt(&userProps, MQTTPROPERTY_CODE_USER_PROPERTY, 1);
+		TEST_EXPECT("MQTTProperties_getPropertyAt(index=1) finds the second occurrence",
+				found != NULL && strncmp(found->value.data.data, "k2", 2) == 0);
+
+		MQTTProperties_free(&userProps);
+	}
+
+	MQTTProperties_free(&props);
+}
+
+
+int main(int argc, char** argv)
+{
+	(void)argc;
+	(void)argv;
+
+	/* initialize heap tracking / mutexes used internally by these modules,
+	 * without needing an actual network connection. MQTTAsync_global_init()
+	 * alone isn't enough: Heap_initialize() is normally only called lazily from
+	 * MQTTAsync_createWithOptions, so it's called explicitly here instead. */
+	MQTTAsync_global_init(NULL);
+#if !defined(HIGH_PERFORMANCE)
+	Heap_initialize();
+#endif
+
+	test_base64();
+	test_tree();
+	test_proxy();
+	test_utf8();
+	test_mqtt_properties();
+
+	if (fails)
+		printf("%u tests failed\n", fails);
+	else
+		printf("all tests passed\n");
+
+	return (int)fails;
+}