From 8e67525301ff4959b37be6241ad50cbc43701424 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Mon, 13 Jul 2026 18:37:54 +0100 Subject: [PATCH] Check incoming and outgoing max queued QoS > 0 messages --- src/Clients.h | 14 ++++++++++++-- src/MQTTAsync.c | 2 ++ src/MQTTAsync.h | 9 ++++++++- src/MQTTAsyncUtils.c | 9 +++------ src/MQTTClient.c | 14 +++++++------- src/MQTTPacketOut.c | 22 ++++++++++++++++++++++ src/MQTTProperties.c | 8 ++++---- src/MQTTProtocolClient.c | 35 ++++++++++++++++++++++++++++++++++- 8 files changed, 92 insertions(+), 21 deletions(-) diff --git a/src/Clients.h b/src/Clients.h index 6ad84376..1e42570e 100644 --- a/src/Clients.h +++ b/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 */ diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index d5bf929c..688c5425 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -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 diff --git a/src/MQTTAsync.h b/src/MQTTAsync.h index e3bf876c..69678a95 100644 --- a/src/MQTTAsync.h +++ b/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; /** diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c index 2754179d..7610dc88 100644 --- a/src/MQTTAsyncUtils.c +++ b/src/MQTTAsyncUtils.c @@ -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 */ @@ -2200,11 +2201,7 @@ 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 diff --git a/src/MQTTClient.c b/src/MQTTClient.c index b27cf874..104bd70a 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -1565,6 +1565,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 +1945,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: @@ -2414,8 +2412,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) diff --git a/src/MQTTPacketOut.c b/src/MQTTPacketOut.c index dc1b3797..6f3852b6 100644 --- a/src/MQTTPacketOut.c +++ b/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; } diff --git a/src/MQTTProperties.c b/src/MQTTProperties.c index 987875f6..ad5570c9 100644 --- a/src/MQTTProperties.c +++ b/src/MQTTProperties.c @@ -483,7 +483,7 @@ exit: } -MQTTProperties MQTTProperties_copy(const MQTTProperties* props) +MQTTProperties MQTTProperties_copy(const MQTTProperties *props) { int i = 0; MQTTProperties result = MQTTProperties_initializer; @@ -491,10 +491,10 @@ MQTTProperties MQTTProperties_copy(const MQTTProperties* props) FUNC_ENTRY; for (i = 0; props != NULL && i < props->count; ++i) { - int rc = 0; + int rc = 0; - if ((rc = MQTTProperties_add(&result, &props->array[i])) != 0) - Log(LOG_ERROR, -1, "Error from MQTTProperties add %d", rc); + if ((rc = MQTTProperties_add(&result, &props->array[i])) != 0) + Log(LOG_ERROR, -1, "Error from MQTTProperties add %d", rc); } FUNC_EXIT; diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index ba8ff3a4..7e99cc77 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -351,14 +351,45 @@ 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) + { + if (publish->MQTTVersion >= MQTTVERSION_5) + { + Log(TRACE_PROTOCOL, -1, "Receive maximum (%d) exceeded by server for client %s, disconnecting", + client->maxInflightMessages, clientid); + MQTTPacket_send_disconnect(client, MQTTREASONCODE_RECEIVE_MAXIMUM_EXCEEDED, NULL); + client->good = 0; + MQTTProtocol_closeSession(client, 1); + rc = SOCKET_ERROR; + } + else + Log(LOG_ERROR, -1, "Max inflight messages (%d) exceeded by server for client %s, ignoring incoming QoS %d publish msgid %d", + client->maxInflightMessages, clientid, publish->header.bits.qos, publish->msgId); + 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) { @@ -1061,6 +1092,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);