diff --git a/build.xml b/build.xml index 5b595fa5..27d89a0b 100644 --- a/build.xml +++ b/build.xml @@ -24,7 +24,7 @@ - + diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index cb1292f9..966091ba 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -22,6 +22,8 @@ * Ian Craggs - MQTT 3.1.1 support * Rong Xiang, Ian Craggs - C++ compatibility * Ian Craggs - fix for bug 442400: reconnecting after network cable unplugged + * Ian Craggs - fix for bug 444934 - incorrect free in freeCommand1 + * Ian Craggs - fix for bug 445891 - assigning msgid is not thread safe *******************************************************************************/ /** @@ -75,6 +77,8 @@ enum MQTTAsync_threadStates enum MQTTAsync_threadStates sendThread_state = STOPPED; enum MQTTAsync_threadStates receiveThread_state = STOPPED; +static thread_id_type sendThread_id = 0, + receiveThread_id = 0; #if defined(WIN32) || defined(WIN64) static mutex_type mqttasync_mutex = NULL; @@ -325,7 +329,7 @@ void MQTTAsync_lock_mutex(mutex_type amutex) { int rc = Thread_lock_mutex(amutex); if (rc != 0) - Log(LOG_ERROR, 0, "Error %d locking mutex", rc); + Log(LOG_ERROR, 0, "Error %s locking mutex", strerror(rc)); } @@ -333,7 +337,7 @@ void MQTTAsync_unlock_mutex(mutex_type amutex) { int rc = Thread_unlock_mutex(amutex); if (rc != 0) - Log(LOG_ERROR, 0, "Error %d unlocking mutex", rc); + Log(LOG_ERROR, 0, "Error %s unlocking mutex", strerror(rc)); } @@ -840,21 +844,19 @@ void MQTTAsync_freeCommand1(MQTTAsync_queuedCommand *command) int i; for (i = 0; i < command->command.details.sub.count; i++) - { free(command->command.details.sub.topics[i]); - free(command->command.details.sub.topics); - free(command->command.details.sub.qoss); - } + + free(command->command.details.sub.topics); + free(command->command.details.sub.qoss); } else if (command->command.type == UNSUBSCRIBE) { int i; for (i = 0; i < command->command.details.unsub.count; i++) - { free(command->command.details.unsub.topics[i]); - free(command->command.details.unsub.topics); - } + + free(command->command.details.unsub.topics); } else if (command->command.type == PUBLISH) { @@ -1255,6 +1257,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); sendThread_state = RUNNING; + sendThread_id = Thread_getid(); MQTTAsync_unlock_mutex(mqttasync_mutex); while (!tostop) { @@ -1281,6 +1284,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) sendThread_state = STOPPING; MQTTAsync_lock_mutex(mqttasync_mutex); sendThread_state = STOPPED; + sendThread_id = 0; MQTTAsync_unlock_mutex(mqttasync_mutex); FUNC_EXIT; return 0; @@ -1455,6 +1459,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); receiveThread_state = RUNNING; + receiveThread_id = Thread_getid(); while (!tostop) { int rc = SOCKET_ERROR; @@ -1671,6 +1676,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) } } receiveThread_state = STOPPED; + receiveThread_id = 0; MQTTAsync_unlock_mutex(mqttasync_mutex); #if !defined(WIN32) && !defined(WIN64) if (sendThread_state != STOPPED) @@ -2143,6 +2149,56 @@ int MQTTAsync_isConnected(MQTTAsync handle) } +int cmdMessageIDCompare(void* a, void* b) +{ + MQTTAsync_queuedCommand* cmd = (MQTTAsync_queuedCommand*)a; + return cmd->command.token == *(int*)b; +} + + +/** + * Assign a new message id for a client. Make sure it isn't already being used and does + * not exceed the maximum. + * @param m a client structure + * @return the next message id to use, or 0 if none available + */ +int MQTTAsync_assignMsgId(MQTTAsyncs* m) +{ + int start_msgid = m->c->msgID; + int msgid = start_msgid; + thread_id_type thread_id = 0; + int locked = 0; + + /* need to check: commands list and response list for a client */ + FUNC_ENTRY; + /* We might be called in a callback. In which case, this mutex will be already locked. */ + thread_id = Thread_getid(); + if (thread_id != sendThread_id && thread_id != receiveThread_id) + { + MQTTAsync_lock_mutex(mqttasync_mutex); + locked = 1; + } + + msgid = (msgid == MAX_MSG_ID) ? 1 : msgid + 1; + while (ListFindItem(commands, &msgid, cmdMessageIDCompare) || + ListFindItem(m->responses, &msgid, cmdMessageIDCompare)) + { + msgid = (msgid == MAX_MSG_ID) ? 1 : msgid + 1; + if (msgid == start_msgid) + { /* we've tried them all - none free */ + msgid = 0; + break; + } + } + if (msgid != 0) + m->c->msgID = msgid; + if (locked) + MQTTAsync_unlock_mutex(mqttasync_mutex); + FUNC_EXIT_RC(msgid); + return msgid; +} + + int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, int* qos, MQTTAsync_responseOptions* response) { MQTTAsyncs* m = handle; @@ -2175,7 +2231,7 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, int goto exit; } } - if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + if ((msgid = MQTTAsync_assignMsgId(m)) == 0) { rc = MQTTASYNC_NO_MORE_MSGIDS; goto exit; @@ -2248,7 +2304,7 @@ int MQTTAsync_unsubscribeMany(MQTTAsync handle, int count, char* const* topic, M goto exit; } } - if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + if ((msgid = MQTTAsync_assignMsgId(m)) == 0) { rc = MQTTASYNC_NO_MORE_MSGIDS; goto exit; @@ -2307,7 +2363,7 @@ int MQTTAsync_send(MQTTAsync handle, const char* destinationName, int payloadlen rc = MQTTASYNC_BAD_UTF8_STRING; else if (qos < 0 || qos > 2) rc = MQTTASYNC_BAD_QOS; - else if (qos > 0 && (msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + else if (qos > 0 && (msgid = MQTTAsync_assignMsgId(m)) == 0) rc = MQTTASYNC_NO_MORE_MSGIDS; if (rc != MQTTASYNC_SUCCESS) diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index 639f89a4..c2469367 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -16,6 +16,7 @@ * Ian Craggs - fix for bug 413429 - connectionLost not called * Ian Craggs - fix for bug 421103 - trying to write to same socket, in retry * Rong Xiang, Ian Craggs - C++ compatibility + * Ian Craggs - turn off DUP flag for PUBREL - MQTT 3.1.1 *******************************************************************************/ /** @@ -596,7 +597,7 @@ void MQTTProtocol_retries(time_t now, Clients* client, int regardless) else if (m->qos && m->nextMessageType == PUBCOMP) { Log(TRACE_MIN, 7, NULL, "PUBREL", client->clientID, client->net.socket, m->msgid); - if (MQTTPacket_send_pubrel(m->msgid, 1, &client->net, client->clientID) != TCPSOCKET_COMPLETE) + if (MQTTPacket_send_pubrel(m->msgid, 0, &client->net, client->clientID) != TCPSOCKET_COMPLETE) { client->good = 0; Log(TRACE_PROTOCOL, 29, NULL, client->clientID, client->net.socket, diff --git a/src/Socket.c b/src/Socket.c index 2a8869a2..8475ae2b 100644 --- a/src/Socket.c +++ b/src/Socket.c @@ -98,7 +98,7 @@ int Socket_error(char* aString, int sock) if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) { if (strcmp(aString, "shutdown") != 0 || (errno != ENOTCONN && errno != ECONNRESET)) - Log(LOG_ERROR, -1, "Socket error %s in %s for socket %d", strerror(errno), aString, sock); + Log(TRACE_MINIMUM, -1, "Socket error %s in %s for socket %d", strerror(errno), aString, sock); } FUNC_EXIT_RC(errno); return errno; diff --git a/test/test4.c b/test/test4.c index 47a606fb..31edbf48 100644 --- a/test/test4.c +++ b/test/test4.c @@ -1143,6 +1143,16 @@ Test7: Persistence char* test7_topic = "C client test7"; int test7_messageCount = 0; +void test7_onDisconnectFailure(void* context, MQTTAsync_failureData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In onDisconnect failure callback %p", c); + + assert("Successful disconnect", 0, "disconnect failed", 0); + + test_finished = 1; +} + void test7_onDisconnect(void* context, MQTTAsync_successData* response) { MQTTAsync c = (MQTTAsync)context; @@ -1170,7 +1180,6 @@ int test7_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync { MQTTAsync c = (MQTTAsync)context; static int message_count = 0; - int rc; MyLog(LOGA_DEBUG, "Test7: received message id %d", message->msgid); @@ -1212,6 +1221,24 @@ void test7_onConnect(void* context, MQTTAsync_successData* response) } +void test7_onConnectOnly(void* context, MQTTAsync_successData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer; + int rc; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context); + dopts.context = context; + dopts.timeout = 1000; + dopts.onSuccess = test7_onDisconnect; + rc = MQTTAsync_disconnect(c, &dopts); + + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test_finished = 1; +} + + /********************************************************************* Test7: Pending tokens @@ -1248,7 +1275,6 @@ int test7(struct Options options) assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); opts.keepAliveInterval = 20; - opts.cleansession = 0; opts.username = "testuser"; opts.password = "testpassword"; opts.MQTTVersion = options.MQTTVersion; @@ -1259,11 +1285,30 @@ int test7(struct Options options) opts.will->retained = 0; opts.will->topicName = "will topic"; opts.will = NULL; - opts.onSuccess = test7_onConnect; + opts.onFailure = NULL; opts.context = c; + opts.cleansession = 1; + opts.onSuccess = test7_onConnectOnly; + MyLog(LOGA_DEBUG, "Connecting to clean up"); + rc = MQTTAsync_connect(c, &opts); + rc = 0; + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + goto exit; + + while (!test_finished) + #if defined(WIN32) + Sleep(100); + #else + usleep(10000L); + #endif + + test_finished = 0; MyLog(LOGA_DEBUG, "Connecting"); + opts.cleansession = 0; + opts.onSuccess = test7_onConnect; rc = MQTTAsync_connect(c, &opts); rc = 0; assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); @@ -1304,6 +1349,7 @@ int test7(struct Options options) /* disconnect immediately without receiving the incoming messages */ dopts.timeout = 0; dopts.onSuccess = test7_onDisconnect; + dopts.context = c; MQTTAsync_disconnect(c, &dopts); /* now there should be "orphaned" publications */ while (!test_finished) @@ -1371,6 +1417,8 @@ int test7(struct Options options) assertions fail against Mosquitto - needs testing */ + dopts.onFailure = test7_onDisconnectFailure; + dopts.onSuccess = test7_onDisconnect; dopts.timeout = 1000; MQTTAsync_disconnect(c, &dopts);