From af104622cf9324365076e34ee05a3b1621a46d10 Mon Sep 17 00:00:00 2001 From: Keith Holman Date: Wed, 7 Feb 2018 14:30:57 -0500 Subject: [PATCH] clients: add defines for connect states To help understand the various connecting states this patch introduces defines with the various states. This helps make the code a bit easier to understand. Signed-off-by: Keith Holman --- src/Clients.h | 13 +++++++++++ src/MQTTAsync.c | 50 +++++++++++++++++++++---------------------- src/MQTTClient.c | 46 +++++++++++++++++++-------------------- src/MQTTProtocolOut.c | 9 ++++---- 4 files changed, 65 insertions(+), 53 deletions(-) diff --git a/src/Clients.h b/src/Clients.h index 02ed23aa..0f85e145 100644 --- a/src/Clients.h +++ b/src/Clients.h @@ -160,6 +160,19 @@ typedef struct #endif } networkHandles; + +/* connection states */ +/** no connection in progress, see connected value */ +#define NOT_IN_PROGRESS 0x0 +/** TCP connection in progress */ +#define TCP_IN_PROGRESS 0x1 +/** SSL connection in progress */ +#define SSL_IN_PROGRESS 0x2 +/** TCP completed, waiting for MQTT ACK */ +#define WAIT_FOR_CONNACK 0x3 +/** Disconnecting */ +#define DISCONNECTING -2 + /** * Data related to one client */ diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index cc06582e..e0ab02b5 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -878,7 +878,7 @@ static int MQTTAsync_addCommand(MQTTAsync_queuedCommand* command, int command_si FUNC_ENTRY; MQTTAsync_lock_mutex(mqttcommand_mutex); /* Don't set start time if the connect command is already in process #218 */ - if ((command->command.type != CONNECT) || (command->client->c->connect_state == 0)) + if ((command->command.type != CONNECT) || (command->client->c->connect_state == NOT_IN_PROGRESS)) command->command.start_time = MQTTAsync_start_clock(); if (command->command.type == CONNECT || (command->command.type == DISCONNECT && command->command.details.dis.internal)) @@ -1192,7 +1192,7 @@ static int MQTTAsync_processCommand(void) continue; if (cmd->command.type == CONNECT || cmd->command.type == DISCONNECT || (cmd->client->c->connected && - cmd->client->c->connect_state == 0 && MQTTAsync_Socket_noPendingWrites(cmd->client->c->net.socket))) + cmd->client->c->connect_state == NOT_IN_PROGRESS && MQTTAsync_Socket_noPendingWrites(cmd->client->c->net.socket))) { if ((cmd->command.type == PUBLISH || cmd->command.type == SUBSCRIBE || cmd->command.type == UNSUBSCRIBE) && cmd->client->c->outboundMsgs->count >= MAX_MSG_ID - 1) @@ -1223,7 +1223,7 @@ static int MQTTAsync_processCommand(void) if (command->command.type == CONNECT) { - if (command->client->c->connect_state != 0 || command->client->c->connected) + if (command->client->c->connect_state != NOT_IN_PROGRESS || command->client->c->connected) rc = 0; else { @@ -1260,7 +1260,7 @@ static int MQTTAsync_processCommand(void) #else rc = MQTTProtocol_connect(serverURI, command->client->c, command->command.details.conn.MQTTVersion); #endif - if (command->client->c->connect_state == 0) + if (command->client->c->connect_state == NOT_IN_PROGRESS) rc = SOCKET_ERROR; /* if the TCP connect is pending, then we must call select to determine when the connect has completed, @@ -1342,11 +1342,11 @@ static int MQTTAsync_processCommand(void) } else if (command->command.type == DISCONNECT) { - if (command->client->c->connect_state != 0 || command->client->c->connected != 0) + if (command->client->c->connect_state != NOT_IN_PROGRESS || command->client->c->connected != 0) { - if (command->client->c->connect_state != 0) + if (command->client->c->connect_state != NOT_IN_PROGRESS) { - command->client->c->connect_state = -2; + command->client->c->connect_state = DISCONNECTING; if (command->client->connect.onFailure) { MQTTAsync_failureData data; @@ -1502,11 +1502,11 @@ static void MQTTAsync_checkTimeouts(void) MQTTAsyncs* m = (MQTTAsyncs*)(current->content); /* check disconnect timeout */ - if (m->c->connect_state == -2) + if (m->c->connect_state == DISCONNECTING) MQTTAsync_checkDisconnect(m, &m->disconnect); /* check connect timeout */ - if (m->c->connect_state != 0 && MQTTAsync_elapsed(m->connect.start_time) > (m->connectTimeout * 1000)) + if (m->c->connect_state != NOT_IN_PROGRESS && MQTTAsync_elapsed(m->connect.start_time) > (m->connectTimeout * 1000)) { nextOrClose(m, MQTTASYNC_FAILURE, "TCP connect timeout"); continue; @@ -1756,7 +1756,7 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack) int rc = MQTTASYNC_FAILURE; FUNC_ENTRY; - if (m->c->connect_state == 3) /* MQTT connect sent - wait for CONNACK */ + if (m->c->connect_state == WAIT_FOR_CONNACK) /* MQTT connect sent - wait for CONNACK */ { Connack* connack = (Connack*)pack; Log(LOG_PROTOCOL, 1, NULL, m->c->net.socket, m->c->clientID, connack->rc); @@ -1765,7 +1765,7 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack) m->retrying = 0; m->c->connected = 1; m->c->good = 1; - m->c->connect_state = 0; + m->c->connect_state = NOT_IN_PROGRESS; if (m->c->cleansession) rc = MQTTAsync_cleanSession(m->c); if (m->c->outboundMsgs->count > 0) @@ -1844,7 +1844,7 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n) MQTTAsync_disconnect_internal(m, 0); MQTTAsync_lock_mutex(mqttasync_mutex); } - else if (m->c->connect_state != 0) + else if (m->c->connect_state != NOT_IN_PROGRESS) nextOrClose(m, rc, "socket error"); else /* calling disconnect_internal won't have any effect if we're already disconnected */ MQTTAsync_closeOnly(m->c); @@ -2032,7 +2032,7 @@ static void MQTTAsync_stop(void) /* find out how many handles are still connected */ while (ListNextElement(handles, ¤t)) { - if (((MQTTAsyncs*)(current->content))->c->connect_state > 0 || + if (((MQTTAsyncs*)(current->content))->c->connect_state > NOT_IN_PROGRESS || ((MQTTAsyncs*)(current->content))->c->connected) ++conn_count; } @@ -2069,7 +2069,7 @@ int MQTTAsync_setCallbacks(MQTTAsync handle, void* context, FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); - if (m == NULL || ma == NULL || m->c->connect_state != 0) + if (m == NULL || ma == NULL || m->c->connect_state != NOT_IN_PROGRESS) rc = MQTTASYNC_FAILURE; else { @@ -2093,7 +2093,7 @@ int MQTTAsync_setConnected(MQTTAsync handle, void* context, MQTTAsync_connected* FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); - if (m == NULL || m->c->connect_state != 0) + if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS) rc = MQTTASYNC_FAILURE; else { @@ -2129,7 +2129,7 @@ static void MQTTAsync_closeOnly(Clients* client) Thread_unlock_mutex(socket_mutex); } client->connected = 0; - client->connect_state = 0; + client->connect_state = NOT_IN_PROGRESS; FUNC_EXIT; } @@ -2907,7 +2907,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) int rc = -1; FUNC_ENTRY; - if (m->c->connect_state == 1) /* TCP connect started - check for completion */ + if (m->c->connect_state == TCP_IN_PROGRESS) /* TCP connect started - check for completion */ { int error; socklen_t len = sizeof(error); @@ -2942,7 +2942,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) if (rc == TCPSOCKET_INTERRUPTED) { rc = MQTTCLIENT_SUCCESS; /* the connect is still in progress */ - m->c->connect_state = 2; + m->c->connect_state = SSL_IN_PROGRESS; } else if (rc == SSL_FATAL) { @@ -2952,7 +2952,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) else if (rc == 1) { rc = MQTTCLIENT_SUCCESS; - m->c->connect_state = 3; + m->c->connect_state = WAIT_FOR_CONNACK; if (MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion) == SOCKET_ERROR) { rc = SOCKET_ERROR; @@ -2971,7 +2971,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) else { #endif - m->c->connect_state = 3; /* TCP/SSL connect completed, in which case send the MQTT connect packet */ + m->c->connect_state = WAIT_FOR_CONNACK; /* TCP/SSL connect completed, in which case send the MQTT connect packet */ if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion)) == SOCKET_ERROR) goto exit; #if defined(OPENSSL) @@ -2979,7 +2979,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) #endif } #if defined(OPENSSL) - else if (m->c->connect_state == 2) /* SSL connect sent - wait for completion */ + else if (m->c->connect_state == SSL_IN_PROGRESS) /* SSL connect sent - wait for completion */ { if ((rc = SSLSocket_connect(m->c->net.ssl, m->c->net.socket, m->serverURI, m->c->sslopts->verify)) != 1) @@ -2987,14 +2987,14 @@ static int MQTTAsync_connecting(MQTTAsyncs* m) if(!m->c->cleansession && m->c->session == NULL) m->c->session = SSL_get1_session(m->c->net.ssl); - m->c->connect_state = 3; /* SSL connect completed, in which case send the MQTT connect packet */ + m->c->connect_state = WAIT_FOR_CONNACK; /* SSL connect completed, in which case send the MQTT connect packet */ if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion)) == SOCKET_ERROR) goto exit; } #endif exit: - if ((rc != 0 && rc != TCPSOCKET_INTERRUPTED && m->c->connect_state != 2) || (rc == SSL_FATAL)) + if ((rc != 0 && rc != TCPSOCKET_INTERRUPTED && m->c->connect_state != SSL_IN_PROGRESS) || (rc == SSL_FATAL)) nextOrClose(m, MQTTASYNC_FAILURE, "TCP/TLS connect failure"); FUNC_EXIT_RC(rc); @@ -3035,11 +3035,11 @@ static MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc) if (m != NULL) { Log(TRACE_MINIMUM, -1, "m->c->connect_state = %d",m->c->connect_state); - if (m->c->connect_state == 1 || m->c->connect_state == 2) + if (m->c->connect_state == TCP_IN_PROGRESS || m->c->connect_state == SSL_IN_PROGRESS) *rc = MQTTAsync_connecting(m); else pack = MQTTPacket_Factory(&m->c->net, rc); - if (m->c->connect_state == 3 && *rc == SOCKET_ERROR) + 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"); diff --git a/src/MQTTClient.c b/src/MQTTClient.c index eb832ed1..7de3f89b 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -592,12 +592,12 @@ static thread_return_type WINAPI MQTTClient_run(void* n) MQTTClient_disconnect_internal(m, 0); else { - if (m->c->connect_state == 2 && !Thread_check_sem(m->connect_sem)) + if (m->c->connect_state == SSL_IN_PROGRESS && !Thread_check_sem(m->connect_sem)) { Log(TRACE_MIN, -1, "Posting connect semaphore for client %s", m->c->clientID); Thread_post_sem(m->connect_sem); } - if (m->c->connect_state == 3 && !Thread_check_sem(m->connack_sem)) + if (m->c->connect_state == WAIT_FOR_CONNACK && !Thread_check_sem(m->connack_sem)) { Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID); Thread_post_sem(m->connack_sem); @@ -656,7 +656,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n) Thread_post_sem(m->unsuback_sem); } } - else if (m->c->connect_state == 1 && !Thread_check_sem(m->connect_sem)) + else if (m->c->connect_state == TCP_IN_PROGRESS && !Thread_check_sem(m->connect_sem)) { int error; socklen_t len = sizeof(error); @@ -667,7 +667,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n) Thread_post_sem(m->connect_sem); } #if defined(OPENSSL) - else if (m->c->connect_state == 2 && !Thread_check_sem(m->connect_sem)) + else if (m->c->connect_state == SSL_IN_PROGRESS && !Thread_check_sem(m->connect_sem)) { rc = SSLSocket_connect(m->c->net.ssl, m->c->net.socket, m->serverURI, m->c->sslopts->verify); @@ -706,7 +706,7 @@ static void MQTTClient_stop(void) /* find out how many handles are still connected */ while (ListNextElement(handles, ¤t)) { - if (((MQTTClients*)(current->content))->c->connect_state > 0 || + if (((MQTTClients*)(current->content))->c->connect_state > NOT_IN_PROGRESS || ((MQTTClients*)(current->content))->c->connected) ++conn_count; } @@ -743,7 +743,7 @@ int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClient_connect FUNC_ENTRY; Thread_lock_mutex(mqttclient_mutex); - if (m == NULL || ma == NULL || m->c->connect_state != 0) + if (m == NULL || ma == NULL || m->c->connect_state != NOT_IN_PROGRESS) rc = MQTTCLIENT_FAILURE; else { @@ -780,7 +780,7 @@ static void MQTTClient_closeSession(Clients* client) #endif } client->connected = 0; - client->connect_state = 0; + client->connect_state = NOT_IN_PROGRESS; if (client->cleansession) MQTTClient_cleanSession(client); @@ -877,13 +877,13 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt if (rc == SOCKET_ERROR) goto exit; - if (m->c->connect_state == 0) + if (m->c->connect_state == NOT_IN_PROGRESS) { rc = SOCKET_ERROR; goto exit; } - if (m->c->connect_state == 1) /* TCP connect started - wait for completion */ + if (m->c->connect_state == TCP_IN_PROGRESS) /* TCP connect started - wait for completion */ { Thread_unlock_mutex(mqttclient_mutex); MQTTClient_waitfor(handle, CONNECT, &rc, millisecsTimeout - MQTTClient_elapsed(start)); @@ -914,7 +914,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt rc = SSLSocket_connect(m->c->net.ssl, m->c->net.socket, m->serverURI, m->c->sslopts->verify); if (rc == TCPSOCKET_INTERRUPTED) - m->c->connect_state = 2; /* the connect is still in progress */ + m->c->connect_state = SSL_IN_PROGRESS; /* the connect is still in progress */ else if (rc == SSL_FATAL) { rc = SOCKET_ERROR; @@ -923,7 +923,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt else if (rc == 1) { rc = MQTTCLIENT_SUCCESS; - m->c->connect_state = 3; + m->c->connect_state = WAIT_FOR_CONNACK; if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR) { rc = SOCKET_ERROR; @@ -942,7 +942,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt else { #endif - m->c->connect_state = 3; /* TCP connect completed, in which case send the MQTT connect packet */ + m->c->connect_state = WAIT_FOR_CONNACK; /* TCP connect completed, in which case send the MQTT connect packet */ if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR) { rc = SOCKET_ERROR; @@ -954,7 +954,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt } #if defined(OPENSSL) - if (m->c->connect_state == 2) /* SSL connect sent - wait for completion */ + if (m->c->connect_state == SSL_IN_PROGRESS) /* SSL connect sent - wait for completion */ { Thread_unlock_mutex(mqttclient_mutex); MQTTClient_waitfor(handle, CONNECT, &rc, millisecsTimeout - MQTTClient_elapsed(start)); @@ -966,7 +966,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt } if(!m->c->cleansession && m->c->session == NULL) m->c->session = SSL_get1_session(m->c->net.ssl); - m->c->connect_state = 3; /* TCP connect completed, in which case send the MQTT connect packet */ + m->c->connect_state = WAIT_FOR_CONNACK; /* TCP connect completed, in which case send the MQTT connect packet */ if (MQTTPacket_send_connect(m->c, MQTTVersion) == SOCKET_ERROR) { rc = SOCKET_ERROR; @@ -975,7 +975,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt } #endif - if (m->c->connect_state == 3) /* MQTT connect sent - wait for CONNACK */ + if (m->c->connect_state == WAIT_FOR_CONNACK) /* MQTT connect sent - wait for CONNACK */ { MQTTPacket* pack = NULL; @@ -992,7 +992,7 @@ static int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOpt { m->c->connected = 1; m->c->good = 1; - m->c->connect_state = 0; + m->c->connect_state = NOT_IN_PROGRESS; if (MQTTVersion == 4) sessionPresent = connack->flags.bits.sessionPresent; if (m->c->cleansession) @@ -1300,7 +1300,7 @@ static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int call_conne rc = MQTTCLIENT_FAILURE; goto exit; } - if (m->c->connected == 0 && m->c->connect_state == 0) + if (m->c->connected == 0 && m->c->connect_state == NOT_IN_PROGRESS) { rc = MQTTCLIENT_DISCONNECTED; goto exit; @@ -1309,7 +1309,7 @@ static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int call_conne if (m->c->connected != 0) { start = MQTTClient_start_clock(); - m->c->connect_state = -2; /* indicate disconnecting */ + m->c->connect_state = DISCONNECTING; /* indicate disconnecting */ while (m->c->inboundMsgs->count > 0 || m->c->outboundMsgs->count > 0) { /* wait for all inflight message flows to finish, up to timeout */ if (MQTTClient_elapsed(start) >= timeout) @@ -1741,7 +1741,7 @@ static MQTTPacket* MQTTClient_cycle(int* sock, unsigned long timeout, int* rc) m = (MQTTClient)(handles->current->content); if (m != NULL) { - if (m->c->connect_state == 1 || m->c->connect_state == 2) + if (m->c->connect_state == TCP_IN_PROGRESS || m->c->connect_state == SSL_IN_PROGRESS) *rc = 0; /* waiting for connect state to clear */ else { @@ -1833,7 +1833,7 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r break; if (pack && (pack->header.bits.type == packet_type)) break; - if (m->c->connect_state == 1) + if (m->c->connect_state == TCP_IN_PROGRESS) { int error; socklen_t len = sizeof(error); @@ -1843,7 +1843,7 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r break; } #if defined(OPENSSL) - else if (m->c->connect_state == 2) + else if (m->c->connect_state == SSL_IN_PROGRESS) { *rc = SSLSocket_connect(m->c->net.ssl, sock, m->serverURI, m->c->sslopts->verify); @@ -1857,7 +1857,7 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r } } #endif - else if (m->c->connect_state == 3) + else if (m->c->connect_state == WAIT_FOR_CONNACK) { int error; socklen_t len = sizeof(error); @@ -1965,7 +1965,7 @@ void MQTTClient_yield(void) if (rc == SOCKET_ERROR && ListFindItem(handles, &sock, clientSockCompare)) { MQTTClients* m = (MQTTClient)(handles->current->content); - if (m->c->connect_state != -2) + if (m->c->connect_state != DISCONNECTING) MQTTClient_disconnect_internal(m, 0); } Thread_unlock_mutex(mqttclient_mutex); diff --git a/src/MQTTProtocolOut.c b/src/MQTTProtocolOut.c index ac3adc85..6378787a 100644 --- a/src/MQTTProtocolOut.c +++ b/src/MQTTProtocolOut.c @@ -108,7 +108,7 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersi addr = MQTTProtocol_addressPort(ip_address, &port); rc = Socket_new(addr, port, &(aClient->net.socket)); if (rc == EINPROGRESS || rc == EWOULDBLOCK) - aClient->connect_state = 1; /* TCP connect called - wait for connect completion */ + aClient->connect_state = TCP_IN_PROGRESS; /* TCP connect called - wait for connect completion */ else if (rc == 0) { /* TCP connect completed. If SSL, send SSL connect */ #if defined(OPENSSL) @@ -119,20 +119,19 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersi rc = SSLSocket_connect(aClient->net.ssl, aClient->net.socket, addr, aClient->sslopts->verify); if (rc == TCPSOCKET_INTERRUPTED) - aClient->connect_state = 2; /* SSL connect called - wait for completion */ + aClient->connect_state = SSL_IN_PROGRESS; /* SSL connect called - wait for completion */ } else rc = SOCKET_ERROR; } #endif - if (rc == 0) { /* Now send the MQTT connect packet */ if ((rc = MQTTPacket_send_connect(aClient, MQTTVersion)) == 0) - aClient->connect_state = 3; /* MQTT Connect sent - wait for CONNACK */ + aClient->connect_state = WAIT_FOR_CONNACK; /* MQTT Connect sent - wait for CONNACK */ else - aClient->connect_state = 0; + aClient->connect_state = NOT_IN_PROGRESS; } } if (addr != ip_address)