mirror of https://github.com/eclipse/paho.mqtt.c
Better control of sending DISCONNECT + reason code
Use maxInflight messages to control incoming window for MQTT 3.1.1
This commit is contained in:
Ian Craggs
parent
1086e2374f
commit
bc20f4d739
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -1640,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);
|
||||
|
|
@ -1685,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;
|
||||
|
|
@ -1755,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;
|
||||
}
|
||||
|
||||
|
|
@ -2115,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
|
||||
{
|
||||
|
|
@ -2206,7 +2210,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
|
|||
}
|
||||
else
|
||||
{
|
||||
nextOrClose(m, rc, "CONNACK return code");
|
||||
nextOrClose(m, rc, "CONNACK return code", 0);
|
||||
}
|
||||
MQTTPacket_freeConnack(connack);
|
||||
}
|
||||
|
|
@ -2374,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
|
||||
{
|
||||
|
|
@ -2450,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;
|
||||
|
|
@ -2459,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);
|
||||
|
|
@ -2481,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))
|
||||
|
|
@ -2768,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);
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -3047,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;
|
||||
|
|
@ -3098,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)
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
@ -1969,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;
|
||||
|
|
@ -2000,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)
|
||||
|
|
@ -2021,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);
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -2041,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;
|
||||
}
|
||||
|
|
@ -2052,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;
|
||||
}
|
||||
|
|
@ -2195,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;
|
||||
|
||||
|
|
@ -2345,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)
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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 */
|
||||
{
|
||||
|
|
@ -361,18 +366,20 @@ int MQTTProtocol_handlePublishes(void* pack, SOCKET sock)
|
|||
|
||||
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);
|
||||
MQTTPacket_send_disconnect(client, MQTTREASONCODE_RECEIVE_MAXIMUM_EXCEEDED, NULL);
|
||||
client->good = 0;
|
||||
MQTTProtocol_closeSession(client, 1);
|
||||
rc = SOCKET_ERROR;
|
||||
MQTTProtocol_closeSession(client, MQTTREASONCODE_RECEIVE_MAXIMUM_EXCEEDED, 1);
|
||||
}
|
||||
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);
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -761,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 &&
|
||||
|
|
@ -773,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))
|
||||
|
|
@ -784,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
|
||||
{
|
||||
|
|
@ -810,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
|
||||
{
|
||||
|
|
@ -875,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
|
||||
|
|
@ -893,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
|
||||
|
|
@ -956,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)
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Reference in New Issue