test_mqtt4async.c 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640
  1. /*******************************************************************************
  2. * Copyright (c) 2009, 2022 IBM Corp., Ian Craggs
  3. *
  4. * All rights reserved. This program and the accompanying materials
  5. * are made available under the terms of the Eclipse Public License v2.0
  6. * and Eclipse Distribution License v1.0 which accompany this distribution.
  7. *
  8. * The Eclipse Public License is available at
  9. * https://www.eclipse.org/legal/epl-2.0/
  10. * and the Eclipse Distribution License is available at
  11. * http://www.eclipse.org/org/documents/edl-v10.php.
  12. *
  13. * Contributors:
  14. * Ian Craggs - initial API and implementation and/or initial documentation
  15. * Ian Craggs - MQTT 3.1.1 support
  16. *******************************************************************************/
  17. /**
  18. * @file
  19. * MQTT 3.1.1 Tests for the asynchronous Paho MQTT C client
  20. */
  21. /*
  22. #if !defined(_RTSHEADER)
  23. #include <rts.h>
  24. #endif
  25. */
  26. #include "MQTTAsync.h"
  27. #include <string.h>
  28. #include <stdlib.h>
  29. #if !defined(_WINDOWS)
  30. #include <sys/time.h>
  31. #include <sys/socket.h>
  32. #include <unistd.h>
  33. #include <errno.h>
  34. #else
  35. #include <winsock2.h>
  36. #include <ws2tcpip.h>
  37. #define MAXHOSTNAMELEN 256
  38. #define EAGAIN WSAEWOULDBLOCK
  39. #define EINTR WSAEINTR
  40. #define EINPROGRESS WSAEINPROGRESS
  41. #define EWOULDBLOCK WSAEWOULDBLOCK
  42. #define ENOTCONN WSAENOTCONN
  43. #define ECONNRESET WSAECONNRESET
  44. #define setenv(a, b, c) _putenv_s(a, b)
  45. #endif
  46. #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
  47. void usage(void)
  48. {
  49. printf("help!!\n");
  50. exit(EXIT_FAILURE);
  51. }
  52. struct Options
  53. {
  54. char* connection; /**< connection to system under test. */
  55. char** haconnections;
  56. int hacount;
  57. int verbose;
  58. int test_no;
  59. int iterations;
  60. } options =
  61. {
  62. "localhost:1883",
  63. NULL,
  64. 0,
  65. 0,
  66. 0,
  67. 1,
  68. };
  69. void getopts(int argc, char** argv)
  70. {
  71. int count = 1;
  72. while (count < argc)
  73. {
  74. if (strcmp(argv[count], "--test_no") == 0)
  75. {
  76. if (++count < argc)
  77. options.test_no = atoi(argv[count]);
  78. else
  79. usage();
  80. }
  81. else if (strcmp(argv[count], "--connection") == 0)
  82. {
  83. if (++count < argc)
  84. {
  85. options.connection = argv[count];
  86. printf("\nSetting connection to %s\n", options.connection);
  87. }
  88. else
  89. usage();
  90. }
  91. else if (strcmp(argv[count], "--haconnections") == 0)
  92. {
  93. if (++count < argc)
  94. {
  95. char* tok = strtok(argv[count], " ");
  96. options.hacount = 0;
  97. options.haconnections = malloc(sizeof(char*) * 5);
  98. while (tok)
  99. {
  100. options.haconnections[options.hacount] = malloc(strlen(tok) + 1);
  101. strcpy(options.haconnections[options.hacount], tok);
  102. options.hacount++;
  103. tok = strtok(NULL, " ");
  104. }
  105. }
  106. else
  107. usage();
  108. }
  109. else if (strcmp(argv[count], "--iterations") == 0)
  110. {
  111. if (++count < argc)
  112. options.iterations = atoi(argv[count]);
  113. else
  114. usage();
  115. }
  116. else if (strcmp(argv[count], "--verbose") == 0)
  117. {
  118. options.verbose = 1;
  119. printf("\nSetting verbose on\n");
  120. }
  121. count++;
  122. }
  123. }
  124. #define LOGA_DEBUG 0
  125. #define LOGA_INFO 1
  126. #include <stdarg.h>
  127. #include <time.h>
  128. #include <sys/timeb.h>
  129. void MyLog(int LOGA_level, char* format, ...)
  130. {
  131. static char msg_buf[256];
  132. va_list args;
  133. #if defined(_WIN32) || defined(_WINDOWS)
  134. struct timeb ts;
  135. #else
  136. struct timeval ts;
  137. #endif
  138. struct tm timeinfo;
  139. if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
  140. return;
  141. #if defined(_WIN32) || defined(_WINDOWS)
  142. ftime(&ts);
  143. localtime_s(&timeinfo, &ts.time);
  144. #else
  145. gettimeofday(&ts, NULL);
  146. localtime_r(&ts.tv_sec, &timeinfo);
  147. #endif
  148. strftime(msg_buf, 80, "%Y%m%d %H%M%S", &timeinfo);
  149. #if defined(_WIN32) || defined(_WINDOWS)
  150. sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
  151. #else
  152. sprintf(&msg_buf[strlen(msg_buf)], ".%.3lu ", ts.tv_usec / 1000L);
  153. #endif
  154. va_start(args, format);
  155. vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
  156. va_end(args);
  157. printf("%s\n", msg_buf);
  158. fflush(stdout);
  159. }
  160. #if defined(_WIN32) || defined(_WINDOWS)
  161. #define mqsleep(A) Sleep(1000*A)
  162. #define START_TIME_TYPE DWORD
  163. static DWORD start_time = 0;
  164. START_TIME_TYPE start_clock(void)
  165. {
  166. return GetTickCount();
  167. }
  168. #elif defined(AIX)
  169. #define mqsleep sleep
  170. #define START_TIME_TYPE struct timespec
  171. START_TIME_TYPE start_clock(void)
  172. {
  173. static struct timespec start;
  174. clock_gettime(CLOCK_REALTIME, &start);
  175. return start;
  176. }
  177. #else
  178. #define mqsleep sleep
  179. #define START_TIME_TYPE struct timeval
  180. /* TODO - unused - remove? static struct timeval start_time; */
  181. START_TIME_TYPE start_clock(void)
  182. {
  183. struct timeval start_time;
  184. gettimeofday(&start_time, NULL);
  185. return start_time;
  186. }
  187. #endif
  188. #if defined(_WIN32)
  189. long elapsed(START_TIME_TYPE start_time)
  190. {
  191. return GetTickCount() - start_time;
  192. }
  193. #elif defined(AIX)
  194. #define assert(a)
  195. long elapsed(struct timespec start)
  196. {
  197. struct timespec now, res;
  198. clock_gettime(CLOCK_REALTIME, &now);
  199. ntimersub(now, start, res);
  200. return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
  201. }
  202. #else
  203. long elapsed(START_TIME_TYPE start_time)
  204. {
  205. struct timeval now, res;
  206. gettimeofday(&now, NULL);
  207. timersub(&now, &start_time, &res);
  208. return (res.tv_sec)*1000 + (res.tv_usec)/1000;
  209. }
  210. #endif
  211. #define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
  212. #define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
  213. int tests = 0;
  214. int failures = 0;
  215. FILE* xml;
  216. START_TIME_TYPE global_start_time;
  217. char output[3000];
  218. char* cur_output = output;
  219. int test_finished = 0;
  220. void write_test_result(void)
  221. {
  222. long duration = elapsed(global_start_time);
  223. fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000);
  224. if (cur_output != output)
  225. {
  226. fprintf(xml, "%s", output);
  227. cur_output = output;
  228. }
  229. fprintf(xml, "</testcase>\n");
  230. }
  231. void myassert(char* filename, int lineno, char* description, int value, char* format, ...)
  232. {
  233. ++tests;
  234. if (!value)
  235. {
  236. va_list args;
  237. ++failures;
  238. MyLog(LOGA_INFO, "Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
  239. va_start(args, format);
  240. vprintf(format, args);
  241. va_end(args);
  242. cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
  243. description, filename, lineno);
  244. }
  245. else
  246. MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
  247. }
  248. void test1_onDisconnect3(void* context, MQTTAsync_successData* response)
  249. {
  250. MQTTAsync c = (MQTTAsync)context;
  251. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  252. test_finished = 1;
  253. }
  254. void test1_onConnect3(void* context, MQTTAsync_successData* response)
  255. {
  256. MQTTAsync c = (MQTTAsync)context;
  257. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  258. int rc;
  259. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  260. opts.onSuccess = test1_onDisconnect3;
  261. opts.context = c;
  262. assert("Correct serverURI returned", strstr(response->alt.connect.serverURI, options.connection) != NULL,
  263. "serverURI was %s", response->alt.connect.serverURI);
  264. assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4,
  265. "MQTTVersion was %d", response->alt.connect.MQTTVersion);
  266. assert("Correct sessionPresent returned", response->alt.connect.sessionPresent == 1,
  267. "sessionPresent was %d", response->alt.connect.sessionPresent);
  268. rc = MQTTAsync_disconnect(c, &opts);
  269. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  270. }
  271. void test1_onDisconnect2(void* context, MQTTAsync_successData* response)
  272. {
  273. MQTTAsync c = (MQTTAsync)context;
  274. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
  275. int rc;
  276. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  277. opts.MQTTVersion = 4;
  278. opts.cleansession = 0;
  279. if (options.haconnections != NULL)
  280. {
  281. opts.serverURIs = options.haconnections;
  282. opts.serverURIcount = options.hacount;
  283. }
  284. opts.onSuccess = test1_onConnect3;
  285. opts.onFailure = NULL;
  286. opts.context = c;
  287. opts.cleansession = 0;
  288. rc = MQTTAsync_connect(c, &opts);
  289. assert("Connect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  290. }
  291. void test1_onConnect2(void* context, MQTTAsync_successData* response)
  292. {
  293. MQTTAsync c = (MQTTAsync)context;
  294. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  295. int rc;
  296. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  297. opts.onSuccess = test1_onDisconnect2;
  298. opts.context = c;
  299. assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0,
  300. "serverURI was %s", response->alt.connect.serverURI);
  301. assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4,
  302. "MQTTVersion was %d", response->alt.connect.MQTTVersion);
  303. assert("Correct sessionPresent returned", response->alt.connect.sessionPresent == 0,
  304. "sessionPresent was %d", response->alt.connect.sessionPresent);
  305. rc = MQTTAsync_disconnect(c, &opts);
  306. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  307. }
  308. void test1_onDisconnect1(void* context, MQTTAsync_successData* response)
  309. {
  310. MQTTAsync c = (MQTTAsync)context;
  311. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
  312. int rc;
  313. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  314. opts.MQTTVersion = 4;
  315. opts.cleansession = 0;
  316. if (options.haconnections != NULL)
  317. {
  318. opts.serverURIs = options.haconnections;
  319. opts.serverURIcount = options.hacount;
  320. }
  321. opts.onSuccess = test1_onConnect2;
  322. opts.onFailure = NULL;
  323. opts.context = c;
  324. opts.cleansession = 0;
  325. rc = MQTTAsync_connect(c, &opts);
  326. assert("Connect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  327. }
  328. void test1_onConnect1(void* context, MQTTAsync_successData* response)
  329. {
  330. MQTTAsync c = (MQTTAsync)context;
  331. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  332. int rc;
  333. MyLog(LOGA_DEBUG, "In connect onSuccess callback 1, context %p", context);
  334. opts.onSuccess = test1_onDisconnect1;
  335. opts.context = c;
  336. assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0,
  337. "serverURI was %s", response->alt.connect.serverURI);
  338. assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4,
  339. "MQTTVersion was %d", response->alt.connect.MQTTVersion);
  340. assert("Correct sessionPresent returned", response->alt.connect.sessionPresent == 0,
  341. "sessionPresent was %d", response->alt.connect.sessionPresent);
  342. rc = MQTTAsync_disconnect(c, &opts);
  343. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  344. }
  345. /*********************************************************************
  346. Test1: sessionPresent
  347. *********************************************************************/
  348. int test1(struct Options options)
  349. {
  350. MQTTAsync c;
  351. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
  352. int rc = 0;
  353. char* test_topic = "C client test1";
  354. fprintf(xml, "<testcase classname=\"test1\" name=\"sessionPresent\"");
  355. global_start_time = start_clock();
  356. test_finished = failures = 0;
  357. MyLog(LOGA_INFO, "Starting test 1 - sessionPresent");
  358. rc = MQTTAsync_create(&c, options.connection, "sesssionPresent",
  359. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
  360. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  361. if (rc != MQTTASYNC_SUCCESS)
  362. {
  363. MQTTAsync_destroy(&c);
  364. goto exit;
  365. }
  366. opts.MQTTVersion = 4;
  367. if (options.haconnections != NULL)
  368. {
  369. opts.serverURIs = options.haconnections;
  370. opts.serverURIcount = options.hacount;
  371. }
  372. opts.onSuccess = test1_onConnect1;
  373. opts.onFailure = NULL;
  374. opts.context = c;
  375. /* Connect cleansession */
  376. opts.cleansession = 1;
  377. MyLog(LOGA_DEBUG, "Connecting");
  378. rc = MQTTAsync_connect(c, &opts);
  379. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  380. if (rc != MQTTASYNC_SUCCESS)
  381. goto exit;
  382. while (!test_finished)
  383. #if defined(_WIN32)
  384. Sleep(100);
  385. #else
  386. usleep(10000L);
  387. #endif
  388. MQTTAsync_destroy(&c);
  389. exit:
  390. MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.",
  391. (failures == 0) ? "passed" : "failed", tests, failures);
  392. write_test_result();
  393. return failures;
  394. }
  395. void test2_onDisconnect(void* context, MQTTAsync_successData* response)
  396. {
  397. MQTTAsync c = (MQTTAsync)context;
  398. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  399. test_finished = 1;
  400. }
  401. void test2_onSubscribe2(void* context, MQTTAsync_failureData* response)
  402. {
  403. MQTTAsync c = (MQTTAsync)context;
  404. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  405. int rc;
  406. MyLog(LOGA_DEBUG, "In subscribe onFailure callback, context %p", context);
  407. assert("Correct subscribe return code", response->code == MQTT_BAD_SUBSCRIBE,
  408. "qos was %d", response->code);
  409. opts.onSuccess = test2_onDisconnect;
  410. rc = MQTTAsync_disconnect(c, &opts);
  411. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  412. }
  413. void test2_onSubscribe1(void* context, MQTTAsync_successData* response)
  414. {
  415. MQTTAsync c = (MQTTAsync)context;
  416. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  417. int rc;
  418. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback, context %p", context);
  419. assert("Correct subscribe return code", response->alt.qos == 2,
  420. "qos was %d", response->alt.qos);
  421. }
  422. void test2_onConnect1(void* context, MQTTAsync_successData* response)
  423. {
  424. MQTTAsync c = (MQTTAsync)context;
  425. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  426. int rc;
  427. MyLog(LOGA_DEBUG, "In connect onSuccess callback 1, context %p", context);
  428. assert("Correct serverURI returned", strcmp(response->alt.connect.serverURI, options.connection) == 0,
  429. "serverURI was %s", response->alt.connect.serverURI);
  430. assert("Correct MQTTVersion returned", response->alt.connect.MQTTVersion == 4,
  431. "MQTTVersion was %d", response->alt.connect.MQTTVersion);
  432. assert("Correct sessionPresent returned", response->alt.connect.sessionPresent == 0,
  433. "sessionPresent was %d", response->alt.connect.sessionPresent);
  434. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  435. opts.context = c;
  436. opts.onSuccess = test2_onSubscribe1;
  437. rc = MQTTAsync_subscribe(c, "a topic I can subscribe to", 2, &opts);
  438. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  439. if (rc != MQTTASYNC_SUCCESS)
  440. test_finished = 1;
  441. opts.onSuccess = NULL;
  442. opts.onFailure = test2_onSubscribe2;
  443. rc = MQTTAsync_subscribe(c, "nosubscribe", 2, &opts);
  444. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  445. if (rc != MQTTASYNC_SUCCESS)
  446. test_finished = 1;
  447. }
  448. /*********************************************************************
  449. Test1: 0x80 from subscribe
  450. *********************************************************************/
  451. int test2(struct Options options)
  452. {
  453. MQTTAsync c;
  454. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
  455. int rc = 0;
  456. char* test_topic = "C client test1";
  457. fprintf(xml, "<testcase classname=\"test2\" name=\"bad subscribe\"");
  458. global_start_time = start_clock();
  459. test_finished = failures = 0;
  460. MyLog(LOGA_INFO, "Starting test 2 - bad subscribe");
  461. rc = MQTTAsync_create(&c, options.connection, "badSubscribe test",
  462. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
  463. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  464. if (rc != MQTTASYNC_SUCCESS)
  465. {
  466. MQTTAsync_destroy(&c);
  467. goto exit;
  468. }
  469. opts.MQTTVersion = 4;
  470. if (options.haconnections != NULL)
  471. {
  472. opts.serverURIs = options.haconnections;
  473. opts.serverURIcount = options.hacount;
  474. }
  475. opts.onSuccess = test2_onConnect1;
  476. opts.onFailure = NULL;
  477. opts.context = c;
  478. /* Connect cleansession */
  479. opts.cleansession = 1;
  480. MyLog(LOGA_DEBUG, "Connecting");
  481. rc = MQTTAsync_connect(c, &opts);
  482. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  483. if (rc != MQTTASYNC_SUCCESS)
  484. goto exit;
  485. while (!test_finished)
  486. #if defined(_WIN32)
  487. Sleep(100);
  488. #else
  489. usleep(10000L);
  490. #endif
  491. MQTTAsync_destroy(&c);
  492. exit:
  493. MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.",
  494. (failures == 0) ? "passed" : "failed", tests, failures);
  495. write_test_result();
  496. return failures;
  497. }
  498. int main(int argc, char** argv)
  499. {
  500. int rc = 0;
  501. int (*tests[])() = {NULL, test1, test2};
  502. int i;
  503. xml = fopen("TEST-MQTT4sync.xml", "w");
  504. fprintf(xml, "<testsuite name=\"test-mqtt4sync\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1));
  505. setenv("MQTT_C_CLIENT_TRACE", "ON", 1);
  506. setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1);
  507. getopts(argc, argv);
  508. for (i = 0; i < options.iterations; ++i)
  509. {
  510. if (options.test_no == 0)
  511. { /* run all the tests */
  512. for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
  513. rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */
  514. }
  515. else
  516. rc = tests[options.test_no](options); /* run just the selected test */
  517. }
  518. if (rc == 0)
  519. MyLog(LOGA_INFO, "verdict pass");
  520. else
  521. MyLog(LOGA_INFO, "verdict fail");
  522. fprintf(xml, "</testsuite>\n");
  523. fclose(xml);
  524. return rc;
  525. }