mirror of https://github.com/eclipse/paho.mqtt.c
MQTTClient connect synchronization #676
This commit is contained in:
Ian Craggs
parent
22746a4cf6
commit
bf5b79b706
|
|
@ -571,6 +571,7 @@ void MQTTClient_destroy(MQTTClient* handle)
|
|||
MQTTClients* m = *handle;
|
||||
|
||||
FUNC_ENTRY;
|
||||
Thread_lock_mutex(connect_mutex);
|
||||
Thread_lock_mutex(mqttclient_mutex);
|
||||
|
||||
if (m == NULL)
|
||||
|
|
@ -605,6 +606,7 @@ void MQTTClient_destroy(MQTTClient* handle)
|
|||
|
||||
exit:
|
||||
Thread_unlock_mutex(mqttclient_mutex);
|
||||
Thread_unlock_mutex(connect_mutex);
|
||||
FUNC_EXIT;
|
||||
}
|
||||
|
||||
|
|
@ -1636,6 +1638,12 @@ MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions*
|
|||
Thread_lock_mutex(mqttclient_mutex);
|
||||
|
||||
rc.reasonCode = SOCKET_ERROR;
|
||||
if (!library_initialized)
|
||||
{
|
||||
rc.reasonCode = MQTTCLIENT_FAILURE;
|
||||
goto exit;
|
||||
}
|
||||
|
||||
if (options == NULL)
|
||||
{
|
||||
rc.reasonCode = MQTTCLIENT_NULL_PARAMETER;
|
||||
|
|
|
|||
|
|
@ -1805,3 +1805,48 @@ IF (PAHO_BUILD_SHARED)
|
|||
PROPERTIES TIMEOUT 540
|
||||
)
|
||||
ENDIF()
|
||||
|
||||
IF (PAHO_BUILD_STATIC)
|
||||
ADD_EXECUTABLE(
|
||||
test_connect_destroy-static
|
||||
test_connect_destroy.c
|
||||
)
|
||||
|
||||
TARGET_LINK_LIBRARIES(
|
||||
test_connect_destroy-static
|
||||
paho-mqtt3c
|
||||
)
|
||||
|
||||
ADD_TEST(
|
||||
NAME test_connect_destroy-1-execute-destroy-during-connect-static
|
||||
COMMAND "test_connect_destroy" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER}
|
||||
)
|
||||
|
||||
SET_TESTS_PROPERTIES(
|
||||
test_connect_destroy-1-execute-destroy-during-connect-static
|
||||
PROPERTIES TIMEOUT 540
|
||||
)
|
||||
ENDIF()
|
||||
|
||||
|
||||
IF (PAHO_BUILD_SHARED)
|
||||
ADD_EXECUTABLE(
|
||||
test_connect_destroy
|
||||
test_connect_destroy.c
|
||||
)
|
||||
|
||||
TARGET_LINK_LIBRARIES(
|
||||
test_connect_destroy
|
||||
paho-mqtt3c
|
||||
)
|
||||
|
||||
ADD_TEST(
|
||||
NAME test_connect_destroy-1-execute-destroy-during-connect
|
||||
COMMAND "test_connect_destroy" "--test_no" "1" "--connection" ${MQTT_TEST_BROKER}
|
||||
)
|
||||
|
||||
SET_TESTS_PROPERTIES(
|
||||
test_connect_destroy-1-execute-destroy-during-connect
|
||||
PROPERTIES TIMEOUT 540
|
||||
)
|
||||
ENDIF()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,404 @@
|
|||
/*******************************************************************************
|
||||
* 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:
|
||||
* Michal Kasperek - initial API and implementation and/or initial documentation
|
||||
*******************************************************************************/
|
||||
|
||||
/**
|
||||
* @file
|
||||
* Test for issue 675: crash when invoking MQTTClient_destroy during MQTTClient_connect
|
||||
*/
|
||||
|
||||
#include "MQTTClient.h"
|
||||
#include "Thread.h"
|
||||
#include <string.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
#if !defined(_WINDOWS)
|
||||
#include <sys/time.h>
|
||||
#include <sys/socket.h>
|
||||
#include <unistd.h>
|
||||
#include <errno.h>
|
||||
#define WINAPI
|
||||
#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;
|
||||
int hacount;
|
||||
int verbose;
|
||||
int test_no;
|
||||
int MQTTVersion;
|
||||
int iterations;
|
||||
} options =
|
||||
{
|
||||
"localhost:1883",
|
||||
NULL,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
MQTTVERSION_DEFAULT,
|
||||
100,
|
||||
};
|
||||
|
||||
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], "--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;
|
||||
|
||||
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)
|
||||
|
||||
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 lock_mutex(mutex_type amutex)
|
||||
{
|
||||
int rc = Thread_lock_mutex(amutex);
|
||||
if (rc != 0)
|
||||
MyLog(LOGA_INFO, "Error %s locking mutex", strerror(rc));
|
||||
}
|
||||
|
||||
void unlock_mutex(mutex_type amutex)
|
||||
{
|
||||
int rc = Thread_unlock_mutex(amutex);
|
||||
if (rc != 0)
|
||||
MyLog(LOGA_INFO, "Error %s unlocking mutex", strerror(rc));
|
||||
}
|
||||
|
||||
|
||||
/*********************************************************************
|
||||
|
||||
Test1: destroy MQTT client while connectig
|
||||
|
||||
*********************************************************************/
|
||||
|
||||
struct thread_parms
|
||||
{
|
||||
MQTTClient* c;
|
||||
};
|
||||
|
||||
thread_return_type WINAPI test1_destroy(void* n)
|
||||
{
|
||||
MySleep(rand() % 500);
|
||||
|
||||
struct thread_parms *parms = n;
|
||||
MQTTClient* c = parms->c;
|
||||
|
||||
MQTTClient_destroy(c);
|
||||
}
|
||||
|
||||
|
||||
int test1(struct Options options)
|
||||
{
|
||||
srand(time(0));
|
||||
|
||||
MQTTClient c;
|
||||
MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
|
||||
MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer;
|
||||
int rc = 0;
|
||||
char* test_topic = "C client test1";
|
||||
|
||||
fprintf(xml, "<testcase classname=\"test1\" name=\"destroy mqtt while connecting\"");
|
||||
global_start_time = start_clock();
|
||||
failures = 0;
|
||||
MyLog(LOGA_INFO, "Starting test 1 - execute destroy while connecting");
|
||||
|
||||
rc = MQTTClient_create(&c, options.connection, "connect crash test",
|
||||
MQTTCLIENT_PERSISTENCE_NONE, NULL);
|
||||
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
|
||||
if (rc != MQTTCLIENT_SUCCESS)
|
||||
{
|
||||
MQTTClient_destroy(&c);
|
||||
goto exit;
|
||||
}
|
||||
|
||||
opts.connectTimeout = 30;
|
||||
opts.keepAliveInterval = 20;
|
||||
opts.cleansession = 1;
|
||||
opts.username = "testuser";
|
||||
opts.password = "testpassword";
|
||||
opts.MQTTVersion = options.MQTTVersion;
|
||||
if (options.haconnections != NULL)
|
||||
{
|
||||
opts.serverURIs = options.haconnections;
|
||||
opts.serverURIcount = options.hacount;
|
||||
}
|
||||
|
||||
opts.will = &wopts;
|
||||
opts.will->message = "will message";
|
||||
opts.will->qos = 1;
|
||||
opts.will->retained = 0;
|
||||
opts.will->topicName = "will topic";
|
||||
opts.will = NULL;
|
||||
|
||||
struct thread_parms parms = {&c};
|
||||
Thread_start(test1_destroy, (void*)&parms);
|
||||
|
||||
MQTTClient_connect(c, &opts);
|
||||
|
||||
MySleep(1000);
|
||||
|
||||
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, test1};
|
||||
int i;
|
||||
|
||||
xml = fopen("TEST-test2.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", 0);
|
||||
|
||||
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 */
|
||||
|
||||
options.test_no = 0;
|
||||
}
|
||||
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;
|
||||
}
|
||||
Loading…
Reference in New Issue