diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 688c5425..7001bd56 100644 --- a/src/MQTTAsync.c +++ b/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); diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c index 7610dc88..6b1c41f6 100644 --- a/src/MQTTAsyncUtils.c +++ b/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) @@ -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) diff --git a/src/MQTTAsyncUtils.h b/src/MQTTAsyncUtils.h index 3debd1aa..2a0419fb 100644 --- a/src/MQTTAsyncUtils.h +++ b/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); diff --git a/src/MQTTClient.c b/src/MQTTClient.c index 104bd70a..cfb94cf2 100644 --- a/src/MQTTClient.c +++ b/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); @@ -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); diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index 7e99cc77..79f8c46f 100644 --- a/src/MQTTProtocolClient.c +++ b/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 */ { @@ -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) diff --git a/src/MQTTProtocolClient.h b/src/MQTTProtocolClient.h index cf41e56c..6e0234f6 100644 --- a/src/MQTTProtocolClient.h +++ b/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);