diff --git a/Makefile b/Makefile index 6cb0d501..e82dda83 100755 --- a/Makefile +++ b/Makefile @@ -90,7 +90,7 @@ SYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_C}} SAMPLE_FILES_A = paho_c_pub paho_c_sub MQTTAsync_subscribe MQTTAsync_publish ASYNC_SAMPLES = ${addprefix ${blddir}/samples/,${SAMPLE_FILES_A}} -TEST_FILES_C = test1 test15 test2 sync_client_test test_mqtt4sync +TEST_FILES_C = test1 test15 test2 sync_client_test test_mqtt4sync test10 SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}} TEST_FILES_CS = test3 diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index de4b55df..d9c92498 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -48,6 +48,7 @@ SET(common_src Heap.c LinkedList.c MQTTProperties.c + MQTTReasonCodes.c Base64.c SHA1.c WebSocket.c diff --git a/src/MQTTClient.c b/src/MQTTClient.c index 2510168f..a0aa54eb 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -205,6 +205,12 @@ typedef struct MQTTClient_deliveryComplete* dc; void* context; + MQTTClient_disconnected* disconnected; + void* disconnected_context; /* the context to be associated with the disconnected callback*/ + + MQTTClient_authHandle* auth_handle; + void* auth_handle_context; /* the context to be associated with the authHandle callback*/ + sem_type connect_sem; int rc; /* getsockopt return code in connect */ sem_type connack_sem; @@ -214,6 +220,13 @@ typedef struct } MQTTClients; +struct props_rc_parms +{ + MQTTClients* m; + MQTTProperties* properties; + enum MQTTReasonCodes reasonCode; +}; + void MQTTClient_sleep(long milliseconds) { FUNC_ENTRY; @@ -546,6 +559,18 @@ void MQTTClient_free(void* memory) } +DLLExport void MQTTResponse_free(MQTTResponse response) +{ + FUNC_ENTRY; + if (response.properties) + { + MQTTProperties_free(response.properties); + free(response.properties); + } + FUNC_EXIT; +} + + static int MQTTClient_deliverMessage(int rc, MQTTClients* m, char** topicName, int* topicLen, MQTTClient_message** message) { qEntry* qe = (qEntry*)(m->c->messageQueue->first->content); @@ -594,6 +619,81 @@ static thread_return_type WINAPI connectionLost_call(void* context) } +int MQTTClient_setDisconnected(MQTTClient handle, void* context, MQTTClient_disconnected* disconnected) +{ + int rc = MQTTCLIENT_SUCCESS; + MQTTClients* m = handle; + + FUNC_ENTRY; + Thread_lock_mutex(mqttclient_mutex); + + if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS) + rc = MQTTCLIENT_FAILURE; + else + { + m->disconnected_context = context; + m->disconnected = disconnected; + } + + Thread_unlock_mutex(mqttclient_mutex); + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Wrapper function to call disconnected on a separate thread. A separate thread is needed to allow the + * disconnected function to make API calls (e.g. connect) + * @param context a pointer to the relevant client + * @return thread_return_type standard thread return value - not used here + */ +static thread_return_type WINAPI call_disconnected(void* context) +{ + struct props_rc_parms* pr = (struct props_rc_parms*)context; + + (*(pr->m->disconnected))(pr->m->disconnected_context, pr->properties, pr->reasonCode); + MQTTProperties_free(pr->properties); + return 0; +} + + +int MQTTClient_setAuthHandle(MQTTClient handle, void* context, MQTTClient_authHandle* auth_handle) +{ + int rc = MQTTCLIENT_SUCCESS; + MQTTClients* m = handle; + + FUNC_ENTRY; + Thread_lock_mutex(mqttclient_mutex); + + if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS) + rc = MQTTCLIENT_FAILURE; + else + { + m->auth_handle_context = context; + m->auth_handle = auth_handle; + } + + Thread_unlock_mutex(mqttclient_mutex); + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Wrapper function to call authHandle on a separate thread. A separate thread is needed to allow the + * disconnected function to make API calls (e.g. MQTTClient_auth) + * @param context a pointer to the relevant client + * @return thread_return_type standard thread return value - not used here + */ +static thread_return_type WINAPI call_auth_handle(void* context) +{ + struct props_rc_parms* pr = (struct props_rc_parms*)context; + + (*(pr->m->auth_handle))(pr->m->auth_handle_context, pr->properties, pr->reasonCode); + return 0; +} + + /* This is the thread function that handles the calling of callback functions if set */ static thread_return_type WINAPI MQTTClient_run(void* n) { @@ -699,6 +799,33 @@ static thread_return_type WINAPI MQTTClient_run(void* n) m->pack = pack; Thread_post_sem(m->unsuback_sem); } + else if (m->c->MQTTVersion >= MQTTVERSION_5) + { + if (pack->header.bits.type == DISCONNECT && m->disconnected) + { + struct props_rc_parms dp; + Ack* disc = (Ack*)pack; + + dp.m = m; + dp.properties = &disc->properties; + dp.reasonCode = disc->rc; + free(pack); + Log(TRACE_MIN, -1, "Calling disconnected for client %s", m->c->clientID); + Thread_start(call_disconnected, &dp); + } + if (pack->header.bits.type == AUTH && m->auth_handle) + { + struct props_rc_parms dp; + Ack* disc = (Ack*)pack; + + dp.m = m; + dp.properties = &disc->properties; + dp.reasonCode = disc->rc; + free(pack); + Log(TRACE_MIN, -1, "Calling auth_handle for client %s", m->c->clientID); + Thread_start(call_auth_handle, &dp); + } + } } else if (m->c->connect_state == TCP_IN_PROGRESS && !Thread_check_sem(m->connect_sem)) { @@ -1112,7 +1239,10 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c rc = MQTTCLIENT_DISCONNECTED; } if (m->c->MQTTVersion == MQTTVERSION_5) - resp.properties = &connack->properties; + { + resp.properties = malloc(sizeof(MQTTProperties)); + *resp.properties = MQTTProperties_copy(&connack->properties); + } } MQTTPacket_freeConnack(connack); m->pack = NULL; @@ -1930,7 +2060,7 @@ MQTTResponse MQTTClient_publishMessage5(MQTTClient handle, const char* topicName goto exit; } - if (message->struct_version == 1) + if (message->struct_version >= 1) props = &message->properties; rc = MQTTClient_publish5(handle, topicName, message->payloadlen, message->payload, diff --git a/src/MQTTClient.h b/src/MQTTClient.h index b4f51a5e..a99e0ece 100644 --- a/src/MQTTClient.h +++ b/src/MQTTClient.h @@ -414,6 +414,67 @@ DLLExport int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClie MQTTClient_messageArrived* ma, MQTTClient_deliveryComplete* dc); +/** + * This is a callback function, which will be called when the client + * library successfully connects. This is superfluous when the connection + * is made in response to a MQTTAsync_connect call, because the onSuccess + * callback can be used. It is intended for use when automatic reconnect + * is enabled, so that when a reconnection attempt succeeds in the background, + * the application is notified and can take any required actions. + * @param context A pointer to the context value originally passed to + * MQTTAsync_setCallbacks(), which contains any application-specific context. + * @param cause The reason for the disconnection. + * Currently, cause is always set to NULL. + */ +typedef void MQTTClient_disconnected(void* context, MQTTProperties* properties, + enum MQTTReasonCodes reasonCode); + +/** + * Sets the MQTTClient_disconnected() callback function for a client. + * @param handle A valid client handle from a successful call to + * MQTTAsync_create(). + * @param context A pointer to any application-specific context. The + * the context pointer is passed to each of the callback functions to + * provide access to the context information in the callback. + * @param co A pointer to an MQTTAsync_connected() callback + * function. NULL removes the callback setting. + * @return ::MQTTASYNC_SUCCESS if the callbacks were correctly set, + * ::MQTTASYNC_FAILURE if an error occurred. + */ +DLLExport int MQTTClient_setDisconnected(MQTTClient handle, void* context, MQTTClient_disconnected* co); + + +/** + * This is a callback function, which will be called when the client + * library successfully connects. This is superfluous when the connection + * is made in response to a MQTTAsync_connect call, because the onSuccess + * callback can be used. It is intended for use when automatic reconnect + * is enabled, so that when a reconnection attempt succeeds in the background, + * the application is notified and can take any required actions. + * @param context A pointer to the context value originally passed to + * MQTTAsync_setCallbacks(), which contains any application-specific context. + * @param cause The reason for the disconnection. + * Currently, cause is always set to NULL. + */ +typedef void MQTTClient_authHandle(void* context, MQTTProperties* properties, + enum MQTTReasonCodes reasonCode); + + +/** + * Sets the MQTTClient_authHandle() callback function for a client. + * @param handle A valid client handle from a successful call to + * MQTTAsync_create(). + * @param context A pointer to any application-specific context. The + * the context pointer is passed to each of the callback functions to + * provide access to the context information in the callback. + * @param co A pointer to an MQTTAsync_connected() callback + * function. NULL removes the callback setting. + * @return ::MQTTASYNC_SUCCESS if the callbacks were correctly set, + * ::MQTTASYNC_FAILURE if an error occurred. + */ +DLLExport int MQTTClient_setAuthHandle(MQTTClient handle, void* context, MQTTClient_authHandle* ah); + + /** * This function creates an MQTT client ready for connection to the * specified server and using the specified persistent storage (see @@ -785,6 +846,8 @@ typedef struct MQTTResponse MQTTProperties* properties; /* optional */ } MQTTResponse; +DLLExport void MQTTResponse_free(MQTTResponse response); + DLLExport MQTTResponse MQTTClient_connect5(MQTTClient handle, MQTTClient_connectOptions* options, MQTTProperties* connectProperties, MQTTProperties* willProperties); diff --git a/src/MQTTPacket.c b/src/MQTTPacket.c index 93475653..8670b735 100644 --- a/src/MQTTPacket.c +++ b/src/MQTTPacket.c @@ -85,7 +85,8 @@ pf new_packets[] = MQTTPacket_unsuback, /**< UNSUBACK */ MQTTPacket_header_only, /**< PINGREQ */ MQTTPacket_header_only, /**< PINGRESP */ - MQTTPacket_header_only /**< DISCONNECT */ + MQTTPacket_ack, /**< DISCONNECT */ + MQTTPacket_ack /**< AUTH */ }; @@ -132,7 +133,9 @@ void* MQTTPacket_Factory(int MQTTVersion, networkHandles* net, int* error) else { ptype = header.bits.type; - if (ptype < CONNECT || ptype > DISCONNECT || new_packets[ptype] == NULL) + if (ptype < CONNECT || (MQTTVersion < MQTTVERSION_5 && ptype >= DISCONNECT) || + (MQTTVersion >= MQTTVERSION_5 && ptype > AUTH) || + new_packets[ptype] == NULL) Log(TRACE_MIN, 2, NULL, ptype); else { @@ -723,7 +726,8 @@ void* MQTTPacket_ack(int MQTTVersion, unsigned char aHeader, char* data, size_t FUNC_ENTRY; pack->MQTTVersion = MQTTVersion; pack->header.byte = aHeader; - pack->msgId = readInt(&curdata); + if (pack->header.bits.type != DISCONNECT) + pack->msgId = readInt(&curdata); if (MQTTVersion >= MQTTVERSION_5) { MQTTProperties props = MQTTProperties_initializer; @@ -736,8 +740,6 @@ void* MQTTPacket_ack(int MQTTVersion, unsigned char aHeader, char* data, size_t if (datalen > 3) { - pack->properties.max_count = 10; - pack->properties.array = malloc(sizeof(MQTTProperty) * pack->properties.max_count); if (MQTTProperties_read(&pack->properties, &curdata, enddata) != 1) { free(pack); diff --git a/src/MQTTProperties.c b/src/MQTTProperties.c index 0a3e7821..b46b459f 100644 --- a/src/MQTTProperties.c +++ b/src/MQTTProperties.c @@ -25,7 +25,7 @@ #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0])) -struct nameToType +static struct nameToType { enum PropertyNames name; enum PropertyTypes type; diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index 40f576d5..2dc0907a 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -163,6 +163,7 @@ int MQTTProtocol_startPublish(Clients* pubclient, Publish* publish, int qos, int p.payload = (*mm)->publish->payload; p.topic = (*mm)->publish->topic; p.properties = (*mm)->properties; + p.MQTTVersion = (*mm)->MQTTVersion; } rc = MQTTProtocol_startPublishCommon(pubclient, &p, qos, retained); FUNC_EXIT_RC(rc); diff --git a/src/MQTTReasonCodes.c b/src/MQTTReasonCodes.c new file mode 100644 index 00000000..40ac80af --- /dev/null +++ b/src/MQTTReasonCodes.c @@ -0,0 +1,100 @@ +/******************************************************************************* + * Copyright (c) 2017, 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 + *******************************************************************************/ + +#include "MQTTReasonCodes.h" + +#include "Heap.h" +#include "StackTrace.h" + +#include + +#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0])) + +static struct { + enum MQTTReasonCodes value; + const char* name; +} nameToString[] = +{ + {SUCCESS, "SUCCESS"}, + {NORMAL_DISCONNECTION, "Normal disconnection"}, + {GRANTED_QOS_0, "Granted QoS 0"}, + {GRANTED_QOS_1, "Granted QoS 1"}, + {GRANTED_QOS_2, "Granted QoS 2"}, + {DISCONNECT_WITH_WILL_MESSAGE, "Disconnect with Will Messge"}, + {NO_MATCHING_SUBSCRIBERS, "No matching subscribers"}, + {NO_SUBSCRIPTION_FOUND, "No subscription found"}, + {CONTINUE_AUTHENTICATION, "Continue authentication"}, + {RE_AUTHENTICATE, "Re-authenticate"}, + {UNSPECIFIED_ERROR, "Unspecified error"}, + {MALFORMED_PACKET, "Malformed Packet"}, + {PROTOCOL_ERROR, "Protocol error"}, + {IMPLEMENTATION_SPECIFIC_ERROR, "Implementation specific error"}, + {UNSUPPORTED_PROTOCOL_VERSION, "Unsupported Protocol Version"}, + {CLIENT_IDENTIFIER_NOT_VALID, "Client Identifier not valid"}, + {BAD_USER_NAME_OR_PASSWORD, "Bad User Name or Password"}, + {NOT_AUTHORIZED, "Not authorized"}, + {SERVER_UNAVAILABLE, "Server unavailable"}, + {SERVER_BUSY, "Server busy"}, + {BANNED, "Banned"}, + {SERVER_SHUTTING_DOWN, "Server shutting down"}, + {BAD_AUTHENTICATION_METHOD, "Bad authentication method"}, + {KEEP_ALIVE_TIMEOUT, "Keep Alive timeout"}, + {SESSION_TAKEN_OVER, "Session taken over"}, + {TOPIC_FILTER_INVALID, "Topic filter invalid"}, + {TOPIC_NAME_INVALID, "Topic name invalid"}, + {PACKET_IDENTIFIER_IN_USE, "Packet Identifier in use"}, + {PACKET_IDENTIFIER_NOT_FOUND, "Packet Identifier not found"}, + {RECEIVE_MAXIMUM_EXCEEDED, "Receive Maximum exceeded"}, + {TOPIC_ALIAS_INVALID, "Topic Alias invalid"}, + {PACKET_TOO_LARGE, "Packet too large"}, + {MESSAGE_RATE_TOO_HIGH, "Message rate too high"}, + {QUOTA_EXCEEDED, "Quota exceeded"}, + {ADMINISTRATIVE_ACTION, "Administrative action"}, + {PAYLOAD_FORMAT_INVALID, "Payload format invalid"}, + {RETAIN_NOT_SUPPORTED, "Retain not supported"}, + {QOS_NOT_SUPPORTED, "QoS not supported"}, + {USE_ANOTHER_SERVER, "Use another server"}, + {SERVER_MOVED, "Server moved"}, + {SHARED_SUBSCRIPTIONS_NOT_SUPPORTED, "Shared subscriptions not supported"}, + {CONNECTION_RATE_EXCEEDED, "Connection rate exceeded"}, + {MAXIMUM_CONNECT_TIME, "Maximum connect time"}, + {SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED, "Subscription Identifiers not supported"}, + {WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED, "Wildcard Subscriptions not supported"} +}; + +const char* MQTTReasonCodeString(enum MQTTReasonCodes value) +{ + int i = 0; + const char* result = NULL; + + for (i = 0; i < ARRAY_SIZE(nameToString); ++i) + { + if (nameToString[i].value == value) + { + result = nameToString[i].name; + break; + } + } + return result; +} + + + + + + + + diff --git a/src/MQTTReasonCodes.h b/src/MQTTReasonCodes.h index d0928421..a7f37173 100644 --- a/src/MQTTReasonCodes.h +++ b/src/MQTTReasonCodes.h @@ -58,11 +58,21 @@ enum MQTTReasonCodes { QOS_NOT_SUPPORTED = 155, USE_ANOTHER_SERVER = 156, SERVER_MOVED = 157, - SHARED_SUBSCRIPTION_NOT_SUPPORTED = 158, + SHARED_SUBSCRIPTIONS_NOT_SUPPORTED = 158, CONNECTION_RATE_EXCEEDED = 159, MAXIMUM_CONNECT_TIME = 160, SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED = 161, - WILDCARD_SUBSCRIPTION_NOT_SUPPORTED = 162 + WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED = 162 }; +#if defined(WIN32) || defined(WIN64) + #define DLLImport __declspec(dllimport) + #define DLLExport __declspec(dllexport) +#else + #define DLLImport extern + #define DLLExport __attribute__ ((visibility ("default"))) +#endif + +DLLExport const char* MQTTReasonCodeString(enum MQTTReasonCodes); + #endif diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index c88c3cb4..dd251f1e 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -564,6 +564,26 @@ SET_TESTS_PROPERTIES( PROPERTIES TIMEOUT 540 ) +ADD_EXECUTABLE( + test10 + test10.c +) + +TARGET_LINK_LIBRARIES( + test10 + paho-mqtt3c +) + +ADD_TEST( + NAME test10-1-client_topic_aliases + COMMAND "test10" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY} +) + +SET_TESTS_PROPERTIES( + test10-1-client_topic_aliases + PROPERTIES TIMEOUT 540 +) + ADD_EXECUTABLE( test_issue373 test_issue373.c diff --git a/test/test10.c b/test/test10.c new file mode 100644 index 00000000..8275ffa4 --- /dev/null +++ b/test/test10.c @@ -0,0 +1,611 @@ +/******************************************************************************* + * 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 + * 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 - MQTT 5.0 support + *******************************************************************************/ + + +/** + * @file + * MQTT V5 specific tests for the MQTT C client + * + * - topic aliases + * - subscription ids + * - session expiry + * - payload format + * - flow control + * - QoS 2 exchange termination + * - request/response + * - shared subscriptions + * - server initiated disconnect + * - auth packets + * - server assigned clientid returned in a property + * - server defined keepalive + * - subscribe failure + */ + + +#include "MQTTClient.h" +#include +#include + +#if !defined(_WINDOWS) + #include + #include + #include + #include +#else + #include + #define setenv(a, b, c) _putenv_s(a, b) +#endif + +#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** haconnections; + char* proxy_connection; + int hacount; + int verbose; + int test_no; + int MQTTVersion; + int iterations; +} options = +{ + "tcp://localhost:1883", + NULL, + "tcp://localhost:1884", + 0, + 0, + 0, + MQTTVERSION_5, + 1, +}; + +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]; + printf("\nSetting connection to %s\n", options.connection); + } + else + usage(); + } + else if (strcmp(argv[count], "--haconnections") == 0) + { + if (++count < argc) + { + char* tok = strtok(argv[count], " "); + options.hacount = 0; + options.haconnections = malloc(sizeof(char*) * 5); + while (tok) + { + options.haconnections[options.hacount] = malloc(strlen(tok) + 1); + strcpy(options.haconnections[options.hacount], tok); + options.hacount++; + tok = strtok(NULL, " "); + } + } + 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], "--MQTTversion") == 0) + { + if (++count < argc) + { + options.MQTTVersion = atoi(argv[count]); + printf("setting MQTT version to %d\n", options.MQTTVersion); + } + else + usage(); + } + else if (strcmp(argv[count], "--iterations") == 0) + { + if (++count < argc) + options.iterations = atoi(argv[count]); + else + usage(); + } + else if (strcmp(argv[count], "--verbose") == 0) + { + options.verbose = 1; + printf("\nSetting verbose on\n"); + } + 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; + + strcpy(msg_buf, ""); + ftime(&ts); + localtime_r(&ts.time, &timeinfo); + 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); +} + + +#if defined(WIN32) || defined(_WINDOWS) +#define mqsleep(A) Sleep(1000*A) +#define START_TIME_TYPE DWORD +static DWORD start_time = 0; +START_TIME_TYPE start_clock(void) +{ + return GetTickCount(); +} +#elif defined(AIX) +#define mqsleep sleep +#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 mqsleep sleep +#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) + +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\n", 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 PROPERTY_TYPE_BYTE: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.byte); + break; + case TWO_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer2); + break; + case FOUR_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4); + break; + case VARIABLE_BYTE_INTEGER: + MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4); + break; + case BINARY_DATA: + case UTF_8_ENCODED_STRING: + MyLog(LOGA_INFO, "Property name value %s %.*s", name, + props->array[i].value.data.len, props->array[i].value.data.data); + break; + case 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; + } + } +} + +struct aa +{ + int disconnected; +} test_topic_aliases_globals = +{ + 0, +}; + +void disconnected(void* context, MQTTProperties* props, enum MQTTReasonCodes rc) +{ + MQTTClient c = (MQTTClient)context; + printf("Callback: disconnected, reason code \"%s\"\n", MQTTReasonCodeString(rc)); + logProperties(props); + test_topic_aliases_globals.disconnected = 1; +} + +static int messages_arrived = 0; + +int messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message) +{ + MyLog(LOGA_DEBUG, "Callback: message received on topic %s is %.*s.", + topicName, message->payloadlen, (char*)(message->payload)); + + assert("Message structure version should be 1", message->struct_version == 1, + "message->struct_version was %d", message->struct_version); + if (message->struct_version == 1) + { + const int props_count = 0; + + assert("Properties count should be 0", message->properties.count == props_count, + "Properties count was %d\n", message->properties.count); + logProperties(&message->properties); + } + messages_arrived++; + + MQTTClient_free(topicName); + MQTTClient_freeMessage(&message); + return 1; +} + + +int test_client_topic_aliases(struct Options options) +{ + int subsqos = 2; + MQTTClient c; + MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer; + MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer; + MQTTProperties props = MQTTProperties_initializer; + MQTTProperties connect_props = MQTTProperties_initializer; + MQTTProperty property; + MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer; + MQTTClient_message pubmsg = MQTTClient_message_initializer; + MQTTResponse response = {SUCCESS, NULL}; + MQTTClient_deliveryToken dt; + int rc = 0; + int count = 0; + char* test_topic = "test_client_topic_aliases"; + int topicAliasMaximum = 0; + + fprintf(xml, " 0", topicAliasMaximum > 0, "topicAliasMaximum was %d", topicAliasMaximum); + + /* subscribe to a topic */ + response = MQTTClient_subscribe5(c, test_topic, 2, NULL, NULL); + assert("Good rc from subscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + + /* then publish to the topic */ + MQTTProperties_free(&pubmsg.properties); + property.identifier = TOPIC_ALIAS; + property.value.integer2 = 1; + MQTTProperties_add(&pubmsg.properties, &property); + + messages_arrived = 0; + response = MQTTClient_publishMessage5(c, test_topic, &pubmsg, &dt); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + + /* should get a response */ + while (messages_arrived == 0 && ++count < 10) + { +#if defined(WIN32) + Sleep(1000); +#else + usleep(1000000L); +#endif + } + assert("1 message should have arrived", messages_arrived == 1, "was %d", messages_arrived); + + /* now publish to the topic alias only */ + messages_arrived = 0; + response = MQTTClient_publishMessage5(c, "", &pubmsg, &dt); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + + /* should get a response */ + while (messages_arrived == 0 && ++count < 10) + { +#if defined(WIN32) + Sleep(1000); +#else + usleep(1000000L); +#endif + } + assert("1 message should have arrived", messages_arrived == 1, "was %d", messages_arrived); + + rc = MQTTClient_disconnect5(c, 1000, SUCCESS, NULL); + + /* Reconnect. Topic aliases should be deleted, but not subscription */ + opts.cleansession = 0; + response = MQTTClient_connect5(c, &opts, NULL, NULL); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + MQTTResponse_free(response); + if (response.reasonCode != MQTTCLIENT_SUCCESS) + goto exit; + + /* then publish to the topic */ + MQTTProperties_free(&pubmsg.properties); + messages_arrived = 0; + response = MQTTClient_publishMessage5(c, test_topic, &pubmsg, &dt); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + + /* should get a response */ + while (messages_arrived == 0 && ++count < 10) + { +#if defined(WIN32) + Sleep(1000); +#else + usleep(1000000L); +#endif + } + assert("1 message should have arrived", messages_arrived == 1, "was %d", messages_arrived); + + /* now publish to the topic alias only */ + messages_arrived = 0; + property.identifier = TOPIC_ALIAS; + property.value.integer2 = 1; + MQTTProperties_add(&pubmsg.properties, &property); + response = MQTTClient_publishMessage5(c, "", &pubmsg, &dt); + assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + + /* should not get a response */ + while (messages_arrived == 0 && ++count < 10) + { +#if defined(WIN32) + Sleep(1000); +#else + usleep(1000000L); +#endif + } + assert("No message should have arrived", messages_arrived == 0, "was %d", messages_arrived); + + MQTTProperties_free(&pubmsg.properties); + MQTTProperties_free(&props); + MQTTProperties_free(&connect_props); + MQTTClient_destroy(&c); + +exit: + MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int main(int argc, char** argv) +{ + int rc = 0; + int (*tests[])() = {NULL, test_client_topic_aliases}; + int i; + + xml = fopen("TEST-test1.xml", "w"); + fprintf(xml, "\n", (int)(ARRAY_SIZE(tests) - 1)); + + setenv("MQTT_C_CLIENT_TRACE", "ON", 1); + setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1); + + getopts(argc, argv); + + for (i = 0; i < options.iterations; ++i) + { + if (options.test_no == 0) + { /* run all the tests */ + for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no) + rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */ + } + else + rc = tests[options.test_no](options); /* run just the selected test */ + } + + if (rc == 0) + MyLog(LOGA_INFO, "verdict pass"); + else + MyLog(LOGA_INFO, "verdict fail"); + + fprintf(xml, "\n"); + fclose(xml); + return rc; +} diff --git a/test/test15.c b/test/test15.c index 3bae917a..63e17eab 100644 --- a/test/test15.c +++ b/test/test15.c @@ -58,9 +58,9 @@ struct Options int iterations; } options = { - "tcp://iot.eclipse.org:1883", - NULL, "tcp://localhost:1883", + NULL, + "tcp://localhost:1884", 0, 0, 0, @@ -291,7 +291,7 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic) char* topicName = NULL; int topicLen; int i = 0; - int iterations = 1; //50; + int iterations = 50; int rc; MQTTResponse resp; MQTTProperty property; @@ -463,7 +463,7 @@ int test1(struct Options options) if (response.properties) { logProperties(response.properties); - MQTTProperties_free(response.properties); + MQTTResponse_free(response); } subopts.retainAsPublished = 1; @@ -508,6 +508,7 @@ int test1(struct Options options) /* Just to make sure we can connect again */ response = MQTTClient_connect5(c, &opts, NULL, NULL); assert("Connect successful", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode); + MQTTResponse_free(response); rc = MQTTClient_disconnect5(c, 0, SUCCESS, &props); assert("Disconnect successful", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); @@ -670,7 +671,9 @@ int test2(struct Options options) MyLog(LOGA_DEBUG, "Connecting"); response = MQTTClient_connect5(c, &opts, &props, &willProps); - assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc); + assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, + "rc was %d", response.reasonCode); + MQTTResponse_free(response); if (rc != MQTTCLIENT_SUCCESS) goto exit; @@ -719,6 +722,7 @@ int test3(struct Options options) MQTTClient c; MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer; MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer; + MQTTResponse response; fprintf(xml, "qos = 1; opts.will->retained = 0; opts.will->topicName = "will topic";*/ - rc = MQTTClient_connect(c, &opts); + response = MQTTClient_connect5(c, &opts, NULL, NULL); //assert("Not authorized", rc == 5, "rc was %d\n", rc); + MQTTResponse_free(response); #if 0 /* successful connection (RC = 0) */ @@ -825,6 +830,7 @@ int test4_run(int qos) 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; @@ -879,6 +885,7 @@ int test4_run(int qos) opts.cleansession = 0; response = MQTTClient_connect5(c, &opts, NULL, 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; @@ -986,6 +993,7 @@ int test5(struct Options options) MyLog(LOGA_DEBUG, "Connecting"); 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) { MQTTClient_destroy(&c); @@ -1101,6 +1109,7 @@ int test6(struct Options options) /* Connect to the broker */ response = MQTTClient_connect5(test6_c1, &opts, NULL, NULL); assert("good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc); + MQTTResponse_free(response); if (rc != MQTTCLIENT_SUCCESS) goto exit; @@ -1117,6 +1126,7 @@ int test6(struct Options options) opts2.cleansession = 1; MyLog(LOGA_INFO, "Connecting Client_2 ..."); response = MQTTClient_connect5(test6_c2, &opts2, NULL, NULL); + MQTTResponse_free(response); assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); response = MQTTClient_subscribe5(test6_c2, test6_will_topic, 2, NULL, NULL); @@ -1211,6 +1221,7 @@ int test6a(struct Options options) response = MQTTClient_connect5(test6_c1, &opts, NULL, NULL); assert("good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); + MQTTResponse_free(response); if (response.reasonCode != MQTTCLIENT_SUCCESS) goto exit; @@ -1227,6 +1238,7 @@ int test6a(struct Options options) opts2.cleansession = 1; MyLog(LOGA_INFO, "Connecting Client_2 ..."); response = MQTTClient_connect5(test6_c2, &opts2, NULL, NULL); + MQTTResponse_free(response); assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d\n", response.reasonCode); response = MQTTClient_subscribe5(test6_c2, test6_will_topic, 2, NULL, NULL);