natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Check incoming and outgoing max queued QoS > 0 messages

This commit is contained in:
Ian Craggs 2026-07-13 18:37:54 +01:00
parent 2260f5a5e0
commit 8e67525301
No known key found for this signature in database
GPG Key ID: A7AE1A8F2CCAB186
8 changed files with 92 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/Clients.h
View File

@ -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 */

2
src/MQTTAsync.c
View File

@ -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

9
src/MQTTAsync.h
View File

@ -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;
/**

9
src/MQTTAsyncUtils.c
View File

@ -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

14
src/MQTTClient.c
View File

@ -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)

22
src/MQTTPacketOut.c
View File

@ -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;
}

8
src/MQTTProperties.c
View File

@ -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;

35
src/MQTTProtocolClient.c
View File

@ -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);