From 296fe57cd2e8fbeb3fb00266ba2300aef4b561d7 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Wed, 5 Sep 2018 22:13:40 +0100 Subject: [PATCH] MQTT V5 persistence updates #534 --- Makefile | 2 +- src/MQTTAsync.c | 107 +- src/MQTTAsync.h | 8 +- src/MQTTClient.c | 26 +- src/MQTTClient.h | 4 + src/MQTTPacket.c | 26 +- src/MQTTPacket.h | 4 +- src/MQTTPacketOut.c | 8 +- src/MQTTPersistence.c | 176 ++- src/MQTTPersistence.h | 17 +- src/MQTTProtocolClient.c | 2 +- test/CMakeLists.txt | 50 + test/test1.c | 9 +- test/test10.c | 58 +- test/test15.c | 170 ++- test/test2.c | 4 +- test/test3.c | 2 +- test/test45.c | 83 +- test/test5.c | 2 +- test/test8.c | 35 +- test/test9.c | 4 +- test/test95.c | 2358 ++++++++++++++++++++++++++++++++++++++ 22 files changed, 2967 insertions(+), 188 deletions(-) create mode 100644 test/test95.c diff --git a/Makefile b/Makefile index 17c02947..064543ec 100755 --- a/Makefile +++ b/Makefile @@ -107,7 +107,7 @@ SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}} TEST_FILES_CS = test3 SYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_CS}} -TEST_FILES_A = test4 test45 test6 test9 test_mqtt4async test11 +TEST_FILES_A = test4 test45 test6 test9 test95 test_mqtt4async test11 ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}} TEST_FILES_AS = test5 diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 74bfdf89..b58995d3 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -379,7 +379,7 @@ static void MQTTAsync_terminate(void); #if !defined(NO_PERSISTENCE) static int MQTTAsync_unpersistCommand(MQTTAsync_queuedCommand* qcmd); static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd); -static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen); +static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen, int MQTTVersion); /*static void MQTTAsync_insertInOrder(List* list, void* content, int size);*/ static int MQTTAsync_restoreCommands(MQTTAsyncs* client); #endif @@ -571,6 +571,7 @@ int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const { m->createOptions = malloc(sizeof(MQTTAsync_createOptions)); memcpy(m->createOptions, options, sizeof(MQTTAsync_createOptions)); + m->c->MQTTVersion = options->MQTTVersion; } #if !defined(NO_PERSISTENCE) @@ -650,12 +651,15 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd) void** bufs = NULL; int bufindex = 0, i, nbufs = 0; char key[PERSISTENCE_MAX_KEY_LENGTH + 1]; + int props_allocated = 0; + int process = 1; FUNC_ENTRY; switch (command->type) { case SUBSCRIBE: - nbufs = 3 + (command->details.sub.count * 2); + nbufs = ((aclient->c->MQTTVersion >= MQTTVERSION_5) ? 4 : 3) + + (command->details.sub.count * 2); lens = (int*)malloc(nbufs * sizeof(int)); bufs = malloc(nbufs * sizeof(char *)); @@ -673,14 +677,31 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd) { bufs[bufindex] = command->details.sub.topics[i]; lens[bufindex++] = (int)strlen(command->details.sub.topics[i]) + 1; - bufs[bufindex] = &command->details.sub.qoss[i]; - lens[bufindex++] = sizeof(command->details.sub.qoss[i]); + + if (aclient->c->MQTTVersion < MQTTVERSION_5) + { + bufs[bufindex] = &command->details.sub.qoss[i]; + lens[bufindex++] = sizeof(command->details.sub.qoss[i]); + } + else + { + if (command->details.sub.count == 1) + { + bufs[bufindex] = &command->details.sub.opts; + lens[bufindex++] = sizeof(command->details.sub.opts); + } + else + { + bufs[bufindex] = &command->details.sub.optlist[i]; + lens[bufindex++] = sizeof(command->details.sub.optlist[i]); + } + } } - sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno); break; case UNSUBSCRIBE: - nbufs = 3 + command->details.unsub.count; + nbufs = ((aclient->c->MQTTVersion >= MQTTVERSION_5) ? 4 : 3) + + command->details.unsub.count; lens = (int*)malloc(nbufs * sizeof(int)); bufs = malloc(nbufs * sizeof(char *)); @@ -699,11 +720,10 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd) bufs[bufindex] = command->details.unsub.topics[i]; lens[bufindex++] = (int)strlen(command->details.unsub.topics[i]) + 1; } - sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno); break; case PUBLISH: - nbufs = 7; + nbufs = (aclient->c->MQTTVersion >= MQTTVERSION_5) ? 8 : 7; lens = (int*)malloc(nbufs * sizeof(int)); bufs = malloc(nbufs * sizeof(char *)); @@ -728,10 +748,27 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd) bufs[bufindex] = &command->details.pub.retained; lens[bufindex++] = sizeof(command->details.pub.retained); + break; - sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno); + default: + process = 0; break; } + if (aclient->c->MQTTVersion >= MQTTVERSION_5 && process) /* persist properties */ + { + int temp_len = 0; + char* ptr = NULL; + + temp_len = MQTTProperties_len(&command->properties); + ptr = bufs[bufindex] = malloc(temp_len); + props_allocated = bufindex; + rc = MQTTProperties_write(&ptr, &command->properties); + lens[bufindex++] = temp_len; + sprintf(key, "%s%u", PERSISTENCE_V5_COMMAND_KEY, ++aclient->command_seqno); + } + else + sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno); + if (nbufs > 0) { if ((rc = aclient->c->persistence->pput(aclient->c->phandle, key, nbufs, (char**)bufs, lens)) != 0) @@ -742,12 +779,14 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd) free(lens); if (bufs) free(bufs); + if (props_allocated > 0) + free(bufs[props_allocated]); FUNC_EXIT_RC(rc); return rc; } -static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen) +static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen, int MQTTVersion) { MQTTAsync_command* command = NULL; MQTTAsync_queuedCommand* qcommand = NULL; @@ -774,8 +813,11 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle if (command->details.sub.count > 0) { - command->details.sub.topics = (char **)malloc(sizeof(char *) * command->details.sub.count); + command->details.sub.topics = (char **)malloc(sizeof(char *) * command->details.sub.count); + if (MQTTVersion < MQTTVERSION_5) command->details.sub.qoss = (int *)malloc(sizeof(int) * command->details.sub.count); + else if (command->details.sub.count > 1) + command->details.sub.optlist = (MQTTSubscribe_options*)malloc(sizeof(MQTTSubscribe_options) * command->details.sub.count); } for (i = 0; i < command->details.sub.count; ++i) @@ -786,8 +828,24 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle strcpy(command->details.sub.topics[i], ptr); ptr += data_size; - command->details.sub.qoss[i] = *(int*)ptr; - ptr += sizeof(int); + if (MQTTVersion < MQTTVERSION_5) + { + command->details.sub.qoss[i] = *(int*)ptr; + ptr += sizeof(int); + } + else + { + if (command->details.sub.count == 1) + { + command->details.sub.opts = *(MQTTSubscribe_options*)ptr; + ptr += sizeof(MQTTSubscribe_options); + } + else + { + command->details.sub.optlist[i] = *(MQTTSubscribe_options*)ptr; + ptr += sizeof(MQTTSubscribe_options); + } + } } break; @@ -797,7 +855,7 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle if (command->details.unsub.count > 0) { - command->details.unsub.topics = (char **)malloc(sizeof(char *) * command->details.unsub.count); + command->details.unsub.topics = (char **)malloc(sizeof(char *) * command->details.unsub.count); } for (i = 0; i < command->details.unsub.count; ++i) @@ -836,6 +894,13 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle qcommand = NULL; } + if (qcommand != NULL && MQTTVersion >= MQTTVERSION_5 && + MQTTProperties_read(&command->properties, &ptr, buffer + buflen) != 1) + { + Log(LOG_ERROR, -1, "Error restoring properties from persistence"); + free(qcommand); + qcommand = NULL; + } FUNC_EXIT; return qcommand; @@ -876,18 +941,22 @@ static int MQTTAsync_restoreCommands(MQTTAsyncs* client) char *buffer = NULL; int buflen; - if (strncmp(msgkeys[i], PERSISTENCE_COMMAND_KEY, strlen(PERSISTENCE_COMMAND_KEY)) != 0) + if (strncmp(msgkeys[i], PERSISTENCE_COMMAND_KEY, strlen(PERSISTENCE_COMMAND_KEY)) != 0 && + strncmp(msgkeys[i], PERSISTENCE_V5_COMMAND_KEY, strlen(PERSISTENCE_V5_COMMAND_KEY)) != 0) { ; } else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0) { - MQTTAsync_queuedCommand* cmd = MQTTAsync_restoreCommand(buffer, buflen); + int MQTTVersion = + (strncmp(msgkeys[i], PERSISTENCE_V5_COMMAND_KEY, strlen(PERSISTENCE_V5_COMMAND_KEY)) == 0) + ? MQTTVERSION_5 : MQTTVERSION_3_1_1; + MQTTAsync_queuedCommand* cmd = MQTTAsync_restoreCommand(buffer, buflen, MQTTVersion); if (cmd) { cmd->client = client; - cmd->seqno = atoi(msgkeys[i]+2); + cmd->seqno = atoi(strchr(msgkeys[i], '-')+1); /* key format is tag'-'seqno */ MQTTPersistence_insertInOrder(commands, cmd, sizeof(MQTTAsync_queuedCommand)); free(buffer); client->command_seqno = max(client->command_seqno, cmd->seqno); @@ -1395,7 +1464,7 @@ static int MQTTAsync_processCommand(void) rc = MQTTProtocol_subscribe(command->client->c, topics, qoss, command->command.token, subopts, props); ListFreeNoContent(topics); ListFreeNoContent(qoss); - if (command->command.details.sub.count > 1) + if (command->client->c->MQTTVersion >= MQTTVERSION_5 && command->command.details.sub.count > 1) free(command->command.details.sub.optlist); } else if (command->command.type == UNSUBSCRIBE) @@ -3393,7 +3462,7 @@ int MQTTAsync_sendMessage(MQTTAsync handle, const char* destinationName, const M goto exit; } - if (m->c->MQTTVersion >= MQTTVERSION_5) + if (m->c->MQTTVersion >= MQTTVERSION_5 && response) response->properties = message->properties; rc = MQTTAsync_send(handle, destinationName, message->payloadlen, message->payload, diff --git a/src/MQTTAsync.h b/src/MQTTAsync.h index 34b3b5ab..132df346 100644 --- a/src/MQTTAsync.h +++ b/src/MQTTAsync.h @@ -816,15 +816,19 @@ typedef struct { /** The eyecatcher for this structure. must be MQCO. */ char struct_id[4]; - /** The version number of this structure. Must be 0 */ + /** The version number of this structure. Must be 0 or 1 + * 0 means no MQTTVersion + */ int struct_version; /** Whether to allow messages to be sent when the client library is not connected. */ int sendWhileDisconnected; /** the maximum number of messages allowed to be buffered while not connected. */ int maxBufferedMessages; + /** Whether the MQTT version is 3 and 4, or 5. To use 5, this must be set. */ + int MQTTVersion; } MQTTAsync_createOptions; -#define MQTTAsync_createOptions_initializer { {'M', 'Q', 'C', 'O'}, 0, 0, 100 } +#define MQTTAsync_createOptions_initializer { {'M', 'Q', 'C', 'O'}, 0, 0, 100, MQTTVERSION_DEFAULT } DLLExport int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const char* clientId, diff --git a/src/MQTTClient.c b/src/MQTTClient.c index 853d3e15..75c45538 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -1493,10 +1493,17 @@ static MQTTResponse MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectO return rc; } +MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions* options, + MQTTProperties* connectProperties, MQTTProperties* willProperties); int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options) { - MQTTResponse response = MQTTClient_connect5(handle, options, NULL, NULL); + MQTTClients* m = handle; + + if (m->c->MQTTVersion >= MQTTVERSION_5) + return MQTTCLIENT_WRONG_MQTT_VERSION; + + MQTTResponse response = MQTTClient_connectAll(handle, options, NULL, NULL); return response.reasonCode; } @@ -1504,6 +1511,22 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options) MQTTResponse MQTTClient_connect5(MQTTClient handle, MQTTClient_connectOptions* options, MQTTProperties* connectProperties, MQTTProperties* willProperties) +{ + MQTTClients* m = handle; + MQTTResponse response = MQTTResponse_initializer; + + if (m->c->MQTTVersion < MQTTVERSION_5) + { + response.reasonCode = MQTTCLIENT_WRONG_MQTT_VERSION; + return response; + } + + return MQTTClient_connectAll(handle, options, connectProperties, willProperties); +} + + +MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions* options, + MQTTProperties* connectProperties, MQTTProperties* willProperties) { MQTTClients* m = handle; MQTTResponse rc = MQTTResponse_initializer; @@ -2130,7 +2153,6 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p { while (m->c->connected == 1 && SocketBuffer_getWrite(m->c->net.socket)) { - printf("getwrite %p\n", SocketBuffer_getWrite(m->c->net.socket)); Thread_unlock_mutex(mqttclient_mutex); MQTTClient_yield(); Thread_lock_mutex(mqttclient_mutex); diff --git a/src/MQTTClient.h b/src/MQTTClient.h index 2113277a..077503ce 100644 --- a/src/MQTTClient.h +++ b/src/MQTTClient.h @@ -191,6 +191,10 @@ * Return code: option not applicable to the requested version of MQTT */ #define MQTTCLIENT_BAD_MQTT_OPTION -15 + /** + * Return code: call not applicable to the requested version of MQTT + */ + #define MQTTCLIENT_WRONG_MQTT_VERSION -16 /** diff --git a/src/MQTTPacket.c b/src/MQTTPacket.c index 3dcdc1a0..c6115d19 100644 --- a/src/MQTTPacket.c +++ b/src/MQTTPacket.c @@ -152,7 +152,7 @@ void* MQTTPacket_Factory(int MQTTVersion, networkHandles* net, int* error) buf[0] = header.byte; buf0len = 1 + MQTTPacket_encode(&buf[1], remaining_length); *error = MQTTPersistence_put(net->socket, buf, buf0len, 1, - &data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1); + &data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1, MQTTVersion); free(buf); } #endif @@ -172,9 +172,11 @@ exit: * @param header the one-byte MQTT header * @param buffer the rest of the buffer to write (not including remaining length) * @param buflen the length of the data in buffer to be written + * @param MQTTVersion the version of MQTT being used * @return the completion code (TCPSOCKET_COMPLETE etc) */ -int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int freeData) +int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int freeData, + int MQTTVersion) { int rc; size_t buf0len; @@ -200,7 +202,7 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf int msgId = readInt(&ptraux); rc = MQTTPersistence_put(net->socket, &buf[ws_header], buf0len, count, &buffer, &buflen, - header.bits.type, msgId, 0); + header.bits.type, msgId, 0, MQTTVersion); } #endif rc = WebSocket_putdatas(net, &buf[ws_header], buf0len, count, &buffer, &buflen, &freeData); @@ -223,9 +225,11 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf * @param count the number of buffers * @param buffers the rest of the buffers to write (not including remaining length) * @param buflens the lengths of the data in the array of buffers to be written + * @param the MQTT version being used * @return the completion code (TCPSOCKET_COMPLETE etc) */ -int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees) +int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, + int* frees, int MQTTVersion) { int i, rc; size_t buf0len, total = 0; @@ -249,7 +253,7 @@ int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffe char *ptraux = buffers[2]; int msgId = readInt(&ptraux); rc = MQTTPersistence_put(net->socket, &buf[ws_header], buf0len, count, buffers, buflens, - header.bits.type, msgId, 0); + header.bits.type, msgId, 0, MQTTVersion); } #endif rc = WebSocket_putdatas(net, &buf[ws_header], buf0len, count, buffers, buflens, frees); @@ -496,12 +500,13 @@ int MQTTPacket_send_disconnect(Clients* client, enum MQTTReasonCodes reason, MQT writeChar(&ptr, reason); if (props) MQTTProperties_write(&ptr, props); - if ((rc = MQTTPacket_send(&client->net, header, buf, buflen, 1)) != TCPSOCKET_INTERRUPTED) + if ((rc = MQTTPacket_send(&client->net, header, buf, buflen, 1, + client->MQTTVersion)) != TCPSOCKET_INTERRUPTED) free(buf); } } else - rc = MQTTPacket_send(&client->net, header, NULL, 0, 0); + rc = MQTTPacket_send(&client->net, header, NULL, 0, 0, client->MQTTVersion); Log(LOG_PROTOCOL, 28, NULL, client->net.socket, client->clientID, rc); FUNC_EXIT_RC(rc); return rc; @@ -523,6 +528,7 @@ void* MQTTPacket_publish(int MQTTVersion, unsigned char aHeader, char* data, siz char* enddata = &data[datalen]; FUNC_ENTRY; + memset(pack, '\0', sizeof(Publish)); pack->MQTTVersion = MQTTVersion; pack->header.byte = aHeader; if ((pack->topic = readUTFlen(&curdata, enddata, &pack->topiclen)) == NULL) /* Topic name on which to publish */ @@ -606,7 +612,7 @@ static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net if (type == PUBREL) header.bits.qos = 1; writeInt(&ptr, msgid); - if ((rc = MQTTPacket_send(net, header, buf, 2, 1)) != TCPSOCKET_INTERRUPTED) + if ((rc = MQTTPacket_send(net, header, buf, 2, 1, MQTTVERSION_3_1_1)) != TCPSOCKET_INTERRUPTED) free(buf); FUNC_EXIT_RC(rc); return rc; @@ -805,7 +811,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo ptr = topiclen; writeInt(&ptr, (int)lens[1]); - rc = MQTTPacket_sends(net, header, 4, bufs, lens, frees); + rc = MQTTPacket_sends(net, header, 4, bufs, lens, frees, pack->MQTTVersion); if (rc != TCPSOCKET_INTERRUPTED) free(bufs[2]); } @@ -817,7 +823,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo int frees[3] = {1, 0, 0}; writeInt(&ptr, (int)lens[1]); - rc = MQTTPacket_sends(net, header, 3, bufs, lens, frees); + rc = MQTTPacket_sends(net, header, 3, bufs, lens, frees, pack->MQTTVersion); } if (rc != TCPSOCKET_INTERRUPTED) free(topiclen); diff --git a/src/MQTTPacket.h b/src/MQTTPacket.h index 07501630..350886e3 100644 --- a/src/MQTTPacket.h +++ b/src/MQTTPacket.h @@ -237,8 +237,8 @@ void writeData(char** pptr, const void* data, int datalen); const char* MQTTPacket_name(int ptype); void* MQTTPacket_Factory(int MQTTVersion, networkHandles* net, int* error); -int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int free); -int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees); +int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int free, int MQTTVersion); +int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees, int MQTTVersion); void* MQTTPacket_header_only(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen); int MQTTPacket_send_disconnect(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props); diff --git a/src/MQTTPacketOut.c b/src/MQTTPacketOut.c index 50c5c569..b17c0be5 100644 --- a/src/MQTTPacketOut.c +++ b/src/MQTTPacketOut.c @@ -117,7 +117,7 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion, if (client->password) writeData(&ptr, client->password, client->passwordlen); - rc = MQTTPacket_send(&client->net, packet.header, buf, len, 1); + rc = MQTTPacket_send(&client->net, packet.header, buf, len, 1, MQTTVersion); Log(LOG_PROTOCOL, 0, NULL, client->net.socket, client->clientID, client->cleansession, rc); exit: if (rc != TCPSOCKET_INTERRUPTED) @@ -197,7 +197,7 @@ int MQTTPacket_send_pingreq(networkHandles* net, const char* clientID) FUNC_ENTRY; header.byte = 0; header.bits.type = PINGREQ; - rc = MQTTPacket_send(net, header, NULL, 0, 0); + rc = MQTTPacket_send(net, header, NULL, 0, 0, MQTTVERSION_3_1_1); Log(LOG_PROTOCOL, 20, NULL, net->socket, clientID, rc); FUNC_EXIT_RC(rc); return rc; @@ -258,7 +258,7 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, MQTTSubscribe_options* o writeChar(&ptr, subopts); ++i; } - rc = MQTTPacket_send(&client->net, header, data, datalen, 1); + rc = MQTTPacket_send(&client->net, header, data, datalen, 1, client->MQTTVersion); Log(LOG_PROTOCOL, 22, NULL, client->net.socket, client->clientID, msgid, rc); if (rc != TCPSOCKET_INTERRUPTED) free(data); @@ -347,7 +347,7 @@ int MQTTPacket_send_unsubscribe(List* topics, MQTTProperties* props, int msgid, elem = NULL; while (ListNextElement(topics, &elem)) writeUTF(&ptr, (char*)(elem->content)); - rc = MQTTPacket_send(&client->net, header, data, datalen, 1); + rc = MQTTPacket_send(&client->net, header, data, datalen, 1, client->MQTTVersion); Log(LOG_PROTOCOL, 25, NULL, client->net.socket, client->clientID, msgid, rc); if (rc != TCPSOCKET_INTERRUPTED) free(data); diff --git a/src/MQTTPersistence.c b/src/MQTTPersistence.c index 4ecea272..9d748b7a 100644 --- a/src/MQTTPersistence.c +++ b/src/MQTTPersistence.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2013 IBM Corp. + * Copyright (c) 2009, 2018 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -14,6 +14,7 @@ * Ian Craggs - initial API and implementation and/or initial documentation * Ian Craggs - async client updates * Ian Craggs - fix for bug 432903 - queue persistence + * Ian Craggs - MQTT V5 updates *******************************************************************************/ /** @@ -31,7 +32,7 @@ #include "Heap.h" -static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen); +static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen, int MQTTVersion); static void MQTTPersistence_insertInSeqOrder(List* list, MQTTPersistence_qEntry* qEntry, size_t size); /** @@ -196,13 +197,43 @@ int MQTTPersistence_restore(Clients *c) } else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0) { - MQTTPacket* pack = MQTTPersistence_restorePacket(c->MQTTVersion, buffer, buflen); + int data_MQTTVersion = MQTTVERSION_3_1_1; + char* cur_key = msgkeys[i]; + + if ( strncmp(cur_key, PERSISTENCE_V5_PUBLISH_RECEIVED, + strlen(PERSISTENCE_V5_PUBLISH_RECEIVED)) == 0) + { + data_MQTTVersion = MQTTVERSION_5; + cur_key = PERSISTENCE_PUBLISH_RECEIVED; + } + else if (strncmp(cur_key, PERSISTENCE_V5_PUBLISH_SENT, + strlen(PERSISTENCE_V5_PUBLISH_SENT)) == 0) + { + data_MQTTVersion = MQTTVERSION_5; + cur_key = PERSISTENCE_PUBLISH_SENT; + } + else if (strncmp(cur_key, PERSISTENCE_V5_PUBREL, + strlen(PERSISTENCE_V5_PUBREL)) == 0) + { + data_MQTTVersion = MQTTVERSION_5; + cur_key = PERSISTENCE_PUBREL; + } + + if (data_MQTTVersion == MQTTVERSION_5 && c->MQTTVersion < MQTTVERSION_5) + { + rc = MQTTCLIENT_PERSISTENCE_ERROR; /* can't restore version 5 data with a version 3 client */ + goto exit; + } + + MQTTPacket* pack = MQTTPersistence_restorePacket(data_MQTTVersion, buffer, buflen); if ( pack != NULL ) { - if ( strstr(msgkeys[i],PERSISTENCE_PUBLISH_RECEIVED) != NULL ) + if (strncmp(cur_key, PERSISTENCE_PUBLISH_RECEIVED, + strlen(PERSISTENCE_PUBLISH_RECEIVED)) == 0) { Publish* publish = (Publish*)pack; Messages* msg = NULL; + publish->MQTTVersion = c->MQTTVersion; msg = MQTTProtocol_createMessage(publish, &msg, publish->header.bits.qos, publish->header.bits.retain); msg->nextMessageType = PUBREL; /* order does not matter for persisted received messages */ @@ -211,14 +242,20 @@ int MQTTPersistence_restore(Clients *c) MQTTPacket_freePublish(publish); msgs_rcvd++; } - else if ( strstr(msgkeys[i],PERSISTENCE_PUBLISH_SENT) != NULL ) + else if (strncmp(cur_key, PERSISTENCE_PUBLISH_SENT, + strlen(PERSISTENCE_PUBLISH_SENT)) == 0) { Publish* publish = (Publish*)pack; Messages* msg = NULL; char *key = malloc(MESSAGE_FILENAME_LENGTH + 1); - sprintf(key, "%s%d", PERSISTENCE_PUBREL, publish->msgId); + + publish->MQTTVersion = c->MQTTVersion; + if (publish->MQTTVersion >= MQTTVERSION_5) + sprintf(key, "%s%d", PERSISTENCE_V5_PUBREL, publish->msgId); + else + sprintf(key, "%s%d", PERSISTENCE_PUBREL, publish->msgId); msg = MQTTProtocol_createMessage(publish, &msg, publish->header.bits.qos, publish->header.bits.retain); - if ( c->persistence->pcontainskey(c->phandle, key) == 0 ) + if (c->persistence->pcontainskey(c->phandle, key) == 0) /* PUBLISH Qo2 and PUBREL sent */ msg->nextMessageType = PUBCOMP; /* else: PUBLISH QoS1, or PUBLISH QoS2 and PUBREL not sent */ @@ -230,13 +267,18 @@ int MQTTPersistence_restore(Clients *c) free(key); msgs_sent++; } - else if ( strstr(msgkeys[i],PERSISTENCE_PUBREL) != NULL ) + else if (strncmp(cur_key, PERSISTENCE_PUBREL, strlen(PERSISTENCE_PUBREL)) == 0) { /* orphaned PUBRELs ? */ Pubrel* pubrel = (Pubrel*)pack; char *key = malloc(MESSAGE_FILENAME_LENGTH + 1); - sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, pubrel->msgId); - if ( c->persistence->pcontainskey(c->phandle, key) != 0 ) + + pubrel->MQTTVersion = c->MQTTVersion; + if (pubrel->MQTTVersion >= MQTTVERSION_5) + sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_SENT, pubrel->msgId); + else + sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, pubrel->msgId); + if (c->persistence->pcontainskey(c->phandle, key) != 0) rc = c->persistence->premove(c->phandle, msgkeys[i]); free(pubrel); free(key); @@ -260,7 +302,7 @@ int MQTTPersistence_restore(Clients *c) Log(TRACE_MINIMUM, -1, "%d sent messages and %d received messages restored for client %s\n", msgs_sent, msgs_rcvd, c->clientID); MQTTPersistence_wrapMsgID(c); - +exit: FUNC_EXIT_RC(rc); return rc; } @@ -335,13 +377,15 @@ void MQTTPersistence_insertInOrder(List* list, void* content, size_t size) * @param count number of buffers representing the variable header and/or the payload. * @param buffers the buffers representing the variable header and/or the payload. * @param buflens length of the buffers representing the variable header and/or the payload. + * @param htype MQTT packet type - PUBLISH or PUBREL * @param msgId the message ID. * @param scr 0 indicates message in the sending direction; 1 indicates message in the * receiving direction. + * @param the MQTT version being used (>= MQTTVERSION_5 means properties included) * @return 0 if success, #MQTTCLIENT_PERSISTENCE_ERROR otherwise. */ int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count, - char** buffers, size_t* buflens, int htype, int msgId, int scr ) + char** buffers, size_t* buflens, int htype, int msgId, int scr, int MQTTVersion) { int rc = 0; extern ClientStates* bstate; @@ -368,15 +412,32 @@ int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count, } /* key */ - if ( scr == 0 ) + if (scr == 0) { /* sending */ - if (htype == PUBLISH) /* PUBLISH QoS1 and QoS2*/ - sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, msgId); - if (htype == PUBREL) /* PUBREL */ - sprintf(key, "%s%d", PERSISTENCE_PUBREL, msgId); + char* key_id = PERSISTENCE_PUBLISH_SENT; + + if (htype == PUBLISH) /* PUBLISH QoS1 and QoS2*/ + { + if (MQTTVersion >= MQTTVERSION_5) + key_id = PERSISTENCE_V5_PUBLISH_SENT; + } + else if (htype == PUBREL) /* PUBREL */ + { + if (MQTTVersion >= MQTTVERSION_5) + key_id = PERSISTENCE_V5_PUBREL; + else + key_id = PERSISTENCE_PUBREL; + } + sprintf(key, "%s%d", key_id, msgId); + } + else if (scr == 1) /* receiving PUBLISH QoS2 */ + { + char* key_id = PERSISTENCE_PUBLISH_RECEIVED; + + if (MQTTVersion >= MQTTVERSION_5) + key_id = PERSISTENCE_V5_PUBLISH_RECEIVED; + sprintf(key, "%s%d", key_id, msgId); } - if ( scr == 1 ) /* receiving PUBLISH QoS2 */ - sprintf(key, "%s%d", PERSISTENCE_PUBLISH_RECEIVED, msgId); rc = client->persistence->pput(client->phandle, key, nbufs, bufs, lens); @@ -407,17 +468,24 @@ int MQTTPersistence_remove(Clients* c, char *type, int qos, int msgId) if (c->persistence != NULL) { char *key = malloc(MESSAGE_FILENAME_LENGTH + 1); - if ( (strcmp(type,PERSISTENCE_PUBLISH_SENT) == 0) && qos == 2 ) + if (strcmp(type, PERSISTENCE_PUBLISH_SENT) == 0) //&& qos == 2 ) { + sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_SENT, msgId) ; + rc = c->persistence->premove(c->phandle, key); + sprintf(key, "%s%d", PERSISTENCE_V5_PUBREL, msgId) ; + rc += c->persistence->premove(c->phandle, key); sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, msgId) ; - rc = c->persistence->premove(c->phandle, key); + rc += c->persistence->premove(c->phandle, key); sprintf(key, "%s%d", PERSISTENCE_PUBREL, msgId) ; - rc = c->persistence->premove(c->phandle, key); + rc += c->persistence->premove(c->phandle, key); } else /* PERSISTENCE_PUBLISH_SENT && qos == 1 */ { /* or PERSISTENCE_PUBLISH_RECEIVED */ - sprintf(key, "%s%d", type, msgId) ; + + sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_RECEIVED, msgId); rc = c->persistence->premove(c->phandle, key); + sprintf(key, "%s%d", PERSISTENCE_PUBLISH_RECEIVED, msgId); + rc += c->persistence->premove(c->phandle, key); } free(key); } @@ -489,17 +557,15 @@ int MQTTPersistence_unpersistQueueEntry(Clients* client, MQTTPersistence_qEntry* int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry* qe) { + const int MAX_NO_OF_BUFFERS = 9; int rc = 0; - int nbufs = 8; int bufindex = 0; char key[PERSISTENCE_MAX_KEY_LENGTH + 1]; - int* lens = NULL; - void** bufs = NULL; + int lens[MAX_NO_OF_BUFFERS]; + void* bufs[MAX_NO_OF_BUFFERS]; + int props_allocated = 0; FUNC_ENTRY; - lens = (int*)malloc(nbufs * sizeof(int)); - bufs = malloc(nbufs * sizeof(char *)); - bufs[bufindex] = &qe->msg->payloadlen; lens[bufindex++] = sizeof(qe->msg->payloadlen); @@ -522,23 +588,43 @@ int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry* lens[bufindex++] = (int)strlen(qe->topicName) + 1; bufs[bufindex] = &qe->topicLen; - lens[bufindex++] = sizeof(qe->topicLen); - - sprintf(key, "%s%u", PERSISTENCE_QUEUE_KEY, ++aclient->qentry_seqno); + lens[bufindex++] = sizeof(qe->topicLen); + + if (aclient->MQTTVersion >= MQTTVERSION_5) /* persist properties */ + { + MQTTProperties no_props = MQTTProperties_initializer; + MQTTProperties* props = &no_props; + int temp_len = 0; + char* ptr = NULL; + + if (qe->msg->struct_version >= 1) + props = &qe->msg->properties; + + temp_len = MQTTProperties_len(props); + ptr = bufs[bufindex] = malloc(temp_len); + props_allocated = bufindex; + rc = MQTTProperties_write(&ptr, props); + lens[bufindex++] = temp_len; + + sprintf(key, "%s%u", PERSISTENCE_V5_QUEUE_KEY, ++aclient->qentry_seqno); + } + else + sprintf(key, "%s%u", PERSISTENCE_QUEUE_KEY, ++aclient->qentry_seqno); + qe->seqno = aclient->qentry_seqno; - if ((rc = aclient->persistence->pput(aclient->phandle, key, nbufs, (char**)bufs, lens)) != 0) + if ((rc = aclient->persistence->pput(aclient->phandle, key, bufindex, (char**)bufs, lens)) != 0) Log(LOG_ERROR, 0, "Error persisting queue entry, rc %d", rc); - free(lens); - free(bufs); + if (props_allocated != 0) + free(bufs[props_allocated]); FUNC_EXIT_RC(rc); return rc; } -static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen) +static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen, int MQTTVersion) { MQTTPersistence_qEntry* qe = NULL; char* ptr = buffer; @@ -551,6 +637,8 @@ static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, s qe->msg = malloc(sizeof(MQTTPersistence_message)); memset(qe->msg, '\0', sizeof(MQTTPersistence_message)); + qe->msg->struct_version = 1; + qe->msg->payloadlen = *(int*)ptr; ptr += sizeof(int); @@ -579,6 +667,10 @@ static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, s qe->topicLen = *(int*)ptr; ptr += sizeof(int); + if (MQTTVersion >= MQTTVERSION_5 && + MQTTProperties_read(&qe->msg->properties, &ptr, buffer + buflen) != 1) + Log(LOG_ERROR, -1, "Error restoring properties from persistence"); + FUNC_EXIT; return qe; } @@ -621,17 +713,21 @@ int MQTTPersistence_restoreMessageQueue(Clients* c) char *buffer = NULL; int buflen; - if (strncmp(msgkeys[i], PERSISTENCE_QUEUE_KEY, strlen(PERSISTENCE_QUEUE_KEY)) != 0) + if (strncmp(msgkeys[i], PERSISTENCE_QUEUE_KEY, strlen(PERSISTENCE_QUEUE_KEY)) != 0 && + strncmp(msgkeys[i], PERSISTENCE_V5_QUEUE_KEY, strlen(PERSISTENCE_V5_QUEUE_KEY)) != 0) { - ; + ; /* ignore if not a queue entry key */ } else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0) { - MQTTPersistence_qEntry* qe = MQTTPersistence_restoreQueueEntry(buffer, buflen); + int MQTTVersion = + (strncmp(msgkeys[i], PERSISTENCE_V5_QUEUE_KEY, strlen(PERSISTENCE_V5_QUEUE_KEY)) == 0) + ? MQTTVERSION_5 : MQTTVERSION_3_1_1; + MQTTPersistence_qEntry* qe = MQTTPersistence_restoreQueueEntry(buffer, buflen, MQTTVersion); if (qe) { - qe->seqno = atoi(msgkeys[i]+2); + qe->seqno = atoi(strchr(msgkeys[i], '-')+1); /* key format is tag'-'seqno */ MQTTPersistence_insertInSeqOrder(c->messageQueue, qe, sizeof(MQTTPersistence_qEntry)); free(buffer); c->qentry_seqno = max(c->qentry_seqno, qe->seqno); diff --git a/src/MQTTPersistence.h b/src/MQTTPersistence.h index 18f1c420..93edb2b9 100644 --- a/src/MQTTPersistence.h +++ b/src/MQTTPersistence.h @@ -14,6 +14,7 @@ * Ian Craggs - initial API and implementation and/or initial documentation * Ian Craggs - async client updates * Ian Craggs - fix for bug 432903 - queue persistence + * Ian Craggs - MQTT V5 updates *******************************************************************************/ #if !defined(MQTTPERSISTENCE_H) @@ -24,6 +25,7 @@ #endif #include "Clients.h" +#include "MQTTProperties.h" /** Stem of the key for a sent PUBLISH QoS1 or QoS2 */ #define PERSISTENCE_PUBLISH_SENT "s-" @@ -31,10 +33,22 @@ #define PERSISTENCE_PUBREL "sc-" /** Stem of the key for a received PUBLISH QoS2 */ #define PERSISTENCE_PUBLISH_RECEIVED "r-" + +/** Stem of the key for a sent MQTT V5 PUBLISH QoS1 or QoS2 */ +#define PERSISTENCE_V5_PUBLISH_SENT "s5-" +/** Stem of the key for a sent MQTT V5 PUBREL */ +#define PERSISTENCE_V5_PUBREL "sc5-" +/** Stem of the key for a received MQTT V5 PUBLISH QoS2 */ +#define PERSISTENCE_V5_PUBLISH_RECEIVED "r5-" + /** Stem of the key for an async client command */ #define PERSISTENCE_COMMAND_KEY "c-" +/** Stem of the key for an MQTT V5 async client command */ +#define PERSISTENCE_V5_COMMAND_KEY "c-" /** Stem of the key for an async client message queue */ #define PERSISTENCE_QUEUE_KEY "q-" +/** Stem of the key for an MQTT V5 message queue */ +#define PERSISTENCE_V5_QUEUE_KEY "q5-" #define PERSISTENCE_MAX_KEY_LENGTH 8 int MQTTPersistence_create(MQTTClient_persistence** per, int type, void* pcontext); @@ -45,7 +59,7 @@ int MQTTPersistence_restore(Clients* c); void* MQTTPersistence_restorePacket(int MQTTVersion, char* buffer, size_t buflen); void MQTTPersistence_insertInOrder(List* list, void* content, size_t size); int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count, - char** buffers, size_t* buflens, int htype, int msgId, int scr); + char** buffers, size_t* buflens, int htype, int msgId, int scr, int MQTTVersion); int MQTTPersistence_remove(Clients* c, char* type, int qos, int msgId); void MQTTPersistence_wrapMsgID(Clients *c); @@ -59,6 +73,7 @@ typedef struct int retained; int dup; int msgid; + MQTTProperties properties; } MQTTPersistence_message; typedef struct diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index 7acbb582..91175a7a 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -309,7 +309,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock) if (m->MQTTVersion >= MQTTVERSION_5) m->properties = MQTTProperties_copy(&publish->properties); m->nextMessageType = PUBREL; - if ( ( listElem = ListFindItem(client->inboundMsgs, &(m->msgid), messageIDCompare) ) != NULL ) + if ((listElem = ListFindItem(client->inboundMsgs, &(m->msgid), messageIDCompare)) != NULL) { /* discard queued publication with same msgID that the current incoming message */ Messages* msg = (Messages*)(listElem->content); MQTTProtocol_removePublication(msg->publish); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index dfb97b35..06bf46ff 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -597,6 +597,56 @@ SET_TESTS_PROPERTIES( PROPERTIES TIMEOUT 540 ) +ADD_EXECUTABLE( + test95 + test95.c +) + +TARGET_LINK_LIBRARIES( + test95 + paho-mqtt3a +) + +ADD_TEST( + NAME test95-1-offline-buffering-send-disconnected + COMMAND test95 "--test_no" "1" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test95-2-offline-buffering-send-disconnected-serverURIs + COMMAND test95 "--test_no" "2" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test95-3-offline-buffering-auto-reconnect + COMMAND test95 "--test_no" "3" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test95-4-offline-buffering-auto-reconnect-serverURIs + COMMAND test95 "--test_no" "4" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test95-5-offline-buffering-max-buffered + COMMAND test95 "--test_no" "5" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +ADD_TEST( + NAME test95-6-offline-buffering-max-buffered-binary-will + COMMAND test95 "--test_no" "6" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +SET_TESTS_PROPERTIES( + test95-1-offline-buffering-send-disconnected + test95-2-offline-buffering-send-disconnected-serverURIs + test95-3-offline-buffering-auto-reconnect + test95-4-offline-buffering-auto-reconnect-serverURIs + test95-5-offline-buffering-max-buffered + test95-6-offline-buffering-max-buffered-binary-will + PROPERTIES TIMEOUT 540 +) + ADD_EXECUTABLE( test10 test10.c diff --git a/test/test1.c b/test/test1.c index fb45fef2..87704622 100644 --- a/test/test1.c +++ b/test/test1.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2017 IBM Corp. + * Copyright (c) 2009, 2018 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -22,13 +22,6 @@ * Tests for the MQ Telemetry MQTT C client */ - -/* -#if !defined(_RTSHEADER) - #include -#endif -*/ - #include "MQTTClient.h" #include #include diff --git a/test/test10.c b/test/test10.c index c8e07b21..cbd08f7d 100644 --- a/test/test10.c +++ b/test/test10.c @@ -391,14 +391,16 @@ int test_client_topic_aliases(struct Options options) int count = 0; char* test_topic = "test_client_topic_aliases"; int topicAliasMaximum = 0; + MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer; fprintf(xml, " 0\n"); - opts.cleanstart = 1; - property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; - property.value.integer4 = 30; /* in seconds */ - MQTTProperties_add(&props, &property); - response = MQTTClient_connect5(c, &opts, &props, NULL); - assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, + if (start_mqtt_version == MQTTVERSION_5) + { + MyLog(LOGA_DEBUG, "Cleanup by connecting clean start, add session expiry > 0\n"); + opts.cleanstart = 1; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; /* in seconds */ + MQTTProperties_add(&props, &property); + response = MQTTClient_connect5(c, &opts, &props, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); - MQTTResponse_free(response); - if (response.reasonCode != MQTTCLIENT_SUCCESS) - return -1; + MQTTResponse_free(response); + if (response.reasonCode != MQTTCLIENT_SUCCESS) + return -1; + } + else + { + MyLog(LOGA_DEBUG, "Cleanup by connecting clean session, then reconnecting non-cleansession\n"); + opts.cleanstart = 0; /* only applies to MQTT V5 */ + opts.cleansession = 1; + rc = MQTTClient_connect(c, &opts); + assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + rc = MQTTClient_disconnect(c, 1000); + assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + opts.cleansession = 0; + rc = MQTTClient_connect(c, &opts); + assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + if (rc != MQTTCLIENT_SUCCESS) + return -1; + } /* subscribe so we can get messages back */ - response = MQTTClient_subscribe5(c, topic, subsqos, NULL, NULL); - assert("Good rc from subscribe", response.reasonCode == subsqos, "rc was %d", response.reasonCode); + if (start_mqtt_version == MQTTVERSION_5) + { + response = MQTTClient_subscribe5(c, topic, subsqos, NULL, NULL); + assert("Good rc from subscribe", response.reasonCode == subsqos, "rc was %d", response.reasonCode); + } + else + { + rc = MQTTClient_subscribe(c, topic, subsqos); + assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + } /* send messages so that we can receive the same ones */ + if (start_mqtt_version == MQTTVERSION_5) + { + property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY; + property.value.data.data = "test user property"; + property.value.data.len = (int)strlen(property.value.data.data); + property.value.value.data = "test user property value"; + property.value.value.len = (int)strlen(property.value.value.data); + MQTTProperties_add(&pub_props, &property); + } + for (i = 0; i < count; ++i) { sprintf(buffer, "Message sequence no %d", i); - response = MQTTClient_publish5(c, topic, 10, buffer, qos, 0, NULL, NULL); - assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + if (start_mqtt_version == MQTTVERSION_5) + { + response = MQTTClient_publish5(c, topic, 10, buffer, qos, 0, &pub_props, NULL); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + } + else + { + rc = MQTTClient_publish(c, topic, 10, buffer, qos, 0, NULL); + assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + } } /* disconnect immediately without receiving the incoming messages */ - MQTTClient_disconnect5(c, 0, MQTTREASONCODE_SUCCESS, NULL); /* now there should be "orphaned" publications */ + if (start_mqtt_version == MQTTVERSION_5) + rc = MQTTClient_disconnect5(c, 0, MQTTREASONCODE_SUCCESS, NULL); /* now there should be "orphaned" publications */ + else + rc = MQTTClient_disconnect(c, 0); /* now there should be "orphaned" publications */ + assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); rc = MQTTClient_getPendingDeliveryTokens(c, &tokens); assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); @@ -865,9 +937,19 @@ int test4_run(int qos) mytoken = tokens[0]; } MQTTProperties_free(&props); + MQTTProperties_free(&pub_props); MQTTClient_destroy(&c); /* force re-reading persistence on create */ - MQTTClient_create(&c, options.connection, "xrctest15_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); + createOpts.MQTTVersion = restore_mqtt_version; + rc = MQTTClient_createWithOptions(&c, options.connection, "xrctest15_test_4", + MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts); + if (start_mqtt_version == MQTTVERSION_5 && restore_mqtt_version == MQTTVERSION_3_1_1) + { + assert("Persistence error from create", rc == MQTTCLIENT_PERSISTENCE_ERROR, "rc was %d", rc); + goto exit; + } + else + assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); rc = MQTTClient_getPendingDeliveryTokens(c, &tokens); assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); @@ -901,6 +983,9 @@ int test4_run(int qos) { assert("No duplicates should be received for qos 2", qos == 1, "qos is %d", qos); MyLog(LOGA_DEBUG, "Duplicate message id %d", m->msgid); + assert("properties are received", m->properties.count > 0, "property count is %d", + m->properties.count); + logProperties(&m->properties); MQTTClient_freeMessage(&m); MQTTClient_free(topicName); dup = 1; @@ -911,6 +996,13 @@ int test4_run(int qos) { MyLog(LOGA_DEBUG, "Received message id %d", m->msgid); assert("topicName is correct", strcmp(topicName, topic) == 0, "topicName is %s", topicName); + if (start_mqtt_version == MQTTVERSION_5) + assert("properties are received", m->properties.count > 0, "property count is %d", + m->properties.count); + else + assert("properties are not received", m->properties.count == 0, "property count is %d", + m->properties.count); + logProperties(&m->properties); MQTTClient_freeMessage(&m); MQTTClient_free(topicName); } @@ -926,7 +1018,7 @@ int test4_run(int qos) MQTTClient_destroy(&c); -/* TODO - unused -remove? exit: */ +exit: MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", (failures == 0) ? "passed" : "failed", testname, tests, failures); @@ -937,9 +1029,12 @@ int test4_run(int qos) int test4(struct Options options) { int rc = 0; - fprintf(xml, "\n", elapsed(global_start_time) / 1000); if (cur_output != output) { @@ -970,13 +1065,16 @@ int test5(struct Options options) MQTTProperties props = MQTTProperties_initializer; MQTTResponse response = MQTTResponse_initializer; int i, rc; + MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer; fprintf(xml, "msgid); - test7_messageCount++; + if (message->qos == 1 && message->dup == 1) + ; /* ignore dups */ + else + test7_messageCount++; MQTTAsync_freeMessage(&message); MQTTAsync_free(topicName); @@ -1369,7 +1372,7 @@ void test7_onConnect(void* context, MQTTAsync_successData5* response) Test7: Pending tokens *********************************************************************/ -int test7(struct Options options) +int test7_run(int qos, int start_mqtt_version, int restore_mqtt_version) { int subsqos = 2; MQTTAsync c; @@ -1383,14 +1386,20 @@ int test7(struct Options options) int msg_count = 6; MQTTProperty property; MQTTProperties props = MQTTProperties_initializer; + MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer; + + MyLog(LOGA_INFO, "Starting test 7 - persistence, qos %d, MQTT versions: %s then %s", qos, + (start_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1", + (restore_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1"); - MyLog(LOGA_INFO, "Starting test 7 - pending tokens"); fprintf(xml, "message = "will message"; @@ -1418,17 +1427,26 @@ int test7(struct Options options) /* connect clean and then leave messages lying around */ test_finished = 0; + test7_subscribed = 0; MyLog(LOGA_DEBUG, "Connecting"); - opts.cleanstart = 1; - opts.connectProperties = &props; - property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; - property.value.integer4 = 999999; - MQTTProperties_add(opts.connectProperties, &property); - opts.onSuccess5 = test7_onConnect; - rc = MQTTAsync_connect(c, &opts); - assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); - if (rc != MQTTASYNC_SUCCESS) - goto exit; + + if (start_mqtt_version == MQTTVERSION_5) + { + opts.cleanstart = 1; + opts.connectProperties = &props; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 999999; + MQTTProperties_add(opts.connectProperties, &property); + opts.onSuccess5 = test7_onConnect; + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + goto exit; + } + else + { + /* MQTT 3 version */ + } while (!test7_subscribed) #if defined(WIN32) @@ -1439,12 +1457,13 @@ int test7(struct Options options) pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11"; pubmsg.payloadlen = 11; - pubmsg.qos = 2; + pubmsg.qos = qos; pubmsg.retained = 0; rc = MQTTAsync_send(c, test7_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &ropts); + assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); MyLog(LOGA_DEBUG, "Token was %d", ropts.token); rc = MQTTAsync_isComplete(c, ropts.token); - /*assert("0 rc from isComplete", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);*/ + assert("0 rc from isComplete", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); rc = MQTTAsync_waitForCompletion(c, ropts.token, 5000L); assert("Good rc from waitForCompletion", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); rc = MQTTAsync_isComplete(c, ropts.token); @@ -1452,12 +1471,11 @@ int test7(struct Options options) test7_messageCount = 0; int i = 0; - pubmsg.qos = 2; + pubmsg.qos = qos; for (i = 0; i < msg_count; ++i) { pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11"; pubmsg.payloadlen = 11; - //pubmsg.qos = (pubmsg.qos == 2) ? 1 : 2; pubmsg.retained = 0; rc = MQTTAsync_sendMessage(c, test7_topic, &pubmsg, &ropts); } @@ -1485,7 +1503,9 @@ int test7(struct Options options) MQTTAsync_destroy(&c); /* force re-reading persistence on create */ MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR); - rc = MQTTAsync_create(&c, options.connection, "async_test7", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); + createOpts.MQTTVersion = restore_mqtt_version; + rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7", + MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts); assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc); if (rc != MQTTASYNC_SUCCESS) { @@ -1510,6 +1530,7 @@ int test7(struct Options options) assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); MyLog(LOGA_DEBUG, "Reconnecting"); + opts.MQTTVersion = restore_mqtt_version; opts.context = c; opts.cleanstart = 0; if (MQTTAsync_connect(c, &opts) != 0) @@ -1556,6 +1577,26 @@ exit: } +int test7(struct Options options) +{ + int rc = 0; + fprintf(xml, "\n", elapsed(global_start_time) / 1000); + if (cur_output != output) + { + fprintf(xml, "%s", output); + cur_output = output; + } + fprintf(xml, "\n"); + return rc; +} + + /********************************************************************* diff --git a/test/test5.c b/test/test5.c index a60b03d6..8868a610 100644 --- a/test/test5.c +++ b/test/test5.c @@ -17,7 +17,7 @@ /** * @file - * SSL tests for the MQ Telemetry Asynchronous MQTT C client + * SSL tests for the Eclipse Paho Asynchronous MQTT C client */ #include "MQTTAsync.h" diff --git a/test/test8.c b/test/test8.c index 088fd53e..ac0c3741 100644 --- a/test/test8.c +++ b/test/test8.c @@ -1,28 +1,25 @@ -/*--------------------------------------------------------------------*/ -/* [Platforms]UNIX NT[/Platforms] */ -/* [Title]MQ Telemetry MQTT Async C client tests - HA and connect */ -/* failures */ -/* [/Title] */ -/* [Testclasses]stcom1 stmqcom1[/Category] */ -/* [Category]MQ Telemetry[/Category] */ -/* */ -/* Copyright IBM 2013 */ -/* All rights reserved. */ -/*--------------------------------------------------------------------*/ - +/******************************************************************************* + * Copyright (c) 2012, 2018 IBM Corp. + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * Contributors: + * Ian Craggs - initial API and implementation and/or initial documentation + *******************************************************************************/ /** * @file - * Tests for the MQ Telemetry MQTT Async C client + * Tests for the Paho MQTT Async C client */ -/* -#if !defined(_RTSHEADER) - #include -#endif -*/ - #include "MQTTAsync.h" #include #include diff --git a/test/test9.c b/test/test9.c index 90bccb28..f0b3ab33 100644 --- a/test/test9.c +++ b/test/test9.c @@ -19,7 +19,7 @@ /** * @file - * Offline buffering and automatic reconnect tests for the MQ Telemetry Asynchronous MQTT C client + * Offline buffering and automatic reconnect tests for the Paho Asynchronous MQTT C client * */ @@ -2121,7 +2121,7 @@ int main(int argc, char** argv) int (*tests[])() = { NULL, test1, test2, test3, test4, test5, test6, test7}; time_t randtime; - srand((unsigned) time(&randtime)); + srand((unsigned) time(&randtime)); sprintf(unique, "%u", rand()); MyLog(LOGA_INFO, "Random prefix/suffix is %s", unique); diff --git a/test/test95.c b/test/test95.c new file mode 100644 index 00000000..6f0a8c04 --- /dev/null +++ b/test/test95.c @@ -0,0 +1,2358 @@ +/******************************************************************************* + * Copyright (c) 2012, 2018 IBM Corp. + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * Contributors: + * Ian Craggs - initial API and implementation and/or initial documentation + * Ian Craggs - correct some compile warnings + * Ian Craggs - add binary will message test + * Ian Craggs - MQTT V5 updates + *******************************************************************************/ + + +/** + * @file + * Offline buffering and automatic reconnect tests for the Paho Asynchronous MQTT C client + * + */ + + +#include "MQTTAsync.h" +#include +#include +#include "Thread.h" + +#if !defined(_WINDOWS) + #include + #include + #include + #include +#else + #include +#endif + +char unique[50]; // unique suffix/prefix to add to clientid/topic etc + +#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0])) + +void usage(void) +{ + printf("help!!\n"); + exit(EXIT_FAILURE); +} + +struct Options +{ + char* connection; /**< connection to system under test. */ + char* proxy_connection; /**< connection to proxy */ + int verbose; + int test_no; +} options = +{ + "iot.eclipse.org:1883", + "localhost:1883", + 0, + 0, +}; + +void getopts(int argc, char** argv) +{ + int count = 1; + + while (count < argc) + { + if (strcmp(argv[count], "--test_no") == 0) + { + if (++count < argc) + options.test_no = atoi(argv[count]); + else + usage(); + } + else if (strcmp(argv[count], "--connection") == 0) + { + if (++count < argc) + options.connection = argv[count]; + else + usage(); + } + else if (strcmp(argv[count], "--proxy_connection") == 0) + { + if (++count < argc) + options.proxy_connection = argv[count]; + else + usage(); + } + else if (strcmp(argv[count], "--verbose") == 0) + options.verbose = 1; + count++; + } +} + + +#define LOGA_DEBUG 0 +#define LOGA_INFO 1 +#include +#include +#include +void MyLog(int LOGA_level, char* format, ...) +{ + static char msg_buf[256]; + va_list args; + struct timeb ts; + + struct tm *timeinfo; + + if (LOGA_level == LOGA_DEBUG && options.verbose == 0) + return; + + ftime(&ts); + timeinfo = localtime(&ts.time); + strftime(msg_buf, 80, "%Y%m%d %H%M%S", timeinfo); + + sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm); + + va_start(args, format); + vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), + format, args); + va_end(args); + + printf("%s\n", msg_buf); + fflush(stdout); +} + +void MySleep(long milliseconds) +{ +#if defined(WIN32) || defined(WIN64) + Sleep(milliseconds); +#else + usleep(milliseconds*1000); +#endif +} + +#if defined(WIN32) || defined(_WINDOWS) +#define START_TIME_TYPE DWORD +static DWORD start_time = 0; +START_TIME_TYPE start_clock(void) +{ + return GetTickCount(); +} +#elif defined(AIX) +#define START_TIME_TYPE struct timespec +START_TIME_TYPE start_clock(void) +{ + static struct timespec start; + clock_gettime(CLOCK_REALTIME, &start); + return start; +} +#else +#define START_TIME_TYPE struct timeval +/* TODO - unused - remove? static struct timeval start_time; */ +START_TIME_TYPE start_clock(void) +{ + struct timeval start_time; + gettimeofday(&start_time, NULL); + return start_time; +} +#endif + +#if defined(WIN32) +long elapsed(START_TIME_TYPE start_time) +{ + return GetTickCount() - start_time; +} +#elif defined(AIX) +#define assert(a) +long elapsed(struct timespec start) +{ + struct timespec now, res; + + clock_gettime(CLOCK_REALTIME, &now); + ntimersub(now, start, res); + return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L; +} +#else +long elapsed(START_TIME_TYPE start_time) +{ + struct timeval now, res; + + gettimeofday(&now, NULL); + timersub(&now, &start_time, &res); + return (res.tv_sec) * 1000 + (res.tv_usec) / 1000; +} +#endif + +#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d) +#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e) + +#define MAXMSGS 30; + +int tests = 0; +int failures = 0; +FILE* xml; +START_TIME_TYPE global_start_time; +char output[3000]; +char* cur_output = output; + + +void write_test_result(void) +{ + long duration = elapsed(global_start_time); + + fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000); + if (cur_output != output) + { + fprintf(xml, "%s", output); + cur_output = output; + } + fprintf(xml, "\n"); +} + +void myassert(char* filename, int lineno, char* description, int value, + char* format, ...) +{ + ++tests; + if (!value) + { + va_list args; + + ++failures; + MyLog(LOGA_INFO, "Assertion failed, file %s, line %d, description: %s", filename, + lineno, description); + + va_start(args, format); + vprintf(format, args); + va_end(args); + + cur_output += sprintf(cur_output, "file %s, line %d \n", + description, filename, lineno); + } + else + MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", + filename, lineno, description); +} + + +void logProperties(MQTTProperties *props) +{ + int i = 0; + + for (i = 0; i < props->count; ++i) + { + int id = props->array[i].identifier; + const char* name = MQTTPropertyName(id); + char* intformat = "Property name %s value %d"; + + switch (MQTTProperty_getType(id)) + { + case MQTTPROPERTY_TYPE_BYTE: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.byte); + break; + case MQTTPROPERTY_TYPE_TWO_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer2); + break; + case MQTTPROPERTY_TYPE_FOUR_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4); + break; + case MQTTPROPERTY_TYPE_VARIABLE_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4); + break; + case MQTTPROPERTY_TYPE_BINARY_DATA: + case MQTTPROPERTY_TYPE_UTF_8_ENCODED_STRING: + MyLog(LOGA_INFO, "Property name %s value len %.*s", name, + props->array[i].value.data.len, props->array[i].value.data.data); + break; + case MQTTPROPERTY_TYPE_UTF_8_STRING_PAIR: + MyLog(LOGA_INFO, "Property name %s key %.*s value %.*s", name, + props->array[i].value.data.len, props->array[i].value.data.data, + props->array[i].value.value.len, props->array[i].value.value.data); + break; + } + } +} + +/********************************************************************* + + Tests: offline buffering - sending messages while disconnected + + 1. send some messages while disconnected, check that they are sent + 2. repeat test 1 using serverURIs + 3. repeat test 1 using auto reconnect + 4. repeat test 2 using auto reconnect + 5. check max-buffered + 6. check auto-reconnect parms alter behaviour as expected + + Tests: automatic reconnect + + - check that connected() is called + - check that reconnect() causes reconnect attempt + - check that reconnect() fails if no connect has been previously attempted + + *********************************************************************/ + + + + +/********************************************************************* + + Test1: offline buffering - sending messages while disconnected + + 1. call connect + 2. use proxy to disconnect the client + 3. while the client is disconnected, send more messages + 4. when the client reconnects, check that those messages are sent + + *********************************************************************/ + +int test1_will_message_received = 0; +int test1_messages_received = 0; + +int test1_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + static int first = 1; + + if (first == 1) + { + first = 0; + return 0; /* to force queue persistence */ + } + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test1_will_message_received = 1; + else + test1_messages_received++; + + if (message->struct_version == 1) + { + assert("Properties count should be > 0", message->properties.count > 0, + "Properties count was %d\n", message->properties.count); + logProperties(&message->properties); + } + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +int test1Finished = 0; + +int test1OnFailureCalled = 0; + +void test1cOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test1OnFailureCalled++; + test1Finished = 1; +} + +void test1dOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test1OnFailureCalled++; + test1Finished = 1; +} + +void test1cOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client d, context %p\n", context); + MQTTAsync c = (MQTTAsync)context; + int rc; + static int done = 0; + + if (done == 0) + { + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + done = 1; /* only do this once */ + } +} + + +int test1dReady = 0; +char willTopic[100]; +char test_topic[50]; + +void test1donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test1dReady = 1; +} + + +void test1dOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test1donSubscribe; + opts.context = c; + + rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test1Finished = 1; +} + +int test1c_connected = 0; + +void test1cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test1c_connected = 1; +} + + +int test1(struct Options options) +{ + char* testname = "test1"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperties willProps = MQTTProperties_initializer; + MQTTProperty property; + + sprintf(willTopic, "paho-test95-1-%s", unique); + sprintf(clientidc, "paho-test95-1-c-%s", unique); + sprintf(clientidd, "paho-test95-1-d-%s", unique); + sprintf(test_topic, "paho-test95-1-test topic %s", unique); + + test1Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 1 - messages while disconnected"); + fprintf(xml, "message = "will message"; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + + property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY; + property.value.data.data = "test user property"; + property.value.data.len = (int)strlen(property.value.data.data); + property.value.value.data = "test user property value"; + property.value.value.len = (int)strlen(property.value.value.data); + MQTTProperties_add(&willProps, &property); + opts.willProperties = &willProps; + + opts.onSuccess5 = test1cOnConnect; + opts.onFailure5 = test1cOnFailure; + opts.context = c; + + opts.cleanstart = 0; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + + opts.connectProperties = &props; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + MQTTProperties_free(&willProps); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for will message */ + while (!test1_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test1c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 3; ++i) + { + char buf[50]; + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + MQTTProperty property; + MQTTProperties props = MQTTProperties_initializer; + + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)strlen(pubmsg.payload) + 1; + pubmsg.qos = i; + pubmsg.retained = 0; + + property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY; + property.value.data.data = "test user property"; + property.value.data.len = (int)strlen(property.value.data.data); + property.value.value.data = "test user property value"; + property.value.value.len = (int)strlen(property.value.value.data); + MQTTProperties_add(&props, &property); + pubmsg.properties = props; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + /* destroy and recreate to read from persistence */ + MyLog(LOGA_DEBUG, "Destroy and recreate client c"); + MQTTAsync_destroy(&c); + + createOptions.sendWhileDisconnected = 1; + createOptions.MQTTVersion = MQTTVERSION_5; + rc = MQTTAsync_createWithOptions(&c, options.proxy_connection, clientidc, + MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOptions); + assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d \n", rc); + if (rc != MQTTASYNC_SUCCESS) + { + MQTTAsync_destroy(&c); + goto exit; + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + rc = MQTTAsync_setConnected(c, c, test1cConnected); + assert("Good rc from setConnectedCallback", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + opts.will = &wopts; + opts.onSuccess5 = test1cOnConnect; + opts.onFailure5 = test1cOnFailure; + opts.context = c; + opts.cleanstart = 0; + MyLog(LOGA_DEBUG, "Reconnecting client c"); + test1c_connected = 0; + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for client to be reconnected */ + while (!test1c_connected && ++count < 10000) + MySleep(100); + + /* wait for messages to be received */ + while (test1_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + + +/********************************************************************* + + Test2: offline buffering - sending messages while disconnected + + 1. call connect + 2. use proxy to disconnect the client + 3. while the client is disconnected, send more messages + 4. when the client reconnects, check that those messages are sent + + *********************************************************************/ + +int test2_will_message_received = 0; +int test2_messages_received = 0; + +int test2_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test2_will_message_received = 1; + else + test2_messages_received++; + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +int test2Finished = 0; + +int test2OnFailureCalled = 0; + +void test2cOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test2OnFailureCalled++; + test2Finished = 1; +} + +void test2dOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test2OnFailureCalled++; + test2Finished = 1; +} + +void test2cOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client d, context %p\n", context); + MQTTAsync c = (MQTTAsync)context; + int rc; + + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + + +int test2dReady = 0; +char willTopic[100]; +char test_topic[50]; + +void test2donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test2dReady = 1; +} + + +void test2dOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test2donSubscribe; + opts.context = c; + + rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test2Finished = 1; +} + +int test2c_connected = 0; + +void test2cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test2c_connected = 1; +} + + +int test2(struct Options options) +{ + char* testname = "test2"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + char *URIs[2] = {"rubbish", options.proxy_connection}; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperty property; + + sprintf(willTopic, "paho-test95-2-%s", unique); + sprintf(clientidc, "paho-test95-2-c-%s", unique); + sprintf(clientidd, "paho-test95-2-d-%s", unique); + sprintf(test_topic, "paho-test95-2-test topic %s", unique); + + test2Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 2 - messages while disconnected with serverURIs"); + fprintf(xml, "message = "will message"; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test2cOnConnect; + opts.onFailure5 = test2cOnFailure; + opts.context = c; + opts.MQTTVersion = MQTTVERSION_5; + opts.cleanstart = 0; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + opts.connectProperties = &props; + opts.serverURIs = URIs; + opts.serverURIcount = 2; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + MQTTProperties_free(&props); + + /* wait for will message */ + while (!test2_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test2c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 3; ++i) + { + char buf[50]; + + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)(strlen(pubmsg.payload) + 1); + pubmsg.qos = i; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + rc = MQTTAsync_reconnect(c); + assert("Good rc from reconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + /* wait for client to be reconnected */ + while (!test2c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test2_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + +/********************************************************************* + + test3: offline buffering - sending messages while disconnected + + 1. call connect + 2. use proxy to disconnect the client + 3. while the client is disconnected, send more messages + 4. when the client auto reconnects, check that those messages are sent + + *********************************************************************/ + +int test3_will_message_received = 0; +int test3_messages_received = 0; + +int test3_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test3_will_message_received = 1; + else + test3_messages_received++; + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +int test3Finished = 0; + +int test3OnFailureCalled = 0; + +void test3cOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test3OnFailureCalled++; + test3Finished = 1; +} + +void test3dOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test3OnFailureCalled++; + test3Finished = 1; +} + +void test3cOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client d, context %p\n", context); + MQTTAsync c = (MQTTAsync)context; + int rc; + + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + + +int test3dReady = 0; +char willTopic[100]; +char test_topic[50]; + +void test3donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test3dReady = 1; +} + + +void test3dOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test3donSubscribe; + opts.context = c; + + rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test3Finished = 1; +} + +int test3c_connected = 0; + +void test3cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test3c_connected = 1; +} + + +int test3(struct Options options) +{ + char* testname = "test3"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperty property; + + sprintf(willTopic, "paho-test95-3-%s", unique); + sprintf(clientidc, "paho-test95-3-c-%s", unique); + sprintf(clientidd, "paho-test95-3-d-%s", unique); + sprintf(test_topic, "paho-test95-3-test topic %s", unique); + + test3Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 3 - messages while disconnected"); + fprintf(xml, "message = "will message"; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test3cOnConnect; + opts.onFailure5 = test3cOnFailure; + opts.context = c; + opts.cleanstart = 0; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + opts.connectProperties = &props; + opts.MQTTVersion = MQTTVERSION_5; + opts.automaticReconnect = 1; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for will message */ + while (!test3_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test3c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 3; ++i) + { + char buf[50]; + + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)(strlen(pubmsg.payload) + 1); + pubmsg.qos = i; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + /* wait for client to be reconnected */ + while (!test3c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test3_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + +/********************************************************************* + + test4: offline buffering - sending messages while disconnected + + 1. call connect + 2. use proxy to disconnect the client + 3. while the client is disconnected, send more messages + 4. when the client auto reconnects, check that those messages are sent + + *********************************************************************/ + +int test4_will_message_received = 0; +int test4_messages_received = 0; + +int test4_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, + message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test4_will_message_received = 1; + else + test4_messages_received++; + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +int test4Finished = 0; + +int test4OnFailureCalled = 0; + +void test4cOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test4OnFailureCalled++; + test4Finished = 1; +} + +void test4dOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test4OnFailureCalled++; + test4Finished = 1; +} + +void test4cOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client d, context %p\n", context); + MQTTAsync c = (MQTTAsync)context; + int rc; + + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + + +int test4dReady = 0; +char willTopic[100]; +char test_topic[50]; + +void test4donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test4dReady = 1; +} + + +void test4dOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test4donSubscribe; + opts.context = c; + + rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test4Finished = 1; +} + +int test4c_connected = 0; + +void test4cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test4c_connected = 1; +} + + +int test4(struct Options options) +{ + char* testname = "test4"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + char *URIs[2] = {"rubbish", options.proxy_connection}; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperty property; + + sprintf(willTopic, "paho-test95-4-%s", unique); + sprintf(clientidc, "paho-test95-4-c-%s", unique); + sprintf(clientidd, "paho-test95-4-d-%s", unique); + sprintf(test_topic, "paho-test95-4-test topic %s", unique); + + test4Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 4 - messages while disconnected with serverURIs"); + fprintf(xml, "message = "will message"; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test4cOnConnect; + opts.onFailure5 = test4cOnFailure; + opts.context = c; + opts.cleanstart = 0; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + opts.connectProperties = &props; + opts.serverURIs = URIs; + opts.serverURIcount = 2; + opts.automaticReconnect = 1; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for will message */ + while (!test4_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test4c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 3; ++i) + { + char buf[50]; + + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)(strlen(pubmsg.payload) + 1); + pubmsg.qos = i; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + /* wait for client to be reconnected */ + while (!test4c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test4_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + + +/********************************************************************* + + test5: offline buffering - check max buffered + + 1. call connect + 2. use proxy to disconnect the client + 3. while the client is disconnected, send more messages + 4. when the client reconnects, check that those messages are sent + + *********************************************************************/ + +int test5_will_message_received = 0; +int test5_messages_received = 0; +int test5Finished = 0; +int test5OnFailureCalled = 0; +int test5c_connected = 0; + +int test5_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test5_will_message_received = 1; + else + test5_messages_received++; + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +void test5cOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test5OnFailureCalled++; + test5Finished = 1; +} + +void test5dOnFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test5OnFailureCalled++; + test5Finished = 1; +} + +void test5cOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client d, context %p\n", context); + MQTTAsync c = (MQTTAsync)context; + int rc; + + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + + +int test5dReady = 0; +char willTopic[100]; +char test_topic[50]; + +void test5donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test5dReady = 1; +} + + +void test5dOnConnect(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int rc; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test5donSubscribe; + opts.context = c; + + rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test5Finished = 1; +} + +void test5cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test5c_connected = 1; +} + + +int test5(struct Options options) +{ + char* testname = "test5"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperty property; + + sprintf(willTopic, "paho-test95-5-%s", unique); + sprintf(clientidc, "paho-test95-5-c-%s", unique); + sprintf(clientidd, "paho-test95-5-d-%s", unique); + sprintf(test_topic, "paho-test95-5-test topic %s", unique); + + test5Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 5 - max buffered"); + fprintf(xml, "message = "will message"; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test5cOnConnect; + opts.onFailure5 = test5cOnFailure; + opts.context = c; + opts.cleanstart = 0; + opts.MQTTVersion = MQTTVERSION_5; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + opts.connectProperties = &props; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for will message */ + while (!test5_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test5c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 5; ++i) + { + char buf[50]; + + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)(strlen(pubmsg.payload) + 1); + pubmsg.qos = i % 3; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + if (i <= 2) + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + else + assert("Bad rc from sendMessage", rc == MQTTASYNC_MAX_BUFFERED_MESSAGES, "rc was %d ", rc); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + rc = MQTTAsync_reconnect(c); + assert("Good rc from reconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + /* wait for client to be reconnected */ + while (!test5c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test5_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + + +int test6(struct Options options) +{ + char* testname = "test6"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + MQTTAsync_token *tokens; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperty property; + + test5_will_message_received = 0; + test5_messages_received = 0; + test5Finished = 0; + test5OnFailureCalled = 0; + test5c_connected = 0; + + sprintf(willTopic, "paho-test95-6-%s", unique); + sprintf(clientidc, "paho-test95-6-c-%s", unique); + sprintf(clientidd, "paho-test95-6-d-%s", unique); + sprintf(test_topic, "paho-test95-6-test topic %s", unique); + + test5Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 6 - max buffered with binary will"); + fprintf(xml, "payload.data = "will message"; + opts.will->payload.len = (int)strlen(opts.will->payload.data) + 1; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test5cOnConnect; + opts.onFailure5 = test5cOnFailure; + opts.context = c; + opts.cleanstart = 0; + opts.MQTTVersion = MQTTVERSION_5; + property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL; + property.value.integer4 = 30; + MQTTProperties_add(&props, &property); + opts.connectProperties = &props; + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + MQTTProperties_free(&props); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + /* wait for will message */ + while (!test5_will_message_received && ++count < 10000) + MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered"); + + test5c_connected = 0; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 5; ++i) + { + char buf[50]; + + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + sprintf(buf, "QoS %d message", i); + pubmsg.payload = buf; + pubmsg.payloadlen = (int)(strlen(pubmsg.payload) + 1); + pubmsg.qos = i % 3; + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts); + if (i <= 2) + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + else + assert("Bad rc from sendMessage", rc == MQTTASYNC_MAX_BUFFERED_MESSAGES, "rc was %d ", rc); + } + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + rc = MQTTAsync_reconnect(c); + assert("Good rc from reconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + /* wait for client to be reconnected */ + while (!test5c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test5_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); + + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + +exit: + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + + +/********************************************************************* + +Test7: Fill up TCP buffer with QoS 0 messages + +*********************************************************************/ +int test7c_connected = 0; +int test7_will_message_received = 0; +int test7_messages_received = 0; +int test7Finished = 0; +int test7OnFailureCalled = 0; +int test7dReady = 0; + +int test7_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message) +{ + MQTTAsync c = (MQTTAsync)context; + static int message_count = 0; + + MyLog(LOGA_DEBUG, "Message received on topic %s, \"%.*s\"", topicName, message->payloadlen, message->payload); + + if (memcmp(message->payload, "will message", message->payloadlen) == 0) + test7_will_message_received = 1; + else + test7_messages_received++; + + MQTTAsync_freeMessage(&message); + MQTTAsync_free(topicName); + + return 1; +} + +void test7cConnected(void* context, char* cause) +{ + MQTTAsync c = (MQTTAsync)context; + + MyLog(LOGA_DEBUG, "In connected callback for client c, context %p\n", context); + test7c_connected = 1; +} + +void test7cOnConnectFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In c connect onFailure callback, context %p", context); + + test7OnFailureCalled++; + test7Finished = 1; +} + +void test7cOnConnectSuccess(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + + /* send a message to the proxy to break the connection */ + pubmsg.payload = "TERMINATE"; + pubmsg.payloadlen = (int)strlen(pubmsg.payload); + pubmsg.qos = 0; + pubmsg.retained = 0; + //rc = MQTTAsync_sendMessage(c, "MQTTSAS topic", &pubmsg, NULL); + //assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); +} + +void test7dOnConnectFailure(void* context, MQTTAsync_failureData5* response) +{ + MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context); + + test7OnFailureCalled++; + test7Finished = 1; +} + + +void test7donSubscribe(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In subscribe onSuccess callback for client d, %p granted qos %d", c, + response->reasonCode); + test7dReady = 1; +} + + +void test7dOnConnectSuccess(void* context, MQTTAsync_successData5* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; + int qoss[2] = {2, 2}; + char* topics[2] = {willTopic, test_topic}; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback for client c, context %p\n", context); + opts.onSuccess5 = test7donSubscribe; + opts.context = c; + + //rc = MQTTAsync_subscribeMany(c, 2, topics, qoss, &opts); + //assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + //if (rc != MQTTASYNC_SUCCESS) + // test5Finished = 1; + test7dReady = 1; +} + + +int test7(struct Options options) +{ + char* testname = "test7"; + int subsqos = 2; + MQTTAsync c, d; + MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5; + MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer; + //MQTTAsync_createOptions createOptions = MQTTAsync_createOptions_initializer; + int rc = 0; + int count = 0; + char clientidc[50]; + char clientidd[50]; + int i = 0; + + test7_will_message_received = 0; + test7_messages_received = 0; + test7Finished = 0; + test7OnFailureCalled = 0; + test7c_connected = 0; + + sprintf(willTopic, "paho-test95-7-%s", unique); + sprintf(clientidc, "paho-test9-7-c-%s", unique); + sprintf(clientidd, "paho-test9-7-d-%s", unique); + sprintf(test_topic, "longer paho-test9-7-test topic %s", unique); + + test7Finished = 0; + failures = 0; + MyLog(LOGA_INFO, "Starting Offline buffering 7 - fill TCP buffer"); + fprintf(xml, "payload.data = "will message"; + opts.will->payload.len = (int)strlen(opts.will->payload.data) + 1; + opts.will->qos = 1; + opts.will->retained = 0; + opts.will->topicName = willTopic; + opts.onSuccess5 = test7cOnConnectSuccess; + opts.onFailure5 = test7cOnConnectFailure; + opts.context = c; + opts.cleansession = 0; + /*opts.automaticReconnect = 1; + opts.minRetryInterval = 3; + opts.maxRetryInterval = 6;*/ + + MyLog(LOGA_DEBUG, "Connecting client c"); + rc = MQTTAsync_connect(c, &opts); + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + if (rc != MQTTASYNC_SUCCESS) + { + failures++; + goto exit; + } + + count = 0; + while (!test7c_connected && ++count < 10000) + MySleep(100); + assert("Count should be less than 10000", count < 10000, "count was %d", count); /* wrong */ + + /* wait for will message */ + //while (test7_will_message_received == 0 && ++count < 10000) + // MySleep(100); + + MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered by TCP"); + + test7c_connected = 0; + char buf[5000000]; + /* send some messages. Then reconnect (check connected callback), and check that those messages are received */ + for (i = 0; i < 50000; ++i) + { + MQTTAsync_message pubmsg = MQTTAsync_message_initializer; + MQTTAsync_responseOptions pubopts = MQTTAsync_responseOptions_initializer; + pubmsg.qos = 0; /*i % 3;*/ + sprintf(buf, "QoS %d message", pubmsg.qos); + pubmsg.payload = buf; + pubmsg.payloadlen = 5000000; //(int)(strlen(pubmsg.payload) + 1); + pubmsg.retained = 0; + rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &pubopts); + assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc); + if (rc != 0) + { + //MyLog(LOGA_DEBUG, "Connecting client c"); + //rc = MQTTAsync_connect(c, &opts); + //MySleep(1000); + break; + } + } + +#if 0 + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 3", i == 3, "i was %d ", i); + + rc = MQTTAsync_reconnect(c); + assert("Good rc from reconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + /* wait for client to be reconnected */ + while (!test5c_connected && ++count < 10000) + MySleep(100); + + /* wait for success or failure callback */ + while (test5_messages_received < 3 && ++count < 10000) + MySleep(100); + + rc = MQTTAsync_getPendingTokens(c, &tokens); + assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + i = 0; + if (tokens) + { + while (tokens[i] != -1) + ++i; + MQTTAsync_free(tokens); + } + assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i); +#endif + +exit: + rc = MQTTAsync_disconnect(c, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + rc = MQTTAsync_disconnect(d, NULL); + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); + + MySleep(200); + MQTTAsync_destroy(&c); + MQTTAsync_destroy(&d); + MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", testname, tests, failures); + write_test_result(); + return failures; +} + + + +void handleTrace(enum MQTTASYNC_TRACE_LEVELS level, char* message) +{ + printf("%s\n", message); +} + + +int main(int argc, char** argv) +{ + int* numtests = &tests; + int rc = 0; + int (*tests[])() = { NULL, test1, test2, test3, test4, test5, test6 }; + time_t randtime; + + srand((unsigned) time(&randtime)); + sprintf(unique, "%u", rand()); + MyLog(LOGA_INFO, "Random prefix/suffix is %s", unique); + + xml = fopen("TEST-test9.xml", "w"); + fprintf(xml, "\n", (int)(ARRAY_SIZE(tests) - 1)); + + MQTTAsync_setTraceCallback(handleTrace); + getopts(argc, argv); + + if (options.test_no == 0) + { /* run all the tests */ + for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no) + { + failures = 0; + MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR); + rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */ + } + } + else + { + MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR); + rc = tests[options.test_no](options); /* run just the selected test */ + } + + MyLog(LOGA_INFO, "Total tests run: %d", *numtests); + if (rc == 0) + MyLog(LOGA_INFO, "verdict pass"); + else + MyLog(LOGA_INFO, "verdict fail"); + + fprintf(xml, "\n"); + fclose(xml); + + return rc; +}