1 /*******************************************************************************
2 * Copyright (c) 2009, 2017 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 v1.0
6 * and Eclipse Distribution License v1.0 which accompany this distribution.
7 *
8 * The Eclipse Public License is available at
9 * http://www.eclipse.org/legal/epl-v10.html
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 implementation for embedded C client
15 *******************************************************************************/
16
17
18 /**
19 * @file
20 * Tests for the Paho embedded C "high" level client
21 */
22
23 #include <stdio.h>
24 #include <string.h>
25 #include <memory.h>
26 //#define MQTT_DEBUG
27 #include "MQTTClient.h"
28
29 #define DEFAULT_STACK_SIZE -1
30
31 #include "linux.cpp"
32
33 #include <sys/time.h>
34 #include <stdlib.h>
35
36 #define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
37
usage(void)38 void usage(void)
39 {
40 printf("help!!\n");
41 exit(EXIT_FAILURE);
42 }
43
44 struct Options
45 {
46 char* host; /**< connection to system under test. */
47 int port;
48 char* proxy_host;
49 int proxy_port;
50 int verbose;
51 int test_no;
52 int MQTTVersion;
53 int iterations;
54 } options =
55 {
56 (char*)"localhost",
57 1883,
58 (char*)"localhost",
59 1885,
60 0,
61 0,
62 4,
63 1,
64 };
65
getopts(int argc,char ** argv)66 void getopts(int argc, char** argv)
67 {
68 int count = 1;
69
70 while (count < argc)
71 {
72 if (strcmp(argv[count], "--test_no") == 0)
73 {
74 if (++count < argc)
75 options.test_no = atoi(argv[count]);
76 else
77 usage();
78 }
79 else if (strcmp(argv[count], "--host") == 0)
80 {
81 if (++count < argc)
82 {
83 options.host = argv[count];
84 printf("\nSetting host to %s\n", options.host);
85 }
86 else
87 usage();
88 }
89 else if (strcmp(argv[count], "--port") == 0)
90 {
91 if (++count < argc)
92 {
93 options.port = atoi(argv[count]);
94 printf("\nSetting port to %d\n", options.port);
95 }
96 else
97 usage();
98 }
99 else if (strcmp(argv[count], "--proxy_host") == 0)
100 {
101 if (++count < argc)
102 options.proxy_host = argv[count];
103 else
104 usage();
105 }
106 else if (strcmp(argv[count], "--proxy_port") == 0)
107 {
108 if (++count < argc)
109 {
110 options.proxy_port = atoi(argv[count]);
111 printf("\nSetting proxy port to %d\n", options.proxy_port);
112 }
113 else
114 usage();
115 }
116 else if (strcmp(argv[count], "--MQTTversion") == 0)
117 {
118 if (++count < argc)
119 {
120 options.MQTTVersion = atoi(argv[count]);
121 printf("setting MQTT version to %d\n", options.MQTTVersion);
122 }
123 else
124 usage();
125 }
126 else if (strcmp(argv[count], "--iterations") == 0)
127 {
128 if (++count < argc)
129 options.iterations = atoi(argv[count]);
130 else
131 usage();
132 }
133 else if (strcmp(argv[count], "--verbose") == 0)
134 {
135 options.verbose = 1;
136 printf("\nSetting verbose on\n");
137 }
138 count++;
139 }
140 }
141
142
143 #define LOGA_DEBUG 0
144 #define LOGA_INFO 1
145 #include <stdarg.h>
146 #include <time.h>
147 #include <sys/timeb.h>
MyLog(int LOGA_level,const char * format,...)148 void MyLog(int LOGA_level, const char* format, ...)
149 {
150 static char msg_buf[256];
151 va_list args;
152 struct timeb ts;
153
154 struct tm *timeinfo;
155
156 if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
157 return;
158
159 ftime(&ts);
160 timeinfo = localtime(&ts.time);
161 strftime(msg_buf, 80, "%Y%m%d %H%M%S", timeinfo);
162
163 sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
164
165 va_start(args, format);
166 vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
167 va_end(args);
168
169 printf("%s\n", msg_buf);
170 fflush(stdout);
171 }
172
173
174 #if defined(WIN32) || defined(_WINDOWS)
175 #define mqsleep(A) Sleep(1000*A)
176 #define START_TIME_TYPE DWORD
177 static DWORD start_time = 0;
start_clock(void)178 START_TIME_TYPE start_clock(void)
179 {
180 return GetTickCount();
181 }
182 #elif defined(AIX)
183 #define mqsleep sleep
184 #define START_TIME_TYPE struct timespec
start_clock(void)185 START_TIME_TYPE start_clock(void)
186 {
187 static struct timespec start;
188 clock_gettime(CLOCK_REALTIME, &start);
189 return start;
190 }
191 #else
192 #define mqsleep sleep
193 #define START_TIME_TYPE struct timeval
194 /* TODO - unused - remove? static struct timeval start_time; */
start_clock(void)195 START_TIME_TYPE start_clock(void)
196 {
197 struct timeval start_time;
198 gettimeofday(&start_time, NULL);
199 return start_time;
200 }
201 #endif
202
203
204 #if defined(WIN32)
elapsed(START_TIME_TYPE start_time)205 long elapsed(START_TIME_TYPE start_time)
206 {
207 return GetTickCount() - start_time;
208 }
209 #elif defined(AIX)
210 #define assert(a)
elapsed(struct timespec start)211 long elapsed(struct timespec start)
212 {
213 struct timespec now, res;
214
215 clock_gettime(CLOCK_REALTIME, &now);
216 ntimersub(now, start, res);
217 return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
218 }
219 #else
elapsed(START_TIME_TYPE start_time)220 long elapsed(START_TIME_TYPE start_time)
221 {
222 struct timeval now, res;
223
224 gettimeofday(&now, NULL);
225 timersub(&now, &start_time, &res);
226 return (res.tv_sec)*1000 + (res.tv_usec)/1000;
227 }
228 #endif
229
230
231 #define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
232 #define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
233
234 int tests = 0;
235 int failures = 0;
236 FILE* xml;
237 START_TIME_TYPE global_start_time;
238 char output[3000];
239 char* cur_output = output;
240
241
write_test_result(void)242 void write_test_result(void)
243 {
244 long duration = elapsed(global_start_time);
245
246 fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000);
247 if (cur_output != output)
248 {
249 fprintf(xml, "%s", output);
250 cur_output = output;
251 }
252 fprintf(xml, "</testcase>\n");
253 }
254
255
myassert(const char * filename,int lineno,const char * description,int value,const char * format,...)256 void myassert(const char* filename, int lineno, const char* description, int value, const char* format, ...)
257 {
258 ++tests;
259 if (!value)
260 {
261 va_list args;
262
263 ++failures;
264 MyLog(LOGA_INFO, (char*)"Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
265
266 va_start(args, format);
267 vprintf(format, args);
268 va_end(args);
269
270 cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
271 description, filename, lineno);
272 }
273 else
274 MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
275 }
276
277
278 static volatile MQTT::MessageData* test1_message_data = NULL;
279 static MQTT::Message pubmsg;
280
messageArrived(MQTT::MessageData & md)281 void messageArrived(MQTT::MessageData& md)
282 {
283 test1_message_data = &md;
284 MQTT::Message &m = md.message;
285
286 assert("Good message lengths", pubmsg.payloadlen == m.payloadlen,
287 "payloadlen was %d", m.payloadlen);
288
289 if (pubmsg.payloadlen == m.payloadlen)
290 assert("Good message contents", memcmp(m.payload, pubmsg.payload, m.payloadlen) == 0,
291 "payload was %s", m.payload);
292 }
293
294
295 /*********************************************************************
296
297 Test1: single-threaded client
298
299 *********************************************************************/
test1_sendAndReceive(MQTT::Client<IPStack,Countdown,1000> & client,int qos,const char * test_topic)300 void test1_sendAndReceive(MQTT::Client<IPStack, Countdown, 1000>& client, int qos, const char* test_topic)
301 {
302 char* topicName = NULL;
303 int topicLen;
304 int i = 0;
305 int iterations = 50;
306 int rc;
307 int wait_seconds;
308
309 MyLog(LOGA_DEBUG, "%d messages at QoS %d", iterations, qos);
310 memset(&pubmsg, '\0', sizeof(pubmsg));
311 pubmsg.payload = (void*)"a much longer message that we can shorten to the extent that we need to payload up to 11";
312 pubmsg.payloadlen = 11;
313 pubmsg.qos = (MQTT::QoS)qos;
314 pubmsg.retained = false;
315 pubmsg.dup = false;
316
317 for (i = 0; i < iterations; ++i)
318 {
319 test1_message_data = NULL;
320 rc = client.publish(test_topic, pubmsg);
321 assert("Good rc from publish", rc == MQTT::SUCCESS, "rc was %d", rc);
322
323 /* wait for the message to be received */
324 wait_seconds = 10;
325 while ((test1_message_data == NULL) && (wait_seconds-- > 0))
326 {
327 client.yield(100);
328 }
329 assert("Message Arrived", wait_seconds > 0, "Time out waiting for message %d\n", i);
330
331 if (!test1_message_data)
332 printf("No message received within timeout period\n");
333 }
334
335 /* wait to receive any outstanding messages */
336 wait_seconds = 2;
337 while (wait_seconds-- > 0)
338 {
339 client.yield(1000);
340 }
341 }
342
343
test1(struct Options options)344 int test1(struct Options options)
345 {
346 MQTT::QoS subsqos = MQTT::QOS2;
347 int rc = 0;
348 const char* test_topic = "C client test1";
349
350 fprintf(xml, "<testcase classname=\"test1\" name=\"single threaded client using receive\"");
351 global_start_time = start_clock();
352 failures = 0;
353 MyLog(LOGA_INFO, "Starting test 1 - single threaded client using receive");
354
355 IPStack ipstack = IPStack();
356 MQTT::Client<IPStack, Countdown, 1000> client = MQTT::Client<IPStack, Countdown, 1000>(ipstack);
357
358 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
359 data.willFlag = 1;
360 data.MQTTVersion = options.MQTTVersion;
361 data.clientID.cstring = (char*)"single-threaded-test";
362 data.username.cstring = (char*)"testuser";
363 data.password.cstring = (char*)"testpassword";
364
365 data.keepAliveInterval = 20;
366 data.cleansession = 1;
367
368 data.will.message.cstring = (char*)"will message";
369 data.will.qos = 1;
370 data.will.retained = 0;
371 data.will.topicName.cstring = (char*)"will topic";
372
373 MyLog(LOGA_DEBUG, "Connecting");
374 rc = ipstack.connect(options.host, options.port);
375 assert("Good rc from TCP connect", rc == MQTT::SUCCESS, "rc was %d", rc);
376 if (rc != MQTT::SUCCESS)
377 goto exit;
378
379 rc = client.connect(data);
380 assert("Good rc from connect", rc == MQTT::SUCCESS, "rc was %d", rc);
381 if (rc != MQTT::SUCCESS)
382 goto exit;
383
384 rc = client.subscribe(test_topic, subsqos, messageArrived);
385 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
386
387 test1_sendAndReceive(client, 0, test_topic);
388 test1_sendAndReceive(client, 1, test_topic);
389 test1_sendAndReceive(client, 2, test_topic);
390
391 MyLog(LOGA_DEBUG, "Stopping\n");
392
393 rc = client.unsubscribe(test_topic);
394 assert("Unsubscribe successful", rc == MQTT::SUCCESS, "rc was %d", rc);
395 rc = client.disconnect();
396 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
397 ipstack.disconnect();
398
399 /* Just to make sure we can connect again */
400 rc = ipstack.connect(options.host, options.port);
401 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
402 rc = client.connect(data);
403 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
404 rc = client.disconnect();
405 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
406 ipstack.disconnect();
407
408 exit:
409 MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.",
410 (failures == 0) ? "passed" : "failed", tests, failures);
411 write_test_result();
412 return failures;
413 }
414
415
416 /*********************************************************************
417
418 Test 2: connack return data
419
420 *********************************************************************/
test2(struct Options options)421 int test2(struct Options options)
422 {
423 MQTT::QoS subsqos = MQTT::QOS2;
424 int rc;
425 const char* test_topic = "C client test2";
426
427 fprintf(xml, "<testcase classname=\"test2\" name=\"connack return data\"");
428 global_start_time = start_clock();
429 failures = 0;
430 MyLog(LOGA_INFO, "Starting test 2 - connack return data");
431
432 IPStack ipstack = IPStack();
433 MQTT::Client<IPStack, Countdown, 1000> client = MQTT::Client<IPStack, Countdown, 1000>(ipstack);
434
435 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
436 data.willFlag = 1;
437 data.MQTTVersion = options.MQTTVersion;
438 data.clientID.cstring = (char*)"connack-return-data";
439 data.username.cstring = (char*)"testuser";
440 data.password.cstring = (char*)"testpassword";
441
442 data.keepAliveInterval = 20;
443 data.cleansession = 1;
444
445 data.will.message.cstring = (char*)"will message";
446 data.will.qos = 1;
447 data.will.retained = 0;
448 data.will.topicName.cstring = (char*)"will topic";
449
450 MyLog(LOGA_DEBUG, "Connecting");
451 rc = ipstack.connect(options.host, options.port);
452 assert("Good rc from TCP connect", rc == MQTT::SUCCESS, "rc was %d", rc);
453 if (rc != MQTT::SUCCESS)
454 goto exit;
455
456 MQTT::connackData connack;
457 rc = client.connect(data, connack);
458 assert("Good rc from connect", rc == MQTT::SUCCESS, "rc was %d", rc);
459 if (rc != MQTT::SUCCESS)
460 goto exit;
461
462 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
463 assert("Session present is 0", connack.sessionPresent == 0,
464 "sessionPresent was %d", connack.sessionPresent);
465
466 rc = client.disconnect();
467 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
468 ipstack.disconnect();
469
470 /* reconnect with cleansession false */
471 data.cleansession = 0;
472 rc = ipstack.connect(options.host, options.port);
473 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
474 rc = client.connect(data, connack);
475 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
476
477 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
478 assert("Session present is 0", connack.sessionPresent == 0,
479 "sessionPresent was %d", connack.sessionPresent);
480
481 MQTT::subackData suback;
482 rc = client.subscribe(test_topic, subsqos, messageArrived, suback);
483 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
484 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
485 "rc was %d", suback.grantedQoS);
486
487 rc = client.disconnect();
488 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
489 ipstack.disconnect();
490
491 /* reconnect with cleansession false */
492 data.cleansession = 0;
493 rc = ipstack.connect(options.host, options.port);
494 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
495 rc = client.connect(data, connack);
496 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
497
498 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
499 assert("Session present is 1", connack.sessionPresent == 1,
500 "sessionPresent was %d", connack.sessionPresent);
501
502 rc = client.disconnect();
503 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
504 ipstack.disconnect();
505
506 /* reconnect with cleansession true */
507 data.cleansession = 1;
508 rc = ipstack.connect(options.host, options.port);
509 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
510 rc = client.connect(data, connack);
511 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
512
513 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
514 assert("Session present is 0", connack.sessionPresent == 0,
515 "sessionPresent was %d", connack.sessionPresent);
516
517 rc = client.disconnect();
518 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
519 ipstack.disconnect();
520
521 exit:
522 MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.",
523 (failures == 0) ? "passed" : "failed", tests, failures);
524 write_test_result();
525 return failures;
526 }
527
528
529 /*********************************************************************
530
531 Test 3: client session state
532
533 *********************************************************************/
534 static volatile MQTT::MessageData* test2_message_data = NULL;
535
messageArrived2(MQTT::MessageData & md)536 void messageArrived2(MQTT::MessageData& md)
537 {
538 test2_message_data = &md;
539 MQTT::Message &m = md.message;
540
541 assert("Good message lengths", pubmsg.payloadlen == m.payloadlen,
542 "payloadlen was %d", m.payloadlen);
543
544 if (pubmsg.payloadlen == m.payloadlen)
545 assert("Good message contents", memcmp(m.payload, pubmsg.payload, m.payloadlen) == 0,
546 "payload was %s", m.payload);
547 }
548
549
check_subs_exist(MQTT::Client<IPStack,Countdown,1000> & client,const char * test_topic,int which)550 int check_subs_exist(MQTT::Client<IPStack, Countdown, 1000>& client, const char* test_topic, int which)
551 {
552 int rc = MQTT::FAILURE;
553 int wait_seconds = 0;
554
555 memset(&pubmsg, '\0', sizeof(pubmsg));
556 pubmsg.payload = (void*)"a much longer message that we can shorten to the extent that we need to payload up to 11";
557 pubmsg.payloadlen = 11;
558 pubmsg.qos = MQTT::QOS2;
559 pubmsg.retained = false;
560 pubmsg.dup = false;
561
562 test1_message_data = test2_message_data = NULL;
563 rc = client.publish(test_topic, pubmsg);
564 assert("Good rc from publish", rc == MQTT::SUCCESS, "rc was %d", rc);
565
566 /* wait for the message to be received */
567 wait_seconds = 10;
568 while (wait_seconds-- > 0)
569 {
570 client.yield(100);
571 }
572
573 rc = (((which == 1 || which == 3) && test1_message_data) ||
574 (which == 2 && test1_message_data == NULL)) ? MQTT::SUCCESS : MQTT::FAILURE;
575 assert("test1 subscription", rc == MQTT::SUCCESS, "test1_message_data %p\n",
576 test1_message_data);
577 rc = (((which == 2 || which == 3) && test2_message_data) ||
578 (which == 1 && test2_message_data == NULL)) ? MQTT::SUCCESS : MQTT::FAILURE;
579 assert("test2 subscription", rc == MQTT::SUCCESS, "test2_message_data %p\n",
580 test2_message_data);
581 return rc;
582 }
583
584
test3(struct Options options)585 int test3(struct Options options)
586 {
587 MQTT::QoS subsqos = MQTT::QOS2;
588 int rc;
589 const char* test_topic = "C client test3";
590 int wait_seconds = 0;
591
592 fprintf(xml, "<testcase classname=\"test3\" name=\"session state\"");
593 global_start_time = start_clock();
594 failures = 0;
595 MyLog(LOGA_INFO, "Starting test 3 - session state");
596
597 IPStack ipstack = IPStack();
598 MQTT::Client<IPStack, Countdown, 1000> client = MQTT::Client<IPStack, Countdown, 1000>(ipstack);
599
600 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
601 data.willFlag = 1;
602 data.MQTTVersion = options.MQTTVersion;
603 data.clientID.cstring = (char*)"connack-return-data";
604 data.username.cstring = (char*)"testuser";
605 data.password.cstring = (char*)"testpassword";
606
607 data.keepAliveInterval = 10;
608 data.cleansession = 1;
609
610 data.will.message.cstring = (char*)"will message";
611 data.will.qos = 1;
612 data.will.retained = 0;
613 data.will.topicName.cstring = (char*)"will topic";
614
615 assert("Good rc in connack", client.isConnected() == false,
616 "isconnected was %d", client.isConnected());
617
618 MyLog(LOGA_DEBUG, "Connecting");
619 rc = ipstack.connect(options.host, options.port);
620 assert("Good rc from TCP connect", rc == MQTT::SUCCESS, "rc was %d", rc);
621 if (rc != MQTT::SUCCESS)
622 goto exit;
623
624 MQTT::connackData connack;
625 rc = client.connect(data, connack);
626 assert("Good rc from connect", rc == MQTT::SUCCESS, "rc was %d", rc);
627 if (rc != MQTT::SUCCESS)
628 goto exit;
629
630 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
631 assert("Session present is 0", connack.sessionPresent == 0,
632 "sessionPresent was %d", connack.sessionPresent);
633
634 assert("Good rc in connack", client.isConnected() == true,
635 "isconnected was %d", client.isConnected());
636
637 rc = client.disconnect();
638 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
639 ipstack.disconnect();
640
641 /* reconnect with cleansession false */
642 data.cleansession = 0;
643 rc = ipstack.connect(options.proxy_host, options.proxy_port);
644 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
645 rc = client.connect(data, connack);
646 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
647
648 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
649 assert("Session present is 0", connack.sessionPresent == 0,
650 "sessionPresent was %d", connack.sessionPresent);
651
652 MQTT::subackData suback;
653 rc = client.subscribe(test_topic, subsqos, messageArrived, suback);
654 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
655 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
656 "rc was %d", suback.grantedQoS);
657
658 check_subs_exist(client, test_topic, 1);
659
660 rc = client.subscribe(test_topic, subsqos, messageArrived2, suback);
661 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
662 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
663 "rc was %d", suback.grantedQoS);
664
665 check_subs_exist(client, test_topic, 2);
666
667 rc = client.disconnect();
668 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
669 ipstack.disconnect();
670
671 /* reconnect with cleansession false */
672 data.cleansession = 0;
673 rc = ipstack.connect(options.proxy_host, options.proxy_port);
674 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
675 rc = client.connect(data, connack);
676 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
677
678 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
679 assert("Session present is 1", connack.sessionPresent == 1,
680 "sessionPresent was %d", connack.sessionPresent);
681
682 check_subs_exist(client, test_topic, 2);
683
684 rc = client.subscribe(test_topic, subsqos, messageArrived, suback);
685 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
686 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
687 "rc was %d", suback.grantedQoS);
688
689 check_subs_exist(client, test_topic, 1);
690
691 // cause a connection FAILURE
692 memset(&pubmsg, '\0', sizeof(pubmsg));
693 pubmsg.payload = (void*)"TERMINATE";
694 pubmsg.payloadlen = strlen((char*)pubmsg.payload);
695 pubmsg.qos = MQTT::QOS0;
696 pubmsg.retained = false;
697 pubmsg.dup = false;
698 rc = client.publish("MQTTSAS topic", pubmsg);
699 assert("Good rc from publish", rc == MQTT::SUCCESS, "rc was %d", rc);
700
701 // wait for failure to be noticed by keepalive
702 wait_seconds = 20;
703 while (client.isConnected() && (wait_seconds-- > 0))
704 {
705 client.yield(1000);
706 }
707 assert("Disconnected", !client.isConnected(), "isConnected was %d",
708 client.isConnected());
709 ipstack.disconnect();
710
711 /* reconnect with cleansession false */
712 data.cleansession = 0;
713 rc = ipstack.connect(options.host, options.port);
714 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
715 rc = client.connect(data, connack);
716 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
717
718 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
719 assert("Session present is 1", connack.sessionPresent == 1,
720 "sessionPresent was %d", connack.sessionPresent);
721
722 check_subs_exist(client, test_topic, 1);
723
724 rc = client.subscribe(test_topic, subsqos, messageArrived2, suback);
725 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
726 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
727 "rc was %d", suback.grantedQoS);
728
729 check_subs_exist(client, test_topic, 2);
730
731 rc = client.disconnect();
732 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
733 ipstack.disconnect();
734
735 /* reconnect with cleansession true to clean up both server and client state */
736 data.cleansession = 1;
737 rc = ipstack.connect(options.host, options.port);
738 assert("TCP connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
739 rc = client.connect(data, connack);
740 assert("Connect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
741
742 assert("Good rc in connack", connack.rc == 0, "rc was %d", connack.rc);
743 assert("Session present is 0", connack.sessionPresent == 0,
744 "sessionPresent was %d", connack.sessionPresent);
745
746 rc = client.subscribe(test_topic, subsqos, messageArrived2, suback);
747 assert("Good rc from subscribe", rc == MQTT::SUCCESS, "rc was %d", rc);
748 assert("Granted QoS rc from subscribe", suback.grantedQoS == MQTT::QOS2,
749 "rc was %d", suback.grantedQoS);
750
751 check_subs_exist(client, test_topic, 2);
752
753 rc = client.disconnect();
754 assert("Disconnect successful", rc == MQTT::SUCCESS, "rc was %d", rc);
755 ipstack.disconnect();
756
757 exit:
758 MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.",
759 (failures == 0) ? "passed" : "failed", tests, failures);
760 write_test_result();
761 return failures;
762 }
763
764
765 #if 0
766 /*********************************************************************
767
768 Test 4: connectionLost and will message
769
770 *********************************************************************/
771 MQTTClient test6_c1, test6_c2;
772 volatile int test6_will_message_arrived = 0;
773 volatile int test6_connection_lost_called = 0;
774
775 void test6_connectionLost(void* context, char* cause)
776 {
777 MQTTClient c = (MQTTClient)context;
778 printf("%s -> Callback: connection lost\n", (c == test6_c1) ? "Client-1" : "Client-2");
779 test6_connection_lost_called = 1;
780 }
781
782 void test6_deliveryComplete(void* context, MQTTClient_deliveryToken token)
783 {
784 printf("Client-2 -> Callback: publish complete for token %d\n", token);
785 }
786
787 char* test6_will_topic = "C Test 2: will topic";
788 char* test6_will_message = "will message from Client-1";
789
790 int test6_messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* m)
791 {
792 MQTTClient c = (MQTTClient)context;
793 printf("%s -> Callback: message received on topic '%s' is '%.*s'.\n",
794 (c == test6_c1) ? "Client-1" : "Client-2", topicName, m->payloadlen, (char*)(m->payload));
795 if (c == test6_c2 && strcmp(topicName, test6_will_topic) == 0 && memcmp(m->payload, test6_will_message, m->payloadlen) == 0)
796 test6_will_message_arrived = 1;
797 MQTTClient_free(topicName);
798 MQTTClient_freeMessage(&m);
799 return 1;
800 }
801
802
803 int test6(struct Options options)
804 {
805 char* testname = "test6";
806 MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
807 MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer;
808 MQTTClient_connectOptions opts2 = MQTTClient_connectOptions_initializer;
809 int rc, count;
810 char* mqttsas_topic = "MQTTSAS topic";
811
812 failures = 0;
813 MyLog(LOGA_INFO, "Starting test 6 - connectionLost and will messages");
814 fprintf(xml, "<testcase classname=\"test1\" name=\"connectionLost and will messages\"");
815 global_start_time = start_clock();
816
817 opts.keepAliveInterval = 2;
818 opts.cleansession = 1;
819 opts.MQTTVersion = MQTTVERSION_3_1_1;
820 opts.will = &wopts;
821 opts.will->message = test6_will_message;
822 opts.will->qos = 1;
823 opts.will->retained = 0;
824 opts.will->topicName = test6_will_topic;
825 if (options.haconnections != NULL)
826 {
827 opts.serverURIs = options.haconnections;
828 opts.serverURIcount = options.hacount;
829 }
830
831 /* Client-1 with Will options */
832 rc = MQTTClient_create(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
833 assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
834 if (rc != MQTTCLIENT_SUCCESS)
835 goto exit;
836
837 rc = MQTTClient_setCallbacks(test6_c1, (void*)test6_c1, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
838 assert("good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
839 if (rc != MQTTCLIENT_SUCCESS)
840 goto exit;
841
842 /* Connect to the broker */
843 rc = MQTTClient_connect(test6_c1, &opts);
844 assert("good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
845 if (rc != MQTTCLIENT_SUCCESS)
846 goto exit;
847
848 /* Client - 2 (multi-threaded) */
849 rc = MQTTClient_create(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
850 assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
851
852 /* Set the callback functions for the client */
853 rc = MQTTClient_setCallbacks(test6_c2, (void*)test6_c2, test6_connectionLost, test6_messageArrived, test6_deliveryComplete);
854 assert("good rc from setCallbacks", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
855
856 /* Connect to the broker */
857 opts2.keepAliveInterval = 20;
858 opts2.cleansession = 1;
859 MyLog(LOGA_INFO, "Connecting Client_2 ...");
860 rc = MQTTClient_connect(test6_c2, &opts2);
861 assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
862
863 rc = MQTTClient_subscribe(test6_c2, test6_will_topic, 2);
864 assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
865
866 /* now send the command which will break the connection and cause the will message to be sent */
867 rc = MQTTClient_publish(test6_c1, mqttsas_topic, (int)strlen("TERMINATE"), "TERMINATE", 0, 0, NULL);
868 assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
869
870 MyLog(LOGA_INFO, "Waiting to receive the will message");
871 count = 0;
872 while (++count < 40)
873 {
874 #if defined(WIN32)
875 Sleep(1000L);
876 #else
877 sleep(1);
878 #endif
879 if (test6_will_message_arrived == 1 && test6_connection_lost_called == 1)
880 break;
881 }
882 assert("will message arrived", test6_will_message_arrived == 1,
883 "will_message_arrived was %d\n", test6_will_message_arrived);
884 assert("connection lost called", test6_connection_lost_called == 1,
885 "connection_lost_called %d\n", test6_connection_lost_called);
886
887 rc = MQTTClient_unsubscribe(test6_c2, test6_will_topic);
888 assert("Good rc from unsubscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
889
890 rc = MQTTClient_isConnected(test6_c2);
891 assert("Client-2 still connected", rc == 1, "isconnected is %d", rc);
892
893 rc = MQTTClient_isConnected(test6_c1);
894 assert("Client-1 not connected", rc == 0, "isconnected is %d", rc);
895
896 rc = MQTTClient_disconnect(test6_c2, 100L);
897 assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
898
899 MQTTClient_destroy(&test6_c1);
900 MQTTClient_destroy(&test6_c2);
901
902 exit:
903 MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.\n",
904 (failures == 0) ? "passed" : "failed", testname, tests, failures);
905 write_test_result();
906 return failures;
907 }
908 #endif
909
main(int argc,char ** argv)910 int main(int argc, char** argv)
911 {
912 int rc = 0;
913 int (*tests[])(Options) = {NULL, test1, test2, test3, /*test4, test5, test6, test6a*/};
914 int i;
915
916 xml = fopen("TEST-test1.xml", "w");
917 fprintf(xml, "<testsuite name=\"test1\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1));
918
919 //setenv("MQTT_C_CLIENT_TRACE", "ON", 1);
920 //setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0);
921
922 getopts(argc, argv);
923
924 for (i = 0; i < options.iterations; ++i)
925 {
926 if (options.test_no == 0)
927 { /* run all the tests */
928 for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
929 rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */
930 }
931 else
932 rc = tests[options.test_no](options); /* run just the selected test */
933 }
934
935 if (rc == 0)
936 MyLog(LOGA_INFO, "verdict pass");
937 else
938 MyLog(LOGA_INFO, "verdict fail");
939
940 fprintf(xml, "</testsuite>\n");
941 fclose(xml);
942 return rc;
943 }
944