natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MQTTClient v5 specific tests. Receive disconnect #469. Client topic aliases #472

This commit is contained in:
Ian Craggs 2018-06-07 14:45:01 +01:00
parent fcef9b5dd5
commit 7ea1333911
12 changed files with 967 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -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

1
src/CMakeLists.txt
View File

@ -48,6 +48,7 @@ SET(common_src
Heap.c
LinkedList.c
MQTTProperties.c
MQTTReasonCodes.c
Base64.c
SHA1.c
WebSocket.c

134
src/MQTTClient.c
View File

@ -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,

63
src/MQTTClient.h
View File

@ -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 <i>context</i> value originally passed to
* MQTTAsync_setCallbacks(), which contains any application-specific context.
* @param cause The reason for the disconnection.
* Currently, <i>cause</i> 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 <i>context</i> 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 <i>context</i> value originally passed to
* MQTTAsync_setCallbacks(), which contains any application-specific context.
* @param cause The reason for the disconnection.
* Currently, <i>cause</i> 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 <i>context</i> 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);

12
src/MQTTPacket.c
View File

@ -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);

2
src/MQTTProperties.c
View File

@ -25,7 +25,7 @@
#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
struct nameToType
static struct nameToType
{
enum PropertyNames name;
enum PropertyTypes type;

1
src/MQTTProtocolClient.c
View File

@ -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);

100
src/MQTTReasonCodes.c Normal file
View File

@ -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 <memory.h>
#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;
}

14
src/MQTTReasonCodes.h
View File

@ -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

20
test/CMakeLists.txt
View File

@ -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

611
test/test10.c Normal file
View File

@ -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 <string.h>
#include <stdlib.h>
#if !defined(_WINDOWS)
#include <sys/time.h>
#include <sys/socket.h>
#include <unistd.h>
#include <errno.h>
#else
#include <windows.h>
#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 <stdarg.h>
#include <time.h>
#include <sys/timeb.h>
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, "</testcase>\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, "<failure type=\"%s\">file %s, line %d </failure>\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, "<testcase classname=\"test_client_topic_aliases\" name=\"client topic aliases\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 1 - client topic aliases");
rc = MQTTClient_create(&c, options.connection, "client_topic_alias_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
MQTTClient_destroy(&c);
goto exit;
}
rc = MQTTClient_setCallbacks(c, NULL, NULL, messageArrived, NULL);
assert("Good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
rc = MQTTClient_setDisconnected(c, c, disconnected);
assert("Good rc from setDisconnected", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
opts.keepAliveInterval = 20;
opts.cleansession = 1;
opts.MQTTVersion = options.MQTTVersion;
if (options.haconnections != NULL)
{
opts.serverURIs = options.haconnections;
opts.serverURIcount = options.hacount;
}
MyLog(LOGA_DEBUG, "Connecting");
response = MQTTClient_connect5(c, &opts, NULL, NULL);
assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
if (response.reasonCode != MQTTCLIENT_SUCCESS)
goto exit;
if (response.properties)
{
logProperties(response.properties);
MQTTResponse_free(response);
}
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 = 1;
pubmsg.retained = 0;
/* a Topic Alias of 0 is not allowed, so we should be disconnected */
property.identifier = TOPIC_ALIAS;
property.value.integer2 = 0;
MQTTProperties_add(&pubmsg.properties, &property);
response = MQTTClient_publishMessage5(c, test_topic, &pubmsg, &dt);
assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
/* Now we expect to receive a disconnect packet telling us why */
count = 0;
while (test_topic_aliases_globals.disconnected == 0 && ++count < 10)
{
#if defined(WIN32)
Sleep(1000);
#else
usleep(1000000L);
#endif
}
assert("Disconnected should be called", test_topic_aliases_globals.disconnected == 1,
"was %d", test_topic_aliases_globals.disconnected);
property.identifier = SESSION_EXPIRY_INTERVAL;
property.value.integer4 = 30;
MQTTProperties_add(&connect_props, &property);
/* Now try a valid topic alias */
response = MQTTClient_connect5(c, &opts, &connect_props, NULL);
assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
if (response.reasonCode != MQTTCLIENT_SUCCESS)
goto exit;
if (response.properties)
{
if (MQTTProperties_hasProperty(response.properties, TOPIC_ALIAS_MAXIMUM))
topicAliasMaximum = MQTTProperties_getNumericValue(response.properties, TOPIC_ALIAS_MAXIMUM);
logProperties(response.properties);
MQTTResponse_free(response);
}
assert("topicAliasMaximum > 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, "<testsuite name=\"test1\" tests=\"%d\">\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, "</testsuite>\n");
fclose(xml);
return rc;
}

24
test/test15.c
View File

@ -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, "<testcase classname=\"test1\" name=\"connack return codes\"");
global_start_time = start_clock();
@ -756,8 +760,9 @@ int test3(struct Options options)
opts.will->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);