A simplistic router for stasis_message's.
[asterisk/asterisk.git] / tests / test_stasis.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 /*!
20  * \file \brief Test Stasis message bus.
21  *
22  * \author\verbatim David M. Lee, II <dlee@digium.com> \endverbatim
23  *
24  * \ingroup tests
25  */
26
27 /*** MODULEINFO
28         <depend>TEST_FRAMEWORK</depend>
29         <support_level>core</support_level>
30  ***/
31
32 #include "asterisk.h"
33
34 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
35
36 #include "asterisk/astobj2.h"
37 #include "asterisk/module.h"
38 #include "asterisk/stasis.h"
39 #include "asterisk/stasis_message_router.h"
40 #include "asterisk/test.h"
41
42 static const char *test_category = "/stasis/core/";
43
44 AST_TEST_DEFINE(message_type)
45 {
46         RAII_VAR(struct stasis_message_type *, uut, NULL, ao2_cleanup);
47
48         switch (cmd) {
49         case TEST_INIT:
50                 info->name = __func__;
51                 info->category = test_category;
52                 info->summary = "Test basic message_type functions";
53                 info->description = "Test basic message_type functions";
54                 return AST_TEST_NOT_RUN;
55         case TEST_EXECUTE:
56                 break;
57         }
58
59         ast_test_validate(test, NULL == stasis_message_type_create(NULL));
60         uut = stasis_message_type_create("SomeMessage");
61         ast_test_validate(test, 0 == strcmp(stasis_message_type_name(uut), "SomeMessage"));
62
63         return AST_TEST_PASS;
64 }
65
66 AST_TEST_DEFINE(message)
67 {
68         RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
69         RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
70         RAII_VAR(char *, data, NULL, ao2_cleanup);
71         char *expected = "SomeData";
72         struct timeval expected_timestamp;
73         struct timeval time_diff;
74
75         switch (cmd) {
76         case TEST_INIT:
77                 info->name = __func__;
78                 info->category = test_category;
79                 info->summary = "Test basic message functions";
80                 info->description = "Test basic message functions";
81                 return AST_TEST_NOT_RUN;
82         case TEST_EXECUTE:
83                 break;
84         }
85
86
87         type = stasis_message_type_create("SomeMessage");
88
89         ast_test_validate(test, NULL == stasis_message_create(NULL, NULL));
90         ast_test_validate(test, NULL == stasis_message_create(type, NULL));
91
92         data = ao2_alloc(strlen(expected) + 1, NULL);
93         strcpy(data, expected);
94         expected_timestamp = ast_tvnow();
95         uut = stasis_message_create(type, data);
96
97         ast_test_validate(test, NULL != uut);
98         ast_test_validate(test, type == stasis_message_type(uut));
99         ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut)));
100         ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut has ref to data */
101
102         time_diff = ast_tvsub(*stasis_message_timestamp(uut), expected_timestamp);
103         /* 10ms is certainly long enough for the two calls to complete */
104         ast_test_validate(test, time_diff.tv_sec == 0);
105         ast_test_validate(test, time_diff.tv_usec < 10000);
106
107         ao2_ref(uut, -1);
108         uut = NULL;
109         ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut unreffed data */
110
111         return AST_TEST_PASS;
112 }
113
114 struct consumer {
115         ast_mutex_t lock;
116         ast_cond_t out;
117         struct stasis_message **messages_rxed;
118         size_t messages_rxed_len;
119         int ignore_subscriptions;
120         int complete;
121 };
122
123 static void consumer_dtor(void *obj) {
124         struct consumer *consumer = obj;
125
126         ast_mutex_destroy(&consumer->lock);
127         ast_cond_destroy(&consumer->out);
128
129         while (consumer->messages_rxed_len > 0) {
130                 ao2_cleanup(consumer->messages_rxed[--consumer->messages_rxed_len]);
131         }
132         ast_free(consumer->messages_rxed);
133         consumer->messages_rxed = NULL;
134 }
135
136 static struct consumer *consumer_create(int ignore_subscriptions) {
137         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
138
139         consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
140
141         if (!consumer) {
142                 return NULL;
143         }
144
145         consumer->ignore_subscriptions = ignore_subscriptions;
146         consumer->messages_rxed = ast_malloc(0);
147         if (!consumer->messages_rxed) {
148                 return NULL;
149         }
150
151         ast_mutex_init(&consumer->lock);
152         ast_cond_init(&consumer->out, NULL);
153
154         ao2_ref(consumer, +1);
155         return consumer;
156 }
157
158 static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
159 {
160         struct consumer *consumer = data;
161         RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
162         SCOPED_MUTEX(lock, &consumer->lock);
163
164         if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change()) {
165
166                 ++consumer->messages_rxed_len;
167                 consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
168                 ast_assert(consumer->messages_rxed != NULL);
169                 consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
170                 ao2_ref(message, +1);
171         }
172
173         if (stasis_subscription_final_message(sub, message)) {
174                 consumer->complete = 1;
175                 consumer_needs_cleanup = consumer;
176         }
177
178         ast_cond_signal(&consumer->out);
179 }
180
181 static int consumer_wait_for(struct consumer *consumer, size_t expected_len)
182 {
183         struct timeval start = ast_tvnow();
184         struct timespec end = {
185                 .tv_sec = start.tv_sec + 30,
186                 .tv_nsec = start.tv_usec * 1000
187         };
188
189         SCOPED_MUTEX(lock, &consumer->lock);
190
191         while (consumer->messages_rxed_len < expected_len) {
192                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
193                 if (r == ETIMEDOUT) {
194                         break;
195                 }
196                 ast_assert(r == 0); /* Not expecting any othet types of errors */
197         }
198         return consumer->messages_rxed_len;
199 }
200
201 static int consumer_wait_for_completion(struct consumer *consumer)
202 {
203         struct timeval start = ast_tvnow();
204         struct timespec end = {
205                 .tv_sec = start.tv_sec + 30,
206                 .tv_nsec = start.tv_usec * 1000
207         };
208
209         SCOPED_MUTEX(lock, &consumer->lock);
210
211         while (!consumer->complete) {
212                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
213                 if (r == ETIMEDOUT) {
214                         break;
215                 }
216                 ast_assert(r == 0); /* Not expecting any othet types of errors */
217         }
218         return consumer->complete;
219 }
220
221 static int consumer_should_stay(struct consumer *consumer, size_t expected_len)
222 {
223         struct timeval start = ast_tvnow();
224         struct timeval diff = {
225                 .tv_sec = 0,
226                 .tv_usec = 100000 /* wait for 100ms */
227         };
228         struct timeval end_tv = ast_tvadd(start, diff);
229         struct timespec end = {
230                 .tv_sec = end_tv.tv_sec,
231                 .tv_nsec = end_tv.tv_usec * 1000
232         };
233
234         SCOPED_MUTEX(lock, &consumer->lock);
235
236         while (consumer->messages_rxed_len == expected_len) {
237                 int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
238                 if (r == ETIMEDOUT) {
239                         break;
240                 }
241                 ast_assert(r == 0); /* Not expecting any othet types of errors */
242         }
243         return consumer->messages_rxed_len;
244 }
245
246 AST_TEST_DEFINE(subscription_messages)
247 {
248         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
249         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
250         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
251         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
252         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
253         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
254         RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
255         int complete;
256         struct stasis_subscription_change *change;
257
258         switch (cmd) {
259         case TEST_INIT:
260                 info->name = __func__;
261                 info->category = test_category;
262                 info->summary = "Test subscribe/unsubscribe messages";
263                 info->description = "Test subscribe/unsubscribe messages";
264                 return AST_TEST_NOT_RUN;
265         case TEST_EXECUTE:
266                 break;
267         }
268
269         topic = stasis_topic_create("TestTopic");
270         ast_test_validate(test, NULL != topic);
271
272         consumer = consumer_create(0);
273         ast_test_validate(test, NULL != consumer);
274
275         uut = stasis_subscribe(topic, consumer_exec, consumer);
276         ast_test_validate(test, NULL != uut);
277         ao2_ref(consumer, +1);
278         expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
279
280         uut = stasis_unsubscribe(uut);
281         complete = consumer_wait_for_completion(consumer);
282         ast_test_validate(test, 1 == complete);
283
284         ast_test_validate(test, 2 == consumer->messages_rxed_len);
285         ast_test_validate(test, stasis_subscription_change() == stasis_message_type(consumer->messages_rxed[0]));
286         ast_test_validate(test, stasis_subscription_change() == stasis_message_type(consumer->messages_rxed[1]));
287
288         change = stasis_message_data(consumer->messages_rxed[0]);
289         ast_test_validate(test, topic == change->topic);
290         ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
291         ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
292
293         change = stasis_message_data(consumer->messages_rxed[1]);
294         ast_test_validate(test, topic == change->topic);
295         ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
296         ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
297
298         return AST_TEST_PASS;
299 }
300
301 AST_TEST_DEFINE(publish)
302 {
303         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
304         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
305         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
306         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
307         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
308         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
309         int actual_len;
310         const char *actual;
311
312         switch (cmd) {
313         case TEST_INIT:
314                 info->name = __func__;
315                 info->category = test_category;
316                 info->summary = "Test simple subscriptions";
317                 info->description = "Test simple subscriptions";
318                 return AST_TEST_NOT_RUN;
319         case TEST_EXECUTE:
320                 break;
321         }
322
323         topic = stasis_topic_create("TestTopic");
324         ast_test_validate(test, NULL != topic);
325
326         consumer = consumer_create(1);
327         ast_test_validate(test, NULL != consumer);
328
329         uut = stasis_subscribe(topic, consumer_exec, consumer);
330         ast_test_validate(test, NULL != uut);
331         ao2_ref(consumer, +1);
332
333         test_data = ao2_alloc(1, NULL);
334         ast_test_validate(test, NULL != test_data);
335         test_message_type = stasis_message_type_create("TestMessage");
336         test_message = stasis_message_create(test_message_type, test_data);
337
338         stasis_publish(topic, test_message);
339
340         actual_len = consumer_wait_for(consumer, 1);
341         ast_test_validate(test, 1 == actual_len);
342         actual = stasis_message_data(consumer->messages_rxed[0]);
343         ast_test_validate(test, test_data == actual);
344
345         return AST_TEST_PASS;
346 }
347
348 AST_TEST_DEFINE(unsubscribe_stops_messages)
349 {
350         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
351         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
352         RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
353         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
354         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
355         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
356         int actual_len;
357
358         switch (cmd) {
359         case TEST_INIT:
360                 info->name = __func__;
361                 info->category = test_category;
362                 info->summary = "Test simple subscriptions";
363                 info->description = "Test simple subscriptions";
364                 return AST_TEST_NOT_RUN;
365         case TEST_EXECUTE:
366                 break;
367         }
368
369         topic = stasis_topic_create("TestTopic");
370         ast_test_validate(test, NULL != topic);
371
372         consumer = consumer_create(1);
373         ast_test_validate(test, NULL != consumer);
374
375         uut = stasis_subscribe(topic, consumer_exec, consumer);
376         ast_test_validate(test, NULL != uut);
377         ao2_ref(consumer, +1);
378
379         uut = stasis_unsubscribe(uut);
380
381         test_data = ao2_alloc(1, NULL);
382         ast_test_validate(test, NULL != test_data);
383         test_message_type = stasis_message_type_create("TestMessage");
384         test_message = stasis_message_create(test_message_type, test_data);
385
386         stasis_publish(topic, test_message);
387
388         actual_len = consumer_should_stay(consumer, 0);
389         ast_test_validate(test, 0 == actual_len);
390
391         return AST_TEST_PASS;
392 }
393
394
395 AST_TEST_DEFINE(forward)
396 {
397         RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
398         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
399
400         RAII_VAR(struct consumer *, parent_consumer, NULL, ao2_cleanup);
401         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
402
403         RAII_VAR(struct stasis_subscription *, forward_sub, NULL, stasis_unsubscribe);
404         RAII_VAR(struct stasis_subscription *, parent_sub, NULL, stasis_unsubscribe);
405         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
406
407         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
408         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
409         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
410         int actual_len;
411
412         switch (cmd) {
413         case TEST_INIT:
414                 info->name = __func__;
415                 info->category = test_category;
416                 info->summary = "Test sending events to a parent topic";
417                 info->description = "Test sending events to a parent topic.\n"
418                         "This test creates three topics (one parent, two children)\n"
419                         "and publishes a message to one child, and verifies it's\n"
420                         "only seen by that child and the parent";
421                 return AST_TEST_NOT_RUN;
422         case TEST_EXECUTE:
423                 break;
424         }
425
426         parent_topic = stasis_topic_create("ParentTestTopic");
427         ast_test_validate(test, NULL != parent_topic);
428         topic = stasis_topic_create("TestTopic");
429         ast_test_validate(test, NULL != topic);
430
431         forward_sub = stasis_forward_all(topic, parent_topic);
432         ast_test_validate(test, NULL != forward_sub);
433
434         parent_consumer = consumer_create(1);
435         ast_test_validate(test, NULL != parent_consumer);
436         consumer = consumer_create(1);
437         ast_test_validate(test, NULL != consumer);
438
439         parent_sub = stasis_subscribe(parent_topic, consumer_exec, parent_consumer);
440         ast_test_validate(test, NULL != parent_sub);
441         ao2_ref(parent_consumer, +1);
442         sub = stasis_subscribe(topic, consumer_exec, consumer);
443         ast_test_validate(test, NULL != sub);
444         ao2_ref(consumer, +1);
445
446         test_data = ao2_alloc(1, NULL);
447         ast_test_validate(test, NULL != test_data);
448         test_message_type = stasis_message_type_create("TestMessage");
449         test_message = stasis_message_create(test_message_type, test_data);
450
451         stasis_publish(topic, test_message);
452
453         actual_len = consumer_wait_for(consumer, 1);
454         ast_test_validate(test, 1 == actual_len);
455         actual_len = consumer_wait_for(parent_consumer, 1);
456         ast_test_validate(test, 1 == actual_len);
457
458         return AST_TEST_PASS;
459 }
460
461 struct cache_test_data {
462         char *id;
463         char *value;
464 };
465
466 static void cache_test_data_dtor(void *obj)
467 {
468         struct cache_test_data *data = obj;
469         ast_free(data->id);
470         ast_free(data->value);
471 }
472
473 static struct stasis_message *cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
474 {
475         RAII_VAR(struct cache_test_data *, data, NULL, ao2_cleanup);
476
477         data = ao2_alloc(sizeof(*data), cache_test_data_dtor);
478         if (data == NULL) {
479                 return NULL;
480         }
481
482         ast_assert(name != NULL);
483         ast_assert(value != NULL);
484
485         data->id = ast_strdup(name);
486         data->value = ast_strdup(value);
487         if (!data->id || !data->value) {
488                 return NULL;
489         }
490
491         return stasis_message_create(type, data);
492 }
493
494 static const char *cache_test_data_id(struct stasis_message *message) {
495         struct cache_test_data *cachable = stasis_message_data(message);
496
497         if (0 != strcmp("Cacheable", stasis_message_type_name(stasis_message_type(message)))) {
498                 return NULL;
499         }
500         return cachable->id;
501 }
502
503 AST_TEST_DEFINE(cache_passthrough)
504 {
505         RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
506         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
507         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
508         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
509         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
510         RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
511         int actual_len;
512         struct stasis_message_type *actual_type;
513
514         switch (cmd) {
515         case TEST_INIT:
516                 info->name = __func__;
517                 info->category = test_category;
518                 info->summary = "Test passing messages through cache topic unscathed.";
519                 info->description = "Test passing messages through cache topic unscathed.";
520                 return AST_TEST_NOT_RUN;
521         case TEST_EXECUTE:
522                 break;
523         }
524
525         non_cache_type = stasis_message_type_create("NonCacheable");
526         ast_test_validate(test, NULL != non_cache_type);
527         topic = stasis_topic_create("SomeTopic");
528         ast_test_validate(test, NULL != topic);
529         caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
530         ast_test_validate(test, NULL != caching_topic);
531         consumer = consumer_create(1);
532         ast_test_validate(test, NULL != consumer);
533         sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
534         ast_test_validate(test, NULL != sub);
535         ao2_ref(consumer, +1);
536
537         test_message = cache_test_message_create(non_cache_type, "1", "1");
538         ast_test_validate(test, NULL != test_message);
539
540         stasis_publish(topic, test_message);
541
542         actual_len = consumer_wait_for(consumer, 1);
543         ast_test_validate(test, 1 == actual_len);
544
545         actual_type = stasis_message_type(consumer->messages_rxed[0]);
546         ast_test_validate(test, non_cache_type == actual_type);
547
548         ast_test_validate(test, test_message == consumer->messages_rxed[0]);
549
550         return AST_TEST_PASS;
551 }
552
553 AST_TEST_DEFINE(cache)
554 {
555         RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
556         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
557         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
558         RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
559         RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
560         RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
561         RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
562         RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
563         RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
564         int actual_len;
565         struct stasis_cache_update *actual_update;
566         struct ao2_container *cache_dump;
567
568         switch (cmd) {
569         case TEST_INIT:
570                 info->name = __func__;
571                 info->category = test_category;
572                 info->summary = "Test passing messages through cache topic unscathed.";
573                 info->description = "Test passing messages through cache topic unscathed.";
574                 return AST_TEST_NOT_RUN;
575         case TEST_EXECUTE:
576                 break;
577         }
578
579         cache_type = stasis_message_type_create("Cacheable");
580         ast_test_validate(test, NULL != cache_type);
581         topic = stasis_topic_create("SomeTopic");
582         ast_test_validate(test, NULL != topic);
583         caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
584         ast_test_validate(test, NULL != caching_topic);
585         consumer = consumer_create(1);
586         ast_test_validate(test, NULL != consumer);
587         sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
588         ast_test_validate(test, NULL != sub);
589         ao2_ref(consumer, +1);
590
591         test_message1_1 = cache_test_message_create(cache_type, "1", "1");
592         ast_test_validate(test, NULL != test_message1_1);
593         test_message2_1 = cache_test_message_create(cache_type, "2", "1");
594         ast_test_validate(test, NULL != test_message2_1);
595
596         /* Post a couple of snapshots */
597         stasis_publish(topic, test_message1_1);
598         stasis_publish(topic, test_message2_1);
599         actual_len = consumer_wait_for(consumer, 2);
600         ast_test_validate(test, 2 == actual_len);
601
602         /* Dump the cache to ensure that it has the correct number of items in it */
603         cache_dump = stasis_cache_dump(caching_topic, NULL);
604         ast_test_validate(test, 2 == ao2_container_count(cache_dump));
605         ao2_ref(cache_dump, -1);
606         cache_dump = NULL;
607
608         /* Check for new snapshot messages */
609         ast_test_validate(test, stasis_cache_update() == stasis_message_type(consumer->messages_rxed[0]));
610         actual_update = stasis_message_data(consumer->messages_rxed[0]);
611         ast_test_validate(test, topic == actual_update->topic);
612         ast_test_validate(test, NULL == actual_update->old_snapshot);
613         ast_test_validate(test, test_message1_1 == actual_update->new_snapshot);
614         ast_test_validate(test, test_message1_1 == stasis_cache_get(caching_topic, cache_type, "1"));
615         /* stasis_cache_get returned a ref, so unref test_message1_1 */
616         ao2_ref(test_message1_1, -1);
617
618         ast_test_validate(test, stasis_cache_update() == stasis_message_type(consumer->messages_rxed[1]));
619         actual_update = stasis_message_data(consumer->messages_rxed[1]);
620         ast_test_validate(test, topic == actual_update->topic);
621         ast_test_validate(test, NULL == actual_update->old_snapshot);
622         ast_test_validate(test, test_message2_1 == actual_update->new_snapshot);
623         ast_test_validate(test, test_message2_1 == stasis_cache_get(caching_topic, cache_type, "2"));
624         /* stasis_cache_get returned a ref, so unref test_message2_1 */
625         ao2_ref(test_message2_1, -1);
626
627         /* Update snapshot 2 */
628         test_message2_2 = cache_test_message_create(cache_type, "2", "2");
629         ast_test_validate(test, NULL != test_message2_2);
630         stasis_publish(topic, test_message2_2);
631
632         actual_len = consumer_wait_for(consumer, 3);
633         ast_test_validate(test, 3 == actual_len);
634
635         actual_update = stasis_message_data(consumer->messages_rxed[2]);
636         ast_test_validate(test, topic == actual_update->topic);
637         ast_test_validate(test, test_message2_1 == actual_update->old_snapshot);
638         ast_test_validate(test, test_message2_2 == actual_update->new_snapshot);
639         ast_test_validate(test, test_message2_2 == stasis_cache_get(caching_topic, cache_type, "2"));
640         /* stasis_cache_get returned a ref, so unref test_message2_2 */
641         ao2_ref(test_message2_2, -1);
642
643         /* Dump the cache to ensure that it has the correct number of items in it */
644         cache_dump = stasis_cache_dump(caching_topic, NULL);
645         ast_test_validate(test, 2 == ao2_container_count(cache_dump));
646         ao2_ref(cache_dump, -1);
647         cache_dump = NULL;
648
649         /* Clear snapshot 1 */
650         test_message1_clear = stasis_cache_clear_create(cache_type, "1");
651         ast_test_validate(test, NULL != test_message1_clear);
652         stasis_publish(topic, test_message1_clear);
653
654         actual_len = consumer_wait_for(consumer, 4);
655         ast_test_validate(test, 4 == actual_len);
656
657         actual_update = stasis_message_data(consumer->messages_rxed[3]);
658         ast_test_validate(test, topic == actual_update->topic);
659         ast_test_validate(test, test_message1_1 == actual_update->old_snapshot);
660         ast_test_validate(test, NULL == actual_update->new_snapshot);
661         ast_test_validate(test, NULL == stasis_cache_get(caching_topic, cache_type, "1"));
662
663         /* Dump the cache to ensure that it has the correct number of items in it */
664         cache_dump = stasis_cache_dump(caching_topic, NULL);
665         ast_test_validate(test, 1 == ao2_container_count(cache_dump));
666         ao2_ref(cache_dump, -1);
667         cache_dump = NULL;
668
669         /* Dump the cache to ensure that it has no subscription change items in it since those aren't cached */
670         cache_dump = stasis_cache_dump(caching_topic, stasis_subscription_change());
671         ast_test_validate(test, 0 == ao2_container_count(cache_dump));
672         ao2_ref(cache_dump, -1);
673         cache_dump = NULL;
674
675         return AST_TEST_PASS;
676 }
677
678 AST_TEST_DEFINE(route_conflicts)
679 {
680         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
681         RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
682         RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
683         RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
684         RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
685         int ret;
686
687         switch (cmd) {
688         case TEST_INIT:
689                 info->name = __func__;
690                 info->category = test_category;
691                 info->summary =
692                         "Multiple routes to the same message_type should fail";
693                 info->description =
694                         "Multiple routes to the same message_type should fail";
695                 return AST_TEST_NOT_RUN;
696         case TEST_EXECUTE:
697                 break;
698         }
699
700         topic = stasis_topic_create("TestTopic");
701         ast_test_validate(test, NULL != topic);
702
703         consumer1 = consumer_create(1);
704         ast_test_validate(test, NULL != consumer1);
705         consumer2 = consumer_create(1);
706         ast_test_validate(test, NULL != consumer2);
707
708         test_message_type = stasis_message_type_create("TestMessage");
709         ast_test_validate(test, NULL != test_message_type);
710
711         uut = stasis_message_router_create(topic);
712         ast_test_validate(test, NULL != uut);
713
714         ret = stasis_message_router_add(
715                 uut, test_message_type, consumer_exec, consumer1);
716         ast_test_validate(test, 0 == ret);
717         ret = stasis_message_router_add(
718                 uut, test_message_type, consumer_exec, consumer2);
719         ast_test_validate(test, 0 != ret);
720
721         return AST_TEST_PASS;
722 }
723
724 AST_TEST_DEFINE(router)
725 {
726         RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
727         RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
728         RAII_VAR(char *, test_data, NULL, ao2_cleanup);
729         RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
730         RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
731         RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
732         RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
733         RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
734         RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
735         RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
736         RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
737         RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
738         int actual_len, ret;
739         struct stasis_message *actual;
740
741         switch (cmd) {
742         case TEST_INIT:
743                 info->name = __func__;
744                 info->category = test_category;
745                 info->summary = "Test simple message routing";
746                 info->description = "Test simple message routing";
747                 return AST_TEST_NOT_RUN;
748         case TEST_EXECUTE:
749                 break;
750         }
751
752         topic = stasis_topic_create("TestTopic");
753         ast_test_validate(test, NULL != topic);
754
755         consumer1 = consumer_create(1);
756         ast_test_validate(test, NULL != consumer1);
757         consumer2 = consumer_create(1);
758         ast_test_validate(test, NULL != consumer2);
759         consumer3 = consumer_create(1);
760         ast_test_validate(test, NULL != consumer3);
761
762         test_message_type1 = stasis_message_type_create("TestMessage1");
763         ast_test_validate(test, NULL != test_message_type1);
764         test_message_type2 = stasis_message_type_create("TestMessage2");
765         ast_test_validate(test, NULL != test_message_type2);
766         test_message_type3 = stasis_message_type_create("TestMessage3");
767         ast_test_validate(test, NULL != test_message_type3);
768
769         uut = stasis_message_router_create(topic);
770         ast_test_validate(test, NULL != uut);
771
772         ret = stasis_message_router_add(
773                 uut, test_message_type1, consumer_exec, consumer1);
774         ast_test_validate(test, 0 == ret);
775         ao2_ref(consumer1, +1);
776         ret = stasis_message_router_add(
777                 uut, test_message_type2, consumer_exec, consumer2);
778         ast_test_validate(test, 0 == ret);
779         ao2_ref(consumer2, +1);
780         ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
781         ast_test_validate(test, 0 == ret);
782         ao2_ref(consumer3, +1);
783
784         test_data = ao2_alloc(1, NULL);
785         ast_test_validate(test, NULL != test_data);
786         test_message1 = stasis_message_create(test_message_type1, test_data);
787         ast_test_validate(test, NULL != test_message1);
788         test_message2 = stasis_message_create(test_message_type2, test_data);
789         ast_test_validate(test, NULL != test_message2);
790         test_message3 = stasis_message_create(test_message_type3, test_data);
791         ast_test_validate(test, NULL != test_message3);
792
793         stasis_publish(topic, test_message1);
794         stasis_publish(topic, test_message2);
795         stasis_publish(topic, test_message3);
796
797         actual_len = consumer_wait_for(consumer1, 1);
798         ast_test_validate(test, 1 == actual_len);
799         actual_len = consumer_wait_for(consumer2, 1);
800         ast_test_validate(test, 1 == actual_len);
801         actual_len = consumer_wait_for(consumer3, 1);
802         ast_test_validate(test, 1 == actual_len);
803
804         actual = consumer1->messages_rxed[0];
805         ast_test_validate(test, test_message1 == actual);
806
807         actual = consumer2->messages_rxed[0];
808         ast_test_validate(test, test_message2 == actual);
809
810         actual = consumer3->messages_rxed[0];
811         ast_test_validate(test, test_message3 == actual);
812
813         /* consumer1 and consumer2 do not get the final message. */
814         ao2_cleanup(consumer1);
815         ao2_cleanup(consumer2);
816
817         return AST_TEST_PASS;
818 }
819
820 static int unload_module(void)
821 {
822         AST_TEST_UNREGISTER(message_type);
823         AST_TEST_UNREGISTER(message);
824         AST_TEST_UNREGISTER(subscription_messages);
825         AST_TEST_UNREGISTER(publish);
826         AST_TEST_UNREGISTER(unsubscribe_stops_messages);
827         AST_TEST_UNREGISTER(forward);
828         AST_TEST_UNREGISTER(cache_passthrough);
829         AST_TEST_UNREGISTER(cache);
830         AST_TEST_UNREGISTER(route_conflicts);
831         AST_TEST_UNREGISTER(router);
832         return 0;
833 }
834
835 static int load_module(void)
836 {
837         AST_TEST_REGISTER(message_type);
838         AST_TEST_REGISTER(message);
839         AST_TEST_REGISTER(subscription_messages);
840         AST_TEST_REGISTER(publish);
841         AST_TEST_REGISTER(unsubscribe_stops_messages);
842         AST_TEST_REGISTER(forward);
843         AST_TEST_REGISTER(cache_passthrough);
844         AST_TEST_REGISTER(cache);
845         AST_TEST_REGISTER(route_conflicts);
846         AST_TEST_REGISTER(router);
847         return AST_MODULE_LOAD_SUCCESS;
848 }
849
850 AST_MODULE_INFO(ASTERISK_GPL_KEY, 0, "Stasis testing",
851                 .load = load_module,
852                 .unload = unload_module
853         );