diff --git a/build.xml b/build.xml index a7fc86fc..5478b9b0 100644 --- a/build.xml +++ b/build.xml @@ -24,7 +24,7 @@ - + diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index f90db203..e1c67a83 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -1558,7 +1558,6 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) else if (pack->header.bits.type == SUBACK) { ListElement* current = NULL; - int handleCalled = 0; /* use the msgid to find the callback to be called */ while (ListNextElement(m->responses, ¤t)) @@ -1566,12 +1565,30 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) MQTTAsync_queuedCommand* command = (MQTTAsync_queuedCommand*)(current->content); if (command->command.token == ((Suback*)pack)->msgId) { + Suback* sub = (Suback*)pack; if (!ListDetach(m->responses, command)) /* remove the response from the list */ Log(LOG_ERROR, -1, "Subscribe command not removed from command list"); - if (command->command.onSuccess) + + /* Call the failure callback if there is one subscribe in the MQTT packet and + * the return code is 0x80 (failure). If the MQTT packet contains >1 subscription + * request, then we call onSuccess with the list of returned QoSs, which inelegantly, + * could include some failures, or worse, the whole list could have failed. + */ + if (sub->qoss->count == 1 && *(int*)(sub->qoss->first->content) == MQTT_BAD_SUBSCRIBE) + { + if (command->command.onFailure) + { + MQTTAsync_failureData data; + + data.token = command->command.token; + data.code = *(int*)(sub->qoss->first->content); + Log(TRACE_MIN, -1, "Calling subscribe failure for client %s", m->c->clientID); + (*(command->command.onFailure))(command->command.context, &data); + } + } + else if (command->command.onSuccess) { MQTTAsync_successData data; - Suback* sub = (Suback*)pack; int* array = NULL; if (sub->qoss->count == 1) @@ -1584,9 +1601,6 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) *element++ = *(int*)(cur_qos->content); } data.token = command->command.token; - - rc = MQTTProtocol_handleSubacks(pack, m->c->net.socket); - handleCalled = 1; Log(TRACE_MIN, -1, "Calling subscribe success for client %s", m->c->clientID); (*(command->command.onSuccess))(command->command.context, &data); if (array) @@ -1596,8 +1610,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) break; } } - if (!handleCalled) - rc = MQTTProtocol_handleSubacks(pack, m->c->net.socket); + rc = MQTTProtocol_handleSubacks(pack, m->c->net.socket); } else if (pack->header.bits.type == UNSUBACK) { diff --git a/test/test_mqtt4async.c b/test/test_mqtt4async.c index 8393a189..603abd55 100644 --- a/test/test_mqtt4async.c +++ b/test/test_mqtt4async.c @@ -68,7 +68,7 @@ struct Options int iterations; } options = { - "tcp://m2m.eclipse.org:1883", + "m2m.eclipse.org:1883", NULL, 0, 0, @@ -289,7 +289,7 @@ void test1_onConnect3(void* context, MQTTAsync_successData* response) opts.onSuccess = test1_onDisconnect3; opts.context = c; - assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0, + assert("Correct serverURI returned", strstr(response->alt.connect.serverURI, options.connection) != NULL, "serverURI was %s", response->alt.connect.serverURI); assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4, "MQTTVersion was %d", response->alt.connect.MQTTVersion); @@ -383,8 +383,6 @@ void test1_onConnect1(void* context, MQTTAsync_successData* response) opts.onSuccess = test1_onDisconnect1; opts.context = c; - printf("response %p\n", response); - assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0, "serverURI was %s", response->alt.connect.serverURI); assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4, @@ -457,105 +455,147 @@ exit: return failures; } -#if 0 + +void test2_onDisconnect(void* context, MQTTAsync_successData* response) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c); + test_finished = 1; +} + + +void test2_onSubscribe2(void* context, MQTTAsync_failureData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer; + int rc; + + MyLog(LOGA_DEBUG, "In subscribe onFailure callback, context %p", context); + + assert("Correct subscribe return code", response->code == MQTT_BAD_SUBSCRIBE, + "qos was %d", response->code); + + opts.onSuccess = test2_onDisconnect; + rc = MQTTAsync_disconnect(c, &opts); + assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + + +void test2_onSubscribe1(void* context, MQTTAsync_successData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback, context %p", context); + + assert("Correct subscribe return code", response->alt.qos == 2, + "qos was %d", response->alt.qos); +} + + +void test2_onConnect1(void* context, MQTTAsync_successData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback 1, context %p", context); + + assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0, + "serverURI was %s", response->alt.connect.serverURI); + assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4, + "MQTTVersion was %d", response->alt.connect.MQTTVersion); + assert("Correct sessionPresent returned", response->alt.connect.sessionPresent == 0, + "sessionPresent was %d", response->alt.connect.sessionPresent); + + MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context); + opts.context = c; + + opts.onSuccess = test2_onSubscribe1; + rc = MQTTAsync_subscribe(c, "a topic I can subscribe to", 2, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test_finished = 1; + + opts.onSuccess = NULL; + opts.onFailure = test2_onSubscribe2; + rc = MQTTAsync_subscribe(c, "nosubscribe", 2, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test_finished = 1; +} + + + /********************************************************************* -Test2: 0x80 return code from subscribe +Test1: 0x80 from subscribe *********************************************************************/ -volatile int test2_arrivedcount = 0; -int test2_deliveryCompleted = 0; -MQTTClient_message test2_pubmsg = MQTTClient_message_initializer; - -void test2_deliveryComplete(void* context, MQTTClient_deliveryToken dt) -{ - ++test2_deliveryCompleted; -} - -int test2_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m) -{ - ++test2_arrivedcount; - MyLog(LOGA_DEBUG, "Callback: %d message received on topic %s is %.*s.", - test2_arrivedcount, topicName, m->payloadlen, (char*)(m->payload)); - MQTTClient_free(topicName); - MQTTClient_freeMessage(&m); - return 1; -} - int test2(struct Options options) { - char* testname = "test2"; - int subsqos = 2; - MQTTClient c; - MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer; + MQTTAsync c; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer; int rc = 0; - char* test_topic = "C client test2"; - char* topics[2] = {"test_topic", "nosubscribe"}; - int qoss[2] = {2, 2}; + char* test_topic = "C client test1"; - fprintf(xml, "