test45.c 60 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098
  1. /*******************************************************************************
  2. * Copyright (c) 2009, 2020 IBM Corp.
  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. * Ian Craggs - test8 - failure callbacks
  17. * Ian Craggs - MQTT V5
  18. *******************************************************************************/
  19. /**
  20. * @file
  21. * Tests for the Paho Asynchronous MQTT C client
  22. */
  23. #include "MQTTAsync.h"
  24. #include <string.h>
  25. #include <stdlib.h>
  26. #if !defined(_WINDOWS)
  27. #include <sys/time.h>
  28. #include <sys/socket.h>
  29. #include <unistd.h>
  30. #include <errno.h>
  31. #else
  32. #include <windows.h>
  33. #endif
  34. #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
  35. void usage(void)
  36. {
  37. printf("help!!\n");
  38. exit(EXIT_FAILURE);
  39. }
  40. struct Options
  41. {
  42. char* connection; /**< connection to system under test. */
  43. int verbose;
  44. int test_no;
  45. int size; /**< size of big message */
  46. int MQTTVersion;
  47. int iterations;
  48. } options =
  49. {
  50. "localhost:1883",
  51. 0,
  52. -1,
  53. 10000,
  54. MQTTVERSION_5,
  55. 1,
  56. };
  57. void getopts(int argc, char** argv)
  58. {
  59. int count = 1;
  60. while (count < argc)
  61. {
  62. if (strcmp(argv[count], "--test_no") == 0)
  63. {
  64. if (++count < argc)
  65. options.test_no = atoi(argv[count]);
  66. else
  67. usage();
  68. }
  69. else if (strcmp(argv[count], "--size") == 0)
  70. {
  71. if (++count < argc)
  72. options.size = atoi(argv[count]);
  73. else
  74. usage();
  75. }
  76. else if (strcmp(argv[count], "--connection") == 0)
  77. {
  78. if (++count < argc)
  79. options.connection = argv[count];
  80. else
  81. usage();
  82. }
  83. else if (strcmp(argv[count], "--MQTTversion") == 0)
  84. {
  85. if (++count < argc)
  86. {
  87. options.MQTTVersion = atoi(argv[count]);
  88. printf("setting MQTT version to %d\n", options.MQTTVersion);
  89. }
  90. else
  91. usage();
  92. }
  93. else if (strcmp(argv[count], "--iterations") == 0)
  94. {
  95. if (++count < argc)
  96. options.iterations = atoi(argv[count]);
  97. else
  98. usage();
  99. }
  100. else if (strcmp(argv[count], "--verbose") == 0)
  101. options.verbose = 1;
  102. count++;
  103. }
  104. }
  105. #define LOGA_DEBUG 0
  106. #define LOGA_INFO 1
  107. #include <stdarg.h>
  108. #include <time.h>
  109. #include <sys/timeb.h>
  110. void MyLog(int LOGA_level, char* format, ...)
  111. {
  112. static char msg_buf[256];
  113. va_list args;
  114. #if defined(_WIN32) || defined(_WINDOWS)
  115. struct timeb ts;
  116. #else
  117. struct timeval ts;
  118. #endif
  119. struct tm timeinfo;
  120. if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
  121. return;
  122. #if defined(_WIN32) || defined(_WINDOWS)
  123. ftime(&ts);
  124. localtime_s(&timeinfo, &ts.time);
  125. #else
  126. gettimeofday(&ts, NULL);
  127. localtime_r(&ts.tv_sec, &timeinfo);
  128. #endif
  129. strftime(msg_buf, 80, "%Y%m%d %H%M%S", &timeinfo);
  130. #if defined(_WIN32) || defined(_WINDOWS)
  131. sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
  132. #else
  133. sprintf(&msg_buf[strlen(msg_buf)], ".%.3lu ", ts.tv_usec / 1000L);
  134. #endif
  135. va_start(args, format);
  136. vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
  137. va_end(args);
  138. printf("%s\n", msg_buf);
  139. fflush(stdout);
  140. }
  141. void MySleep(long milliseconds)
  142. {
  143. #if defined(_WIN32) || defined(_WIN64)
  144. Sleep(milliseconds);
  145. #else
  146. usleep(milliseconds*1000);
  147. #endif
  148. }
  149. #if defined(_WIN32) || defined(_WINDOWS)
  150. #define START_TIME_TYPE DWORD
  151. static DWORD start_time = 0;
  152. START_TIME_TYPE start_clock(void)
  153. {
  154. return GetTickCount();
  155. }
  156. #elif defined(AIX)
  157. #define START_TIME_TYPE struct timespec
  158. START_TIME_TYPE start_clock(void)
  159. {
  160. static struct timespec start;
  161. clock_gettime(CLOCK_REALTIME, &start);
  162. return start;
  163. }
  164. #else
  165. #define START_TIME_TYPE struct timeval
  166. /* TODO - unused - remove? static struct timeval start_time; */
  167. START_TIME_TYPE start_clock(void)
  168. {
  169. struct timeval start_time;
  170. gettimeofday(&start_time, NULL);
  171. return start_time;
  172. }
  173. #endif
  174. #if defined(_WIN32)
  175. long elapsed(START_TIME_TYPE start_time)
  176. {
  177. return GetTickCount() - start_time;
  178. }
  179. #elif defined(AIX)
  180. #define assert(a)
  181. long elapsed(struct timespec start)
  182. {
  183. struct timespec now, res;
  184. clock_gettime(CLOCK_REALTIME, &now);
  185. ntimersub(now, start, res);
  186. return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
  187. }
  188. #else
  189. long elapsed(START_TIME_TYPE start_time)
  190. {
  191. struct timeval now, res;
  192. gettimeofday(&now, NULL);
  193. timersub(&now, &start_time, &res);
  194. return (res.tv_sec)*1000 + (res.tv_usec)/1000;
  195. }
  196. #endif
  197. #define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
  198. #define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
  199. int tests = 0;
  200. int failures = 0;
  201. FILE* xml;
  202. START_TIME_TYPE global_start_time;
  203. char output[3000];
  204. char* cur_output = output;
  205. void write_test_result(void)
  206. {
  207. long duration = elapsed(global_start_time);
  208. fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000);
  209. if (cur_output != output)
  210. {
  211. fprintf(xml, "%s", output);
  212. cur_output = output;
  213. }
  214. fprintf(xml, "</testcase>\n");
  215. }
  216. void myassert(char* filename, int lineno, char* description, int value, char* format, ...)
  217. {
  218. ++tests;
  219. if (!value)
  220. {
  221. va_list args;
  222. ++failures;
  223. printf("Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
  224. va_start(args, format);
  225. vprintf(format, args);
  226. va_end(args);
  227. cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
  228. description, filename, lineno);
  229. }
  230. else
  231. MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
  232. }
  233. void logProperties(MQTTProperties *props)
  234. {
  235. int i = 0;
  236. for (i = 0; i < props->count; ++i)
  237. {
  238. int id = props->array[i].identifier;
  239. const char* name = MQTTPropertyName(id);
  240. char* intformat = "Property name %s value %d";
  241. switch (MQTTProperty_getType(id))
  242. {
  243. case MQTTPROPERTY_TYPE_BYTE:
  244. MyLog(LOGA_INFO, intformat, name, props->array[i].value.byte);
  245. break;
  246. case MQTTPROPERTY_TYPE_TWO_BYTE_INTEGER:
  247. MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer2);
  248. break;
  249. case MQTTPROPERTY_TYPE_FOUR_BYTE_INTEGER:
  250. MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
  251. break;
  252. case MQTTPROPERTY_TYPE_VARIABLE_BYTE_INTEGER:
  253. MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
  254. break;
  255. case MQTTPROPERTY_TYPE_BINARY_DATA:
  256. case MQTTPROPERTY_TYPE_UTF_8_ENCODED_STRING:
  257. MyLog(LOGA_INFO, "Property name %s value len %.*s", name,
  258. props->array[i].value.data.len, props->array[i].value.data.data);
  259. break;
  260. case MQTTPROPERTY_TYPE_UTF_8_STRING_PAIR:
  261. MyLog(LOGA_INFO, "Property name %s key %.*s value %.*s", name,
  262. props->array[i].value.data.len, props->array[i].value.data.data,
  263. props->array[i].value.value.len, props->array[i].value.value.data);
  264. break;
  265. }
  266. }
  267. }
  268. void waitForNoPendingTokens(MQTTAsync c)
  269. {
  270. int i = 0, rc = 0, count = 0;
  271. MQTTAsync_token *tokens;
  272. /* acks for outgoing messages could arrive after incoming exchanges are complete */
  273. do
  274. {
  275. rc = MQTTAsync_getPendingTokens(c, &tokens);
  276. assert("Good rc from getPendingTokens", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc);
  277. i = 0;
  278. if (tokens)
  279. {
  280. while (tokens[i] != -1)
  281. ++i;
  282. MQTTAsync_free(tokens);
  283. }
  284. if (i > 0)
  285. MySleep(100);
  286. }
  287. while (i > 0 && ++count < 100);
  288. assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i);
  289. }
  290. volatile int test_finished = 0;
  291. char* test_topic = "async test topic";
  292. void test1_onDisconnect(void* context, MQTTAsync_successData* response)
  293. {
  294. MQTTAsync c = (MQTTAsync)context;
  295. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  296. test_finished = 1;
  297. }
  298. void test1_onUnsubscribe(void* context, MQTTAsync_successData5* response)
  299. {
  300. MQTTAsync c = (MQTTAsync)context;
  301. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  302. MQTTProperty property;
  303. int rc;
  304. MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess callback %p", c);
  305. MyLog(LOGA_INFO, "Unsuback properties:");
  306. logProperties(&response->properties);
  307. assert("A property should exist", response->properties.count > 0,
  308. "Property count was %d\n", response->properties);
  309. opts.onSuccess = test1_onDisconnect;
  310. opts.context = c;
  311. opts.reasonCode = MQTTREASONCODE_UNSPECIFIED_ERROR;
  312. property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
  313. property.value.integer4 = 0;
  314. MQTTProperties_add(&opts.properties, &property);
  315. rc = MQTTAsync_disconnect(c, &opts);
  316. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  317. MQTTProperties_free(&opts.properties);
  318. }
  319. int test1_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
  320. {
  321. MQTTAsync c = (MQTTAsync)context;
  322. static int message_count = 0;
  323. int rc;
  324. MyLog(LOGA_DEBUG, "In messageArrived callback %p", c);
  325. assert("Message structure version should be 1", message->struct_version == 1,
  326. "message->struct_version was %d", message->struct_version);
  327. if (message->struct_version == 1)
  328. {
  329. assert("Properties count should be 2", message->properties.count == 2,
  330. "Properties count was %d\n", message->properties.count);
  331. logProperties(&message->properties);
  332. }
  333. if (++message_count == 1)
  334. {
  335. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  336. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  337. MQTTProperty property;
  338. MQTTProperties props = MQTTProperties_initializer;
  339. property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
  340. property.value.data.data = "test user property";
  341. property.value.data.len = (int)strlen(property.value.data.data);
  342. property.value.value.data = "test user property value";
  343. property.value.value.len = (int)strlen(property.value.value.data);
  344. MQTTProperties_add(&props, &property);
  345. pubmsg.properties = props;
  346. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  347. pubmsg.payloadlen = 11;
  348. pubmsg.qos = 2;
  349. pubmsg.retained = 0;
  350. rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
  351. assert("Publish should return 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  352. MQTTProperties_free(&props);
  353. }
  354. else
  355. {
  356. MQTTProperty property;
  357. MQTTProperties props = MQTTProperties_initializer;
  358. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  359. property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
  360. property.value.data.data = "test user property";
  361. property.value.data.len = (int)strlen(property.value.data.data);
  362. property.value.value.data = "test user property value";
  363. property.value.value.len = (int)strlen(property.value.value.data);
  364. MQTTProperties_add(&props, &property);
  365. opts.properties = props;
  366. opts.onSuccess5 = test1_onUnsubscribe;
  367. opts.context = c;
  368. rc = MQTTAsync_unsubscribe(c, test_topic, &opts);
  369. assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  370. MQTTProperties_free(&props);
  371. }
  372. MQTTAsync_freeMessage(&message);
  373. MQTTAsync_free(topicName);
  374. return 1;
  375. }
  376. void test1_onSubscribe(void* context, MQTTAsync_successData5* response)
  377. {
  378. MQTTAsync c = (MQTTAsync)context;
  379. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  380. int rc;
  381. MQTTProperty property;
  382. MQTTProperties props = MQTTProperties_initializer;
  383. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  384. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c, response->reasonCode);
  385. assert("Subscribe response should be 2", response->reasonCode == MQTTREASONCODE_GRANTED_QOS_2,
  386. "response was %d", response->reasonCode);
  387. MyLog(LOGA_INFO, "Suback properties:");
  388. logProperties(&response->properties);
  389. property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
  390. property.value.data.data = "test user property";
  391. property.value.data.len = (int)strlen(property.value.data.data);
  392. property.value.value.data = "test user property value";
  393. property.value.value.len = (int)strlen(property.value.value.data);
  394. MQTTProperties_add(&props, &property);
  395. opts.properties = props;
  396. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  397. pubmsg.payloadlen = 11;
  398. pubmsg.qos = 2;
  399. pubmsg.retained = 0;
  400. rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &opts);
  401. MQTTProperties_free(&props);
  402. }
  403. void test1_onConnect(void* context, MQTTAsync_successData5* response)
  404. {
  405. MQTTAsync c = (MQTTAsync)context;
  406. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  407. MQTTProperty property;
  408. MQTTProperties props = MQTTProperties_initializer;
  409. int rc;
  410. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  411. assert("Reason code should be 0", response->reasonCode == MQTTREASONCODE_SUCCESS,
  412. "Reason code was %d\n", response->reasonCode);
  413. MyLog(LOGA_INFO, "Connack properties:");
  414. logProperties(&response->properties);
  415. opts.onSuccess5 = test1_onSubscribe;
  416. opts.context = c;
  417. property.identifier = MQTTPROPERTY_CODE_SUBSCRIPTION_IDENTIFIER;
  418. property.value.integer4 = 33;
  419. MQTTProperties_add(&props, &property);
  420. opts.properties = props;
  421. opts.subscribeOptions.retainAsPublished = 1;
  422. rc = MQTTAsync_subscribe(c, test_topic, 2, &opts);
  423. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  424. if (rc != MQTTASYNC_SUCCESS)
  425. test_finished = 1;
  426. MQTTProperties_free(&props);
  427. }
  428. /*********************************************************************
  429. Test1: Basic connect, subscribe send and receive.
  430. *********************************************************************/
  431. int test1(struct Options options)
  432. {
  433. int subsqos = 2;
  434. MQTTAsync c;
  435. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  436. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  437. MQTTProperties props = MQTTProperties_initializer;
  438. MQTTProperties willProps = MQTTProperties_initializer;
  439. MQTTProperty property;
  440. int rc = 0;
  441. char* test_topic = "V5 C client test1";
  442. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  443. MyLog(LOGA_INFO, "Starting V5 test 1 - asynchronous connect");
  444. fprintf(xml, "<testcase classname=\"test45\" name=\"asynchronous connect\"");
  445. global_start_time = start_clock();
  446. createOpts.MQTTVersion = MQTTVERSION_5;
  447. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test",
  448. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  449. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  450. if (rc != MQTTASYNC_SUCCESS)
  451. {
  452. MQTTAsync_destroy(&c);
  453. goto exit;
  454. }
  455. rc = MQTTAsync_setCallbacks(c, c, NULL, test1_messageArrived, NULL);
  456. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  457. opts.keepAliveInterval = 20;
  458. opts.username = "testuser";
  459. opts.password = "testpassword";
  460. opts.MQTTVersion = options.MQTTVersion;
  461. opts.will = &wopts;
  462. opts.will->message = "will message";
  463. opts.will->qos = 1;
  464. opts.will->retained = 0;
  465. opts.will->topicName = "will topic";
  466. opts.will = NULL;
  467. opts.onSuccess5 = test1_onConnect;
  468. opts.onFailure5 = NULL;
  469. opts.context = c;
  470. property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
  471. property.value.integer4 = 30;
  472. MQTTProperties_add(&props, &property);
  473. property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
  474. property.value.data.data = "test user property";
  475. property.value.data.len = (int)strlen(property.value.data.data);
  476. property.value.value.data = "test user property value";
  477. property.value.value.len = (int)strlen(property.value.value.data);
  478. MQTTProperties_add(&props, &property);
  479. opts.connectProperties = &props;
  480. opts.willProperties = &willProps;
  481. MyLog(LOGA_DEBUG, "Connecting");
  482. rc = MQTTAsync_connect(c, &opts);
  483. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  484. if (rc != MQTTASYNC_SUCCESS)
  485. goto exit;
  486. while (!test_finished)
  487. MySleep(100);
  488. MQTTProperties_free(&props);
  489. MQTTProperties_free(&willProps);
  490. MQTTAsync_destroy(&c);
  491. exit:
  492. MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.",
  493. (failures == 0) ? "passed" : "failed", tests, failures);
  494. write_test_result();
  495. return failures;
  496. }
  497. int test2_onFailure_called = 0;
  498. void test2_onFailure(void* context, MQTTAsync_failureData5* response)
  499. {
  500. MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
  501. test2_onFailure_called++;
  502. test_finished = 1;
  503. }
  504. void test2_onConnect(void* context, MQTTAsync_successData5* response)
  505. {
  506. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p\n", context);
  507. assert("Connect should not succeed", 0, "connect success callback was called", 0);
  508. test_finished = 1;
  509. }
  510. /*********************************************************************
  511. Test2: connect timeout
  512. *********************************************************************/
  513. int test2(struct Options options)
  514. {
  515. int subsqos = 2;
  516. MQTTAsync c;
  517. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  518. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  519. int rc = 0;
  520. char* test_topic = "C client test2";
  521. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  522. test_finished = 0;
  523. MyLog(LOGA_INFO, "Starting test 2 - connect timeout");
  524. fprintf(xml, "<testcase classname=\"test4\" name=\"connect timeout\"");
  525. global_start_time = start_clock();
  526. createOpts.MQTTVersion = MQTTVERSION_5;
  527. rc = MQTTAsync_createWithOptions(&c, "tcp://9.20.96.160:66", "connect timeout",
  528. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  529. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  530. if (rc != MQTTASYNC_SUCCESS)
  531. {
  532. MQTTAsync_destroy(&c);
  533. goto exit;
  534. }
  535. rc = MQTTAsync_setCallbacks(c, c, NULL, test1_messageArrived, NULL);
  536. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  537. opts.connectTimeout = 5;
  538. opts.keepAliveInterval = 20;
  539. opts.username = "testuser";
  540. opts.binarypwd.data = "testpassword";
  541. opts.binarypwd.len = (int)strlen(opts.binarypwd.data);
  542. opts.MQTTVersion = options.MQTTVersion;
  543. opts.will = &wopts;
  544. opts.will->message = "will message";
  545. opts.will->qos = 1;
  546. opts.will->retained = 0;
  547. opts.will->topicName = "will topic";
  548. opts.will = NULL;
  549. opts.onSuccess5 = test2_onConnect;
  550. opts.onFailure5 = test2_onFailure;
  551. opts.context = c;
  552. MyLog(LOGA_DEBUG, "Connecting");
  553. rc = MQTTAsync_connect(c, &opts);
  554. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  555. if (rc != MQTTASYNC_SUCCESS)
  556. goto exit;
  557. while (!test_finished)
  558. MySleep(100);
  559. MQTTAsync_destroy(&c);
  560. exit:
  561. assert("Connect onFailure should be called once", test2_onFailure_called == 1,
  562. "connect onFailure was called %d times", test2_onFailure_called);
  563. MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.",
  564. (failures == 0) ? "passed" : "failed", tests, failures);
  565. write_test_result();
  566. return failures;
  567. }
  568. typedef struct
  569. {
  570. MQTTAsync c;
  571. int index;
  572. char clientid[24];
  573. char test_topic[100];
  574. int message_count;
  575. } client_data;
  576. void test3_onDisconnect(void* context, MQTTAsync_successData5* response)
  577. {
  578. client_data* cd = (client_data*)context;
  579. MyLog(LOGA_DEBUG, "In onDisconnect callback for client \"%s\"", cd->clientid);
  580. test_finished++;
  581. }
  582. void test3_onPublish(void* context, MQTTAsync_successData5* response)
  583. {
  584. client_data* cd = (client_data*)context;
  585. MyLog(LOGA_DEBUG, "In QoS 0 onPublish callback for client \"%s\"", cd->clientid);
  586. }
  587. void test3_onUnsubscribe(void* context, MQTTAsync_successData5* response)
  588. {
  589. client_data* cd = (client_data*)context;
  590. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  591. int rc;
  592. MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess callback \"%s\"", cd->clientid);
  593. opts.onSuccess5 = test3_onDisconnect;
  594. opts.context = cd;
  595. rc = MQTTAsync_disconnect(cd->c, &opts);
  596. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  597. }
  598. int test3_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
  599. {
  600. client_data* cd = (client_data*)context;
  601. int rc;
  602. MyLog(LOGA_DEBUG, "In messageArrived callback \"%s\" message count ", cd->clientid);
  603. if (++cd->message_count == 1)
  604. {
  605. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  606. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  607. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  608. pubmsg.payloadlen = 25;
  609. pubmsg.qos = 1;
  610. pubmsg.retained = 0;
  611. rc = MQTTAsync_sendMessage(cd->c, cd->test_topic, &pubmsg, &opts);
  612. assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  613. }
  614. else if (cd->message_count == 2)
  615. {
  616. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  617. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  618. pubmsg.payload = "a QoS 0 message that we can shorten to the extent that we need to payload up to 11";
  619. pubmsg.payloadlen = 29;
  620. pubmsg.qos = 0;
  621. pubmsg.retained = 0;
  622. opts.context = cd;
  623. opts.onSuccess5 = test3_onPublish;
  624. rc = MQTTAsync_sendMessage(cd->c, cd->test_topic, &pubmsg, &opts);
  625. assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  626. }
  627. else
  628. {
  629. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  630. opts.onSuccess5 = test3_onUnsubscribe;
  631. opts.context = cd;
  632. rc = MQTTAsync_unsubscribe(cd->c, cd->test_topic, &opts);
  633. assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  634. }
  635. MQTTAsync_freeMessage(&message);
  636. MQTTAsync_free(topicName);
  637. return 1;
  638. }
  639. void test3_onSubscribe(void* context, MQTTAsync_successData5* response)
  640. {
  641. client_data* cd = (client_data*)context;
  642. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  643. int rc;
  644. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback \"%s\"", cd->clientid);
  645. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  646. pubmsg.payloadlen = 11;
  647. pubmsg.qos = 2;
  648. pubmsg.retained = 0;
  649. rc = MQTTAsync_send(cd->c, cd->test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, NULL);
  650. assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  651. }
  652. void test3_onConnect(void* context, MQTTAsync_successData5* response)
  653. {
  654. client_data* cd = (client_data*)context;
  655. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  656. int rc;
  657. MyLog(LOGA_DEBUG, "In connect onSuccess callback, \"%s\"", cd->clientid);
  658. opts.onSuccess5 = test3_onSubscribe;
  659. opts.context = cd;
  660. rc = MQTTAsync_subscribe(cd->c, cd->test_topic, 2, &opts);
  661. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  662. if (rc != MQTTASYNC_SUCCESS)
  663. test_finished++;
  664. }
  665. void test3_onFailure(void* context, MQTTAsync_failureData5* response)
  666. {
  667. client_data* cd = (client_data*)context;
  668. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  669. assert("Should have connected", 0, "%s failed to connect\n", cd->clientid);
  670. MyLog(LOGA_DEBUG, "In connect onFailure callback, \"%s\" rc %d\n", cd->clientid, response ? response->reasonCode : -999);
  671. if (response && response->message)
  672. MyLog(LOGA_DEBUG, "In connect onFailure callback, \"%s\"\n", response->message);
  673. test_finished++;
  674. }
  675. /*********************************************************************
  676. Test3: More than one client object - simultaneous working.
  677. *********************************************************************/
  678. int test3(struct Options options)
  679. {
  680. #define num_clients 10
  681. int subsqos = 2;
  682. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  683. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  684. int rc = 0;
  685. int i;
  686. client_data clientdata[num_clients];
  687. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  688. test_finished = 0;
  689. MyLog(LOGA_INFO, "Starting test 3 - multiple connections");
  690. fprintf(xml, "<testcase classname=\"test4\" name=\"multiple connections\"");
  691. global_start_time = start_clock();
  692. for (i = 0; i < num_clients; ++i)
  693. {
  694. sprintf(clientdata[i].clientid, "async_test3_num_%d", i);
  695. sprintf(clientdata[i].test_topic, "async test3 topic num %d", i);
  696. clientdata[i].index = i;
  697. clientdata[i].message_count = 0;
  698. createOpts.MQTTVersion = MQTTVERSION_5;
  699. rc = MQTTAsync_createWithOptions(&(clientdata[i].c), options.connection, clientdata[i].clientid,
  700. MQTTCLIENT_PERSISTENCE_NONE, NULL, &createOpts);
  701. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  702. rc = MQTTAsync_setCallbacks(clientdata[i].c, &clientdata[i], NULL, test3_messageArrived, NULL);
  703. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  704. opts.keepAliveInterval = 20;
  705. opts.username = "testuser";
  706. opts.password = "testpassword";
  707. opts.MQTTVersion = options.MQTTVersion;
  708. opts.will = &wopts;
  709. opts.will->message = "will message";
  710. opts.will->qos = 1;
  711. opts.will->retained = 0;
  712. opts.will->topicName = "will topic";
  713. opts.onSuccess5 = test3_onConnect;
  714. opts.onFailure5 = test3_onFailure;
  715. opts.context = &clientdata[i];
  716. MyLog(LOGA_DEBUG, "Connecting");
  717. rc = MQTTAsync_connect(clientdata[i].c, &opts);
  718. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  719. }
  720. while (test_finished < num_clients)
  721. {
  722. MyLog(LOGA_DEBUG, "num_clients %d test_finished %d\n", num_clients, test_finished);
  723. MySleep(100);
  724. }
  725. MyLog(LOGA_DEBUG, "TEST3: destroying clients");
  726. for (i = 0; i < num_clients; ++i)
  727. MQTTAsync_destroy(&clientdata[i].c);
  728. //exit:
  729. MyLog(LOGA_INFO, "TEST3: test %s. %d tests run, %d failures.",
  730. (failures == 0) ? "passed" : "failed", tests, failures);
  731. write_test_result();
  732. return failures;
  733. }
  734. void* test4_payload = NULL;
  735. int test4_payloadlen = 0;
  736. void test4_onPublish(void* context, MQTTAsync_successData5* response)
  737. {
  738. MQTTAsync c = (MQTTAsync)context;
  739. MyLog(LOGA_DEBUG, "In publish onSuccess callback, context %p", context);
  740. }
  741. int test4_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
  742. {
  743. MQTTAsync c = (MQTTAsync)context;
  744. static int message_count = 0;
  745. int rc, i;
  746. MyLog(LOGA_DEBUG, "In messageArrived callback %p", c);
  747. assert("Message size correct", message->payloadlen == test4_payloadlen,
  748. "message size was %d", message->payloadlen);
  749. for (i = 0; i < options.size; ++i)
  750. {
  751. if (((char*)test4_payload)[i] != ((char*)message->payload)[i])
  752. {
  753. assert("Message contents correct", ((char*)test4_payload)[i] != ((char*)message->payload)[i],
  754. "message content was %c", ((char*)message->payload)[i]);
  755. break;
  756. }
  757. }
  758. if (++message_count == 1)
  759. {
  760. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  761. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  762. pubmsg.payload = test4_payload;
  763. pubmsg.payloadlen = test4_payloadlen;
  764. pubmsg.qos = 1;
  765. pubmsg.retained = 0;
  766. opts.onSuccess5 = test4_onPublish;
  767. opts.context = c;
  768. rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
  769. }
  770. else if (message_count == 2)
  771. {
  772. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  773. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  774. pubmsg.payload = test4_payload;
  775. pubmsg.payloadlen = test4_payloadlen;
  776. pubmsg.qos = 0;
  777. pubmsg.retained = 0;
  778. opts.onSuccess5 = test4_onPublish;
  779. opts.context = c;
  780. rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
  781. }
  782. else
  783. {
  784. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  785. MQTTProperties props = MQTTProperties_initializer;
  786. MQTTProperty property;
  787. opts.onSuccess5 = test1_onUnsubscribe;
  788. opts.context = c;
  789. property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
  790. property.value.data.data = "test user property";
  791. property.value.data.len = (int)strlen(property.value.data.data);
  792. property.value.value.data = "test user property value";
  793. property.value.value.len = (int)strlen(property.value.value.data);
  794. MQTTProperties_add(&props, &property);
  795. opts.properties = props;
  796. rc = MQTTAsync_unsubscribe(c, test_topic, &opts);
  797. assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  798. MQTTProperties_free(&props);
  799. }
  800. MQTTAsync_freeMessage(&message);
  801. MQTTAsync_free(topicName);
  802. return 1;
  803. }
  804. int test4_packet_size = 10000;
  805. #if !defined(min)
  806. #define min(a, b) ((a < b) ? a : b)
  807. #endif
  808. void test4_onSubscribe(void* context, MQTTAsync_successData5* response)
  809. {
  810. MQTTAsync c = (MQTTAsync)context;
  811. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  812. int rc, i;
  813. int max_packet_size = min(test4_packet_size, options.size);
  814. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p", c);
  815. pubmsg.payload = test4_payload = malloc(options.size);
  816. pubmsg.payloadlen = test4_payloadlen = options.size;
  817. MyLog(LOGA_INFO, "Max packet size %d", max_packet_size);
  818. srand(33);
  819. for (i = 0; i < (max_packet_size-100); ++i)
  820. ((char*)pubmsg.payload)[i] = rand() % 256;
  821. pubmsg.qos = 2;
  822. pubmsg.retained = 0;
  823. rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, NULL);
  824. }
  825. void test4_onConnect(void* context, MQTTAsync_successData5* response)
  826. {
  827. MQTTAsync c = (MQTTAsync)context;
  828. MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
  829. int rc;
  830. test4_packet_size = MQTTProperties_getNumericValue(&response->properties, MQTTPROPERTY_CODE_MAXIMUM_PACKET_SIZE);
  831. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  832. opts.onSuccess5 = test4_onSubscribe;
  833. opts.context = c;
  834. rc = MQTTAsync_subscribe(c, test_topic, 2, &opts);
  835. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  836. if (rc != MQTTASYNC_SUCCESS)
  837. test_finished = 1;
  838. }
  839. /*********************************************************************
  840. Test4: Send and receive big messages
  841. *********************************************************************/
  842. int test4(struct Options options)
  843. {
  844. int subsqos = 2;
  845. MQTTAsync c;
  846. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  847. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  848. int rc = 0;
  849. char* test_topic = "C client test4";
  850. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  851. test_finished = failures = 0;
  852. MyLog(LOGA_INFO, "Starting test 4 - big messages");
  853. fprintf(xml, "<testcase classname=\"test4\" name=\"big messages\"");
  854. global_start_time = start_clock();
  855. createOpts.MQTTVersion = MQTTVERSION_5;
  856. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test_4",
  857. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  858. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  859. if (rc != MQTTASYNC_SUCCESS)
  860. {
  861. MQTTAsync_destroy(&c);
  862. goto exit;
  863. }
  864. rc = MQTTAsync_setCallbacks(c, c, NULL, test4_messageArrived, NULL);
  865. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  866. opts.keepAliveInterval = 20;
  867. opts.username = "testuser";
  868. opts.password = "testpassword";
  869. opts.MQTTVersion = options.MQTTVersion;
  870. opts.will = &wopts;
  871. opts.will->message = "will message";
  872. opts.will->qos = 1;
  873. opts.will->retained = 0;
  874. opts.will->topicName = "will topic";
  875. opts.will = NULL;
  876. opts.onSuccess5 = test4_onConnect;
  877. opts.onFailure5 = NULL;
  878. opts.context = c;
  879. MyLog(LOGA_DEBUG, "Connecting");
  880. rc = MQTTAsync_connect(c, &opts);
  881. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  882. if (rc != MQTTASYNC_SUCCESS)
  883. goto exit;
  884. while (!test_finished)
  885. MySleep(100);
  886. MQTTAsync_destroy(&c);
  887. exit:
  888. MyLog(LOGA_INFO, "TEST4: test %s. %d tests run, %d failures.",
  889. (failures == 0) ? "passed" : "failed", tests, failures);
  890. write_test_result();
  891. return failures;
  892. }
  893. void test5_onConnectFailure(void* context, MQTTAsync_failureData5* response)
  894. {
  895. MQTTAsync c = (MQTTAsync)context;
  896. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  897. MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
  898. MyLog(LOGA_INFO, "Connack rc is %d", response ? response->reasonCode : -999);
  899. test_finished = 1;
  900. }
  901. void test5_onConnect(void* context, MQTTAsync_successData5* response)
  902. {
  903. MQTTAsync c = (MQTTAsync)context;
  904. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  905. MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
  906. test_finished = 1;
  907. }
  908. /********************************************************************
  909. Test5: Connack return codes
  910. *********************************************************************/
  911. int test5(struct Options options)
  912. {
  913. int subsqos = 2;
  914. MQTTAsync c;
  915. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  916. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  917. int rc = 0;
  918. char* test_topic = "C client test1";
  919. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  920. test_finished = failures = 0;
  921. MyLog(LOGA_INFO, "Starting test 5 - connack return codes");
  922. fprintf(xml, "<testcase classname=\"test45\" name=\"connack return codes\"");
  923. global_start_time = start_clock();
  924. createOpts.MQTTVersion = MQTTVERSION_5;
  925. rc = MQTTAsync_createWithOptions(&c, options.connection, "a clientid that is too long to be accepted",
  926. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  927. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  928. if (rc != MQTTASYNC_SUCCESS)
  929. {
  930. MQTTAsync_destroy(&c);
  931. goto exit;
  932. }
  933. rc = MQTTAsync_setCallbacks(c, c, NULL, test1_messageArrived, NULL);
  934. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  935. opts.onSuccess5 = test5_onConnect;
  936. opts.onFailure5 = test5_onConnectFailure;
  937. opts.context = c;
  938. MyLog(LOGA_DEBUG, "Connecting");
  939. rc = MQTTAsync_connect(c, &opts);
  940. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  941. if (rc != MQTTASYNC_SUCCESS)
  942. goto exit;
  943. while (!test_finished)
  944. MySleep(100);
  945. MQTTAsync_destroy(&c);
  946. exit:
  947. MyLog(LOGA_INFO, "TEST5: test %s. %d tests run, %d failures.",
  948. (failures == 0) ? "passed" : "failed", tests, failures);
  949. write_test_result();
  950. return failures;
  951. }
  952. typedef struct
  953. {
  954. MQTTAsync c;
  955. int should_fail;
  956. } test6_client_info;
  957. void test6_onConnectFailure(void* context, MQTTAsync_failureData5* response)
  958. {
  959. test6_client_info cinfo = *(test6_client_info*)context;
  960. MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
  961. if (response)
  962. MyLog(LOGA_INFO, "Connack rc is %d", response->reasonCode);
  963. assert("Should fail to connect", cinfo.should_fail, "should_fail was %d", cinfo.should_fail);
  964. test_finished = 1;
  965. }
  966. void test6_onConnect(void* context, MQTTAsync_successData5* response)
  967. {
  968. test6_client_info cinfo = *(test6_client_info*)context;
  969. MyLog(LOGA_DEBUG, "In connect success callback, context %p", context);
  970. assert("Should connect correctly", !cinfo.should_fail, "should_fail was %d", cinfo.should_fail);
  971. test_finished = 1;
  972. }
  973. void test6_onDisconnect5(void* context, MQTTAsync_successData5* response)
  974. {
  975. test6_client_info cinfo = *(test6_client_info*)context;
  976. MyLog(LOGA_DEBUG, "In onDisconnect5 callback %p", cinfo.c);
  977. test_finished = 1;
  978. }
  979. /********************************************************************
  980. Test6: HA connections
  981. *********************************************************************/
  982. int test6(struct Options options)
  983. {
  984. int subsqos = 2;
  985. test6_client_info cinfo;
  986. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  987. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  988. int rc = 0;
  989. char* test_topic = "C client test1";
  990. char* uris[2] = {options.connection, options.connection};
  991. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  992. MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
  993. failures = 0;
  994. MyLog(LOGA_INFO, "Starting test 6 - HA connections");
  995. fprintf(xml, "<testcase classname=\"test4\" name=\"HA connections\"");
  996. global_start_time = start_clock();
  997. test_finished = 0;
  998. cinfo.should_fail = 1; /* fail to connect */
  999. createOpts.MQTTVersion = MQTTVERSION_5;
  1000. rc = MQTTAsync_createWithOptions(&cinfo.c, "tcp://rubbish:1883", "async ha connection",
  1001. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  1002. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  1003. if (rc != MQTTASYNC_SUCCESS)
  1004. {
  1005. MQTTAsync_destroy(&cinfo.c);
  1006. goto exit;
  1007. }
  1008. rc = MQTTAsync_setCallbacks(cinfo.c, cinfo.c, NULL, test1_messageArrived, NULL);
  1009. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1010. opts.onSuccess5 = test6_onConnect;
  1011. opts.onFailure5 = test6_onConnectFailure;
  1012. opts.context = &cinfo;
  1013. opts.MQTTVersion = options.MQTTVersion;
  1014. MyLog(LOGA_DEBUG, "Connecting");
  1015. rc = MQTTAsync_connect(cinfo.c, &opts);
  1016. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1017. if (rc != MQTTASYNC_SUCCESS)
  1018. goto exit;
  1019. while (!test_finished)
  1020. MySleep(100);
  1021. test_finished = 0;
  1022. cinfo.should_fail = 0; /* should connect through the serverURIs in connect options*/
  1023. opts.serverURIs = uris;
  1024. opts.serverURIcount = 2;
  1025. MyLog(LOGA_DEBUG, "Connecting");
  1026. rc = MQTTAsync_connect(cinfo.c, &opts);
  1027. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1028. if (rc != MQTTASYNC_SUCCESS)
  1029. goto exit;
  1030. while (!test_finished)
  1031. MySleep(100);
  1032. test_finished = 0;
  1033. dopts.timeout = 0;
  1034. dopts.onSuccess5 = test6_onDisconnect5;
  1035. dopts.context = cinfo.c;
  1036. dopts.timeout = 0;
  1037. MQTTAsync_disconnect(cinfo.c, &dopts);
  1038. while (!test_finished)
  1039. MySleep(100);
  1040. MQTTAsync_destroy(&cinfo.c);
  1041. exit:
  1042. MyLog(LOGA_INFO, "TEST6: test %s. %d tests run, %d failures.",
  1043. (failures == 0) ? "passed" : "failed", tests, failures);
  1044. write_test_result();
  1045. return failures;
  1046. }
  1047. /********************************************************************
  1048. Test7: Persistence
  1049. *********************************************************************/
  1050. char* test7_topic = "C client test7";
  1051. int test7_messageCount = 0;
  1052. void test7_onDisconnectFailure(void* context, MQTTAsync_failureData* response)
  1053. {
  1054. MQTTAsync c = (MQTTAsync)context;
  1055. MyLog(LOGA_DEBUG, "In onDisconnect failure callback %p", c);
  1056. assert("Successful disconnect", 0, "disconnect failed", 0);
  1057. test_finished = 1;
  1058. }
  1059. void test7_onDisconnectFailure5(void* context, MQTTAsync_failureData5* response)
  1060. {
  1061. MQTTAsync c = (MQTTAsync)context;
  1062. MyLog(LOGA_DEBUG, "In onDisconnect5 failure callback %p", c);
  1063. assert("Successful disconnect", 0, "disconnect failed", 0);
  1064. test_finished = 1;
  1065. }
  1066. void test7_onDisconnect(void* context, MQTTAsync_successData* response)
  1067. {
  1068. MQTTAsync c = (MQTTAsync)context;
  1069. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  1070. test_finished = 1;
  1071. }
  1072. void test7_onDisconnect5(void* context, MQTTAsync_successData5* response)
  1073. {
  1074. MQTTAsync c = (MQTTAsync)context;
  1075. MyLog(LOGA_DEBUG, "In onDisconnect5 callback %p", c);
  1076. test_finished = 1;
  1077. }
  1078. void test7_onUnsubscribe(void* context, MQTTAsync_successData* response)
  1079. {
  1080. MQTTAsync c = (MQTTAsync)context;
  1081. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  1082. int rc;
  1083. MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess callback %p", c);
  1084. opts.onSuccess = test7_onDisconnect;
  1085. opts.context = c;
  1086. rc = MQTTAsync_disconnect(c, &opts);
  1087. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1088. }
  1089. void test7_onUnsubscribe5(void* context, MQTTAsync_successData5* response)
  1090. {
  1091. MQTTAsync c = (MQTTAsync)context;
  1092. MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
  1093. int rc;
  1094. MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess5 callback %p", c);
  1095. opts.onSuccess5 = test7_onDisconnect5;
  1096. opts.context = c;
  1097. rc = MQTTAsync_disconnect(c, &opts);
  1098. assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1099. }
  1100. int test7_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
  1101. {
  1102. MQTTAsync c = (MQTTAsync)context;
  1103. static int message_count = 0;
  1104. MyLog(LOGA_DEBUG, "Test7: received message id %d, %.*s", message->msgid,
  1105. message->payloadlen, message->payload);
  1106. test7_messageCount++;
  1107. MQTTAsync_freeMessage(&message);
  1108. MQTTAsync_free(topicName);
  1109. return 1;
  1110. }
  1111. static int test7_subscribed = 0;
  1112. void test7_onSubscribe5(void* context, MQTTAsync_successData5* response)
  1113. {
  1114. MQTTAsync c = (MQTTAsync)context;
  1115. MyLog(LOGA_DEBUG, "In subscribe onSuccess5 callback %p granted qos %d", c, response->reasonCode);
  1116. test7_subscribed = 1;
  1117. }
  1118. void test7_onSubscribe(void* context, MQTTAsync_successData* response)
  1119. {
  1120. MQTTAsync c = (MQTTAsync)context;
  1121. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c,
  1122. response->alt.qos);
  1123. test7_subscribed = 1;
  1124. }
  1125. static int test7_just_connect = 0;
  1126. void test7_onConnect(void* context, MQTTAsync_successData* response)
  1127. {
  1128. MQTTAsync c = (MQTTAsync)context;
  1129. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  1130. int rc;
  1131. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  1132. if (test7_just_connect == 1)
  1133. {
  1134. test7_just_connect = 2;
  1135. return;
  1136. }
  1137. opts.onSuccess = test7_onSubscribe;
  1138. opts.context = c;
  1139. rc = MQTTAsync_subscribe(c, test7_topic, 2, &opts);
  1140. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1141. if (rc != MQTTASYNC_SUCCESS)
  1142. test_finished = 1;
  1143. }
  1144. void test7_onConnect5(void* context, MQTTAsync_successData5* response)
  1145. {
  1146. MQTTAsync c = (MQTTAsync)context;
  1147. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  1148. int rc;
  1149. MyLog(LOGA_DEBUG, "In connect onSuccess5 callback, context %p", context);
  1150. if (test7_just_connect == 1)
  1151. { /* we don't need to subscribe if we reconnected */
  1152. test7_subscribed = 1;
  1153. return;
  1154. }
  1155. opts.onSuccess5 = test7_onSubscribe5;
  1156. opts.context = c;
  1157. rc = MQTTAsync_subscribe(c, test7_topic, 2, &opts);
  1158. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1159. if (rc != MQTTASYNC_SUCCESS)
  1160. test_finished = 1;
  1161. }
  1162. /*********************************************************************
  1163. Test7: Pending tokens
  1164. *********************************************************************/
  1165. int test7_run(int qos, int start_mqtt_version, int restore_mqtt_version)
  1166. {
  1167. int subsqos = 2;
  1168. MQTTAsync c;
  1169. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  1170. MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
  1171. int rc = 0;
  1172. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  1173. MQTTAsync_responseOptions ropts = MQTTAsync_responseOptions_initializer;
  1174. MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
  1175. MQTTAsync_token* tokens = NULL;
  1176. int msg_count = 6;
  1177. MQTTProperty property;
  1178. MQTTProperties props = MQTTProperties_initializer;
  1179. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  1180. int i = 0;
  1181. MyLog(LOGA_INFO, "Starting test 7 - persistence, qos %d, MQTT versions: %s then %s", qos,
  1182. (start_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1",
  1183. (restore_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1");
  1184. fprintf(xml, "<testcase classname=\"test45\" name=\"pending tokens\"");
  1185. global_start_time = start_clock();
  1186. test_finished = 0;
  1187. createOpts.MQTTVersion = start_mqtt_version;
  1188. MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
  1189. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
  1190. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  1191. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  1192. if (rc != MQTTASYNC_SUCCESS)
  1193. {
  1194. MQTTAsync_destroy(&c);
  1195. goto exit;
  1196. }
  1197. rc = MQTTAsync_setCallbacks(c, c, NULL, test7_messageArrived, NULL);
  1198. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1199. opts.keepAliveInterval = 20;
  1200. opts.username = "testuser";
  1201. opts.password = "testpassword";
  1202. opts.MQTTVersion = start_mqtt_version;
  1203. opts.will = &wopts;
  1204. opts.will->message = "will message";
  1205. opts.will->qos = 1;
  1206. opts.will->retained = 0;
  1207. opts.will->topicName = "will topic";
  1208. opts.will = NULL;
  1209. opts.context = c;
  1210. /* connect clean and then leave messages lying around */
  1211. test_finished = 0;
  1212. test7_subscribed = 0;
  1213. MyLog(LOGA_DEBUG, "Connecting");
  1214. if (start_mqtt_version == MQTTVERSION_5)
  1215. {
  1216. opts.cleanstart = 1;
  1217. test7_just_connect = 0;
  1218. opts.connectProperties = &props;
  1219. property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
  1220. property.value.integer4 = 999999;
  1221. MQTTProperties_add(opts.connectProperties, &property);
  1222. opts.onSuccess5 = test7_onConnect5;
  1223. opts.onFailure5 = NULL;
  1224. rc = MQTTAsync_connect(c, &opts);
  1225. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1226. MQTTProperties_free(opts.connectProperties);
  1227. if (rc != MQTTASYNC_SUCCESS)
  1228. goto exit;
  1229. }
  1230. else
  1231. { /* MQTT 3 version */
  1232. opts.cleanstart = 0;
  1233. opts.cleansession = 1; /* clean up */
  1234. test7_just_connect = 1;
  1235. opts.onSuccess = test7_onConnect;
  1236. opts.onFailure = NULL;
  1237. rc = MQTTAsync_connect(c, &opts);
  1238. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1239. if (rc != MQTTASYNC_SUCCESS)
  1240. goto exit;
  1241. while (test7_just_connect < 2)
  1242. MySleep(100);
  1243. test_finished = 0;
  1244. dopts.onSuccess5 = NULL;
  1245. dopts.onFailure5 = NULL;
  1246. dopts.onFailure = test7_onDisconnectFailure;
  1247. dopts.onSuccess = test7_onDisconnect;
  1248. rc = MQTTAsync_disconnect(c, &dopts);
  1249. assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1250. while (!test_finished)
  1251. MySleep(100);
  1252. test_finished = 0;
  1253. opts.cleansession = 0; /* now it's clean */
  1254. test7_just_connect = 0;
  1255. rc = MQTTAsync_connect(c, &opts);
  1256. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1257. if (rc != MQTTASYNC_SUCCESS)
  1258. goto exit;
  1259. }
  1260. while (!test7_subscribed)
  1261. MySleep(100);
  1262. test7_messageCount = 0;
  1263. pubmsg.payload = "first test of a much longer message that we can shorten to the extent that we need to payload up to 11";
  1264. pubmsg.payloadlen = 11;
  1265. pubmsg.qos = qos;
  1266. pubmsg.retained = 0;
  1267. rc = MQTTAsync_send(c, test7_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &ropts);
  1268. assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1269. MyLog(LOGA_DEBUG, "Token was %d", ropts.token);
  1270. rc = MQTTAsync_waitForCompletion(c, ropts.token, 5000L);
  1271. assert("Good rc from waitForCompletion", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1272. rc = MQTTAsync_isComplete(c, ropts.token);
  1273. assert("1 rc from isComplete", rc == 1, "rc was %d", rc);
  1274. i = 0;
  1275. pubmsg.qos = qos;
  1276. for (i = 0; i < msg_count; ++i)
  1277. {
  1278. char buf[100];
  1279. sprintf(buf, "%d a much longer message that we can shorten to the extent that we need", i);
  1280. pubmsg.payload = buf;
  1281. pubmsg.payloadlen = 11 + i;
  1282. pubmsg.retained = 0;
  1283. rc = MQTTAsync_sendMessage(c, test7_topic, &pubmsg, &ropts);
  1284. }
  1285. /* disconnect immediately without receiving the incoming messages */
  1286. dopts.timeout = 0;
  1287. if (start_mqtt_version == MQTTVERSION_5)
  1288. dopts.onSuccess5 = test7_onDisconnect5;
  1289. else
  1290. dopts.onSuccess = test7_onDisconnect;
  1291. dopts.context = c;
  1292. dopts.timeout = 0;
  1293. MQTTAsync_disconnect(c, &dopts); /* now there should be "orphaned" publications */
  1294. while (!test_finished)
  1295. MySleep(100);
  1296. test_finished = 0;
  1297. rc = MQTTAsync_getPendingTokens(c, &tokens);
  1298. assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1299. assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
  1300. if (tokens)
  1301. MQTTAsync_free(tokens);
  1302. MQTTProperties_free(opts.connectProperties);
  1303. MQTTAsync_destroy(&c); /* force re-reading persistence on create */
  1304. MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
  1305. createOpts.MQTTVersion = restore_mqtt_version;
  1306. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
  1307. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  1308. if (start_mqtt_version == MQTTVERSION_5 && restore_mqtt_version == MQTTVERSION_3_1_1)
  1309. assert("Persistence error from create", rc == MQTTASYNC_PERSISTENCE_ERROR, "rc was %d", rc);
  1310. else
  1311. assert("Good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1312. if (rc != MQTTASYNC_SUCCESS)
  1313. {
  1314. //MQTTAsync_destroy(&c);
  1315. goto exit;
  1316. }
  1317. rc = MQTTAsync_getPendingTokens(c, &tokens);
  1318. assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1319. assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
  1320. if (tokens)
  1321. {
  1322. int i = 0;
  1323. while (tokens[i] != -1)
  1324. MyLog(LOGA_DEBUG, "Delivery token %d", tokens[i++]);
  1325. MQTTAsync_free(tokens);
  1326. assert("no of tokens should be > 0", i > 0, "no of tokens %d", i);
  1327. }
  1328. rc = MQTTAsync_setCallbacks(c, c, NULL, test7_messageArrived, NULL);
  1329. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1330. MyLog(LOGA_DEBUG, "Reconnecting");
  1331. opts.context = c;
  1332. opts.MQTTVersion = restore_mqtt_version;
  1333. if (restore_mqtt_version == MQTTVERSION_5)
  1334. {
  1335. opts.cleanstart = opts.cleansession = 0;
  1336. test7_just_connect = 1;
  1337. opts.connectProperties = &props;
  1338. property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
  1339. property.value.integer4 = 0; /* clean up at end of test */
  1340. MQTTProperties_add(opts.connectProperties, &property);
  1341. opts.onSuccess = NULL;
  1342. opts.onFailure = NULL;
  1343. opts.onSuccess5 = test7_onConnect5;
  1344. opts.onFailure5 = NULL;
  1345. rc = MQTTAsync_connect(c, &opts);
  1346. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1347. MQTTProperties_free(opts.connectProperties);
  1348. if (rc != MQTTASYNC_SUCCESS)
  1349. goto exit;
  1350. }
  1351. else
  1352. { /* MQTT 3 version */
  1353. opts.cleanstart = opts.cleansession = 0;
  1354. test7_just_connect = 1;
  1355. opts.onSuccess5 = NULL;
  1356. opts.onFailure5 = NULL;
  1357. opts.onSuccess = test7_onConnect;
  1358. opts.onFailure = NULL;
  1359. rc = MQTTAsync_connect(c, &opts);
  1360. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1361. if (rc != MQTTASYNC_SUCCESS)
  1362. goto exit;
  1363. }
  1364. waitForNoPendingTokens(c);
  1365. if (qos == 2)
  1366. assert("no of messages should be count", test7_messageCount == msg_count + 1,
  1367. "messages received %d\n", test7_messageCount);
  1368. else if (qos == 1)
  1369. assert("no of messages should be at least count", test7_messageCount >= msg_count + 1,
  1370. "messages received %d\n", test7_messageCount);
  1371. if (restore_mqtt_version == MQTTVERSION_5)
  1372. {
  1373. dopts.onFailure5 = test7_onDisconnectFailure5;
  1374. dopts.onSuccess5 = test7_onDisconnect5;
  1375. }
  1376. else
  1377. {
  1378. dopts.onFailure = test7_onDisconnectFailure;
  1379. dopts.onSuccess = test7_onDisconnect;
  1380. }
  1381. dopts.timeout = 300;
  1382. rc = MQTTAsync_disconnect(c, &dopts);
  1383. assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1384. i = 0;
  1385. while (!test_finished && ++i < 20)
  1386. MySleep(100);
  1387. assert("Test finished should be true", test_finished,
  1388. "test_finished was %d", test_finished);
  1389. MQTTAsync_destroy(&c);
  1390. exit:
  1391. MyLog(LOGA_INFO, "TEST7: test %s. %d tests run, %d failures.",
  1392. (failures == 0) ? "passed" : "failed", tests, failures);
  1393. write_test_result();
  1394. return failures;
  1395. }
  1396. int test7(struct Options options)
  1397. {
  1398. int rc = 0;
  1399. fprintf(xml, "<testcase classname=\"test7\" name=\"persistence\"");
  1400. global_start_time = start_clock();
  1401. rc = test7_run(1, MQTTVERSION_5, MQTTVERSION_5) +
  1402. test7_run(2, MQTTVERSION_5, MQTTVERSION_5) +
  1403. test7_run(1, MQTTVERSION_3_1_1, MQTTVERSION_3_1_1) +
  1404. test7_run(2, MQTTVERSION_3_1_1, MQTTVERSION_3_1_1) /*+
  1405. test7_run(2, MQTTVERSION_3_1_1, MQTTVERSION_5) +
  1406. test7_run(2, MQTTVERSION_5, MQTTVERSION_3_1_1)*/;
  1407. fprintf(xml, " time=\"%ld\" >\n", elapsed(global_start_time) / 1000);
  1408. if (cur_output != output)
  1409. {
  1410. fprintf(xml, "%s", output);
  1411. cur_output = output;
  1412. }
  1413. fprintf(xml, "</testcase>\n");
  1414. return rc;
  1415. }
  1416. /*********************************************************************
  1417. Test8: Incomplete commands and requests
  1418. *********************************************************************/
  1419. char* test8_topic = "C client test8";
  1420. int test8_messageCount = 0;
  1421. int test8_subscribed = 0;
  1422. int test8_publishFailures = 0;
  1423. void test8_onPublish(void* context, MQTTAsync_successData5* response)
  1424. {
  1425. MQTTAsync c = (MQTTAsync)context;
  1426. MyLog(LOGA_DEBUG, "In publish onSuccess callback %p token %d", c, response->token);
  1427. }
  1428. void test8_onPublishFailure(void* context, MQTTAsync_failureData5* response)
  1429. {
  1430. MQTTAsync c = (MQTTAsync)context;
  1431. MyLog(LOGA_DEBUG, "In onPublish failure callback %p", c);
  1432. assert("Response code should be interrupted", response->code == MQTTASYNC_OPERATION_INCOMPLETE,
  1433. "rc was %d", response->code);
  1434. test8_publishFailures++;
  1435. }
  1436. void test8_onDisconnectFailure(void* context, MQTTAsync_failureData5* response)
  1437. {
  1438. MQTTAsync c = (MQTTAsync)context;
  1439. MyLog(LOGA_DEBUG, "In onDisconnect failure callback %p", c);
  1440. assert("Successful disconnect", 0, "disconnect failed", 0);
  1441. test_finished = 1;
  1442. }
  1443. void test8_onDisconnect(void* context, MQTTAsync_successData5* response)
  1444. {
  1445. MQTTAsync c = (MQTTAsync)context;
  1446. MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
  1447. test_finished = 1;
  1448. }
  1449. void test8_onSubscribe(void* context, MQTTAsync_successData5* response)
  1450. {
  1451. MQTTAsync c = (MQTTAsync)context;
  1452. MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c, response->reasonCode);
  1453. test8_subscribed = 1;
  1454. }
  1455. void test8_onConnect(void* context, MQTTAsync_successData5* response)
  1456. {
  1457. MQTTAsync c = (MQTTAsync)context;
  1458. MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
  1459. int rc;
  1460. MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
  1461. opts.onSuccess5 = test8_onSubscribe;
  1462. opts.context = c;
  1463. rc = MQTTAsync_subscribe(c, test8_topic, 2, &opts);
  1464. assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1465. if (rc != MQTTASYNC_SUCCESS)
  1466. test_finished = 1;
  1467. }
  1468. int test8_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
  1469. {
  1470. MQTTAsync c = (MQTTAsync)context;
  1471. static int message_count = 0;
  1472. MyLog(LOGA_DEBUG, "Test8: received message id %d", message->msgid);
  1473. test8_messageCount++;
  1474. MQTTAsync_freeMessage(&message);
  1475. MQTTAsync_free(topicName);
  1476. return 1;
  1477. }
  1478. int test8(struct Options options)
  1479. {
  1480. int subsqos = 2;
  1481. MQTTAsync c;
  1482. MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer5;
  1483. int rc = 0;
  1484. MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
  1485. MQTTAsync_responseOptions ropts = MQTTAsync_responseOptions_initializer;
  1486. MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
  1487. MQTTAsync_token* tokens = NULL;
  1488. int msg_count = 6;
  1489. MQTTProperty property;
  1490. MQTTProperties props = MQTTProperties_initializer;
  1491. MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
  1492. MyLog(LOGA_INFO, "Starting test 8 - incomplete commands");
  1493. fprintf(xml, "<testcase classname=\"test4\" name=\"incomplete commands\"");
  1494. global_start_time = start_clock();
  1495. test_finished = 0;
  1496. createOpts.MQTTVersion = MQTTVERSION_5;
  1497. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test8",
  1498. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  1499. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  1500. if (rc != MQTTASYNC_SUCCESS)
  1501. {
  1502. MQTTAsync_destroy(&c);
  1503. goto exit;
  1504. }
  1505. rc = MQTTAsync_setCallbacks(c, c, NULL, test8_messageArrived, NULL);
  1506. assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1507. opts.keepAliveInterval = 20;
  1508. opts.username = "testuser";
  1509. opts.password = "testpassword";
  1510. opts.MQTTVersion = options.MQTTVersion;
  1511. opts.onFailure = NULL;
  1512. opts.context = c;
  1513. MyLog(LOGA_DEBUG, "Connecting");
  1514. opts.cleanstart = 1;
  1515. opts.onSuccess5 = test8_onConnect;
  1516. rc = MQTTAsync_connect(c, &opts);
  1517. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1518. if (rc != MQTTASYNC_SUCCESS)
  1519. goto exit;
  1520. while (!test8_subscribed)
  1521. MySleep(100);
  1522. int i = 0;
  1523. pubmsg.qos = 2;
  1524. ropts.onSuccess5 = test8_onPublish;
  1525. ropts.onFailure5 = test8_onPublishFailure;
  1526. ropts.context = c;
  1527. for (i = 0; i < msg_count; ++i)
  1528. {
  1529. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  1530. pubmsg.payloadlen = 11;
  1531. pubmsg.qos = (pubmsg.qos == 2) ? 1 : 2; /* alternate */
  1532. pubmsg.retained = 0;
  1533. rc = MQTTAsync_sendMessage(c, test8_topic, &pubmsg, &ropts);
  1534. assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1535. }
  1536. /* disconnect immediately without completing the commands */
  1537. dopts.timeout = 0;
  1538. dopts.onSuccess5 = test8_onDisconnect;
  1539. dopts.context = c;
  1540. rc = MQTTAsync_disconnect(c, &dopts); /* now there should be incomplete commands */
  1541. assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1542. while (!test_finished)
  1543. MySleep(100);
  1544. test_finished = 0;
  1545. waitForNoPendingTokens(c);
  1546. assert("test8_publishFailures > 0", test8_publishFailures > 0,
  1547. "test8_publishFailures = %d", test8_publishFailures);
  1548. /* Now elicit failure callbacks on destroy */
  1549. test8_subscribed = test8_publishFailures = 0;
  1550. MyLog(LOGA_DEBUG, "Connecting");
  1551. opts.onSuccess5 = test8_onConnect;
  1552. property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
  1553. property.value.integer4 = 30;
  1554. MQTTProperties_add(&props, &property);
  1555. opts.connectProperties = &props;
  1556. rc = MQTTAsync_connect(c, &opts);
  1557. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1558. if (rc != MQTTASYNC_SUCCESS)
  1559. goto exit;
  1560. while (!test8_subscribed)
  1561. MySleep(100);
  1562. i = 0;
  1563. pubmsg.qos = 2;
  1564. ropts.onSuccess5 = test8_onPublish;
  1565. ropts.onFailure5 = test8_onPublishFailure;
  1566. ropts.context = c;
  1567. for (i = 0; i < msg_count; ++i)
  1568. {
  1569. pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
  1570. pubmsg.payloadlen = 11;
  1571. pubmsg.qos = (pubmsg.qos == 2) ? 1 : 2; /* alternate */
  1572. pubmsg.retained = 0;
  1573. rc = MQTTAsync_sendMessage(c, test8_topic, &pubmsg, &ropts);
  1574. assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1575. }
  1576. /* disconnect immediately without completing the commands */
  1577. dopts.timeout = 0;
  1578. dopts.onSuccess5 = test8_onDisconnect;
  1579. dopts.context = c;
  1580. rc = MQTTAsync_disconnect(c, &dopts); /* now there should be incomplete commands */
  1581. assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1582. while (!test_finished)
  1583. MySleep(100);
  1584. test_finished = 0;
  1585. rc = MQTTAsync_getPendingTokens(c, &tokens);
  1586. assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1587. assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
  1588. MQTTAsync_free(tokens);
  1589. MQTTProperties_free(&props);
  1590. assert("test8_publishFailures == 0", test8_publishFailures == 0,
  1591. "test8_publishFailures = %d", test8_publishFailures);
  1592. MQTTAsync_destroy(&c);
  1593. assert("test8_publishFailures > 0", test8_publishFailures > 0,
  1594. "test8_publishFailures = %d", test8_publishFailures);
  1595. /* cleanup persistence of any left over message data*/
  1596. MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
  1597. rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test8",
  1598. MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
  1599. assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
  1600. if (rc != MQTTASYNC_SUCCESS)
  1601. {
  1602. MQTTAsync_destroy(&c);
  1603. goto exit;
  1604. }
  1605. test8_subscribed = 0;
  1606. opts.connectProperties = NULL;
  1607. opts.cleanstart = 1;
  1608. opts.context = c;
  1609. rc = MQTTAsync_connect(c, &opts);
  1610. assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1611. if (rc != MQTTASYNC_SUCCESS)
  1612. goto exit;
  1613. while (!test8_subscribed)
  1614. MySleep(100);
  1615. test_finished = 0;
  1616. dopts.context = c;
  1617. rc = MQTTAsync_disconnect(c, &dopts);
  1618. assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
  1619. while (!test_finished)
  1620. MySleep(100);
  1621. MQTTAsync_destroy(&c);
  1622. exit:
  1623. MyLog(LOGA_INFO, "TEST8: test %s. %d tests run, %d failures.",
  1624. (failures == 0) ? "passed" : "failed", tests, failures);
  1625. write_test_result();
  1626. return failures;
  1627. }
  1628. void trace_callback(enum MQTTASYNC_TRACE_LEVELS level, char* message)
  1629. {
  1630. printf("Trace : %d, %s\n", level, message);
  1631. }
  1632. int main(int argc, char** argv)
  1633. {
  1634. int rc = -1;
  1635. int (*tests[])() = {NULL, test1, test2, test3, test4, test5, test6, test7, test8}; /* indexed starting from 1 */
  1636. MQTTAsync_nameValue* info;
  1637. int i;
  1638. xml = fopen("TEST-test4.xml", "w");
  1639. fprintf(xml, "<testsuite name=\"test4\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests)) - 1);
  1640. getopts(argc, argv);
  1641. MQTTAsync_setTraceCallback(trace_callback);
  1642. info = MQTTAsync_getVersionInfo();
  1643. while (info->name)
  1644. {
  1645. MyLog(LOGA_INFO, "%s: %s", info->name, info->value);
  1646. info++;
  1647. }
  1648. for (i = 0; i < options.iterations; ++i)
  1649. {
  1650. if (options.test_no == -1)
  1651. { /* run all the tests */
  1652. for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
  1653. {
  1654. failures = rc = 0;
  1655. MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
  1656. rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */
  1657. }
  1658. }
  1659. else
  1660. {
  1661. if (options.test_no >= ARRAY_SIZE(tests))
  1662. MyLog(LOGA_INFO, "No test number %d", options.test_no);
  1663. else
  1664. {
  1665. MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
  1666. rc = tests[options.test_no](options); /* run just the selected test */
  1667. }
  1668. }
  1669. }
  1670. if (rc == 0)
  1671. MyLog(LOGA_INFO, "verdict pass");
  1672. else
  1673. MyLog(LOGA_INFO, "verdict fail");
  1674. fprintf(xml, "</testsuite>\n");
  1675. fclose(xml);
  1676. return rc;
  1677. }