Asterisk - The Open Source Telephony Project GIT-master-b023714
Loading...
Searching...
No Matches
Data Structures | Macros | Functions | Variables
test_stasis.c File Reference

Test Stasis message bus. More...

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/module.h"
#include "asterisk/stasis.h"
#include "asterisk/stasis_message_router.h"
#include "asterisk/test.h"
Include dependency graph for test_stasis.c:

Go to the source code of this file.

Data Structures

struct  cache_test_data
 
struct  consumer
 
struct  cts
 
struct  test_message_types
 

Macros

#define test_category   "/stasis/core/"
 

Functions

static void __reg_module (void)
 
static void __unreg_module (void)
 
struct ast_module * AST_MODULE_SELF_SYM (void)
 
 AST_TEST_DEFINE (cache)
 
 AST_TEST_DEFINE (cache_dump)
 
 AST_TEST_DEFINE (cache_eid_aggregate)
 
 AST_TEST_DEFINE (cache_filter)
 
 AST_TEST_DEFINE (caching_dtor_order)
 
 AST_TEST_DEFINE (combo_filters)
 
 AST_TEST_DEFINE (dtor_order)
 
 AST_TEST_DEFINE (formatter_filters)
 
 AST_TEST_DEFINE (forward)
 
 AST_TEST_DEFINE (interleaving)
 
 AST_TEST_DEFINE (message)
 
 AST_TEST_DEFINE (message_type)
 
 AST_TEST_DEFINE (no_to_ami)
 
 AST_TEST_DEFINE (no_to_json)
 
 AST_TEST_DEFINE (publish)
 
 AST_TEST_DEFINE (publish_pool)
 
 AST_TEST_DEFINE (publish_sync)
 
 AST_TEST_DEFINE (router)
 
 AST_TEST_DEFINE (router_cache_updates)
 
 AST_TEST_DEFINE (router_pool)
 
 AST_TEST_DEFINE (subscription_interleaving)
 
 AST_TEST_DEFINE (subscription_messages)
 
 AST_TEST_DEFINE (subscription_pool_messages)
 
 AST_TEST_DEFINE (to_ami)
 
 AST_TEST_DEFINE (to_json)
 
 AST_TEST_DEFINE (type_filters)
 
 AST_TEST_DEFINE (unsubscribe_stops_messages)
 
static const char * cache_simple (struct stasis_message *message)
 
static struct stasis_message * cache_test_aggregate_calc_fn (struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
 
static void cache_test_aggregate_publish_fn (struct stasis_topic *topic, struct stasis_message *aggregate)
 
static void cache_test_data_dtor (void *obj)
 
static const char * cache_test_data_id (struct stasis_message *message)
 
static struct stasis_message * cache_test_message_create (struct stasis_message_type *type, const char *name, const char *value)
 
static struct stasis_message * cache_test_message_create_full (struct stasis_message_type *type, const char *name, const char *value, struct ast_eid *eid)
 
static int check_cache_aggregate (struct stasis_cache *cache, struct stasis_message_type *cache_type, const char *id, const char *value)
 
static struct consumer * consumer_create (int ignore_subscriptions)
 
static void consumer_dtor (void *obj)
 
static void consumer_exec (void *data, struct stasis_subscription *sub, struct stasis_message *message)
 
static void consumer_exec_sync (void *data, struct stasis_subscription *sub, struct stasis_message *message)
 
static int consumer_should_stay (struct consumer *consumer, size_t expected_len)
 
static int consumer_wait_for (struct consumer *consumer, size_t expected_len)
 
static int consumer_wait_for_completion (struct consumer *consumer)
 
static struct cts * create_cts (struct ast_test *test)
 
static struct test_message_types * create_message_types (struct ast_test *test)
 
static void destroy_cts (void *obj)
 
static void destroy_message_types (void *obj)
 
static void dump_consumer (struct ast_test *test, struct cts *cts)
 
static struct ast_manager_event_blob * fake_ami (struct stasis_message *message)
 
static struct ast_event * fake_event (struct stasis_message *message)
 
static struct ast_json * fake_json (struct stasis_message *message, const struct stasis_message_sanitizer *sanitize)
 
static int is_msg (struct stasis_message *msg, struct stasis_message_type *mtype, const char *data)
 
static int load_module (void)
 
static void noop (void *data, struct stasis_subscription *sub, struct stasis_message *message)
 
static const char * noop_get_id (struct stasis_message *message)
 
static int send_msg (struct ast_test *test, struct cts *cts, struct stasis_message_type *msg_type, const char *data)
 
static int unload_module (void)
 

Variables

static struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = 0 , .description = "Stasis testing" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module }
 
static const struct ast_module_info * ast_module_info = &__mod_info
 
static struct stasis_message_vtable fake_vtable
 

Detailed Description

Test Stasis message bus.

Author
David M. Lee, II <dlee@digium.com>

Definition in file test_stasis.c.

Macro Definition Documentation

◆ test_category

#define test_category   "/stasis/core/"

Definition at line 41 of file test_stasis.c.

Function Documentation

◆ __reg_module()

static void __reg_module ( void  )
static

Definition at line 2504 of file test_stasis.c.

◆ __unreg_module()

static void __unreg_module ( void  )
static

Definition at line 2504 of file test_stasis.c.

◆ AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM ( void  )

Definition at line 2504 of file test_stasis.c.

◆ AST_TEST_DEFINE() [1/27]

AST_TEST_DEFINE ( cache  )

Definition at line 1033 of file test_stasis.c.

1034{
1035 RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
1036 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1037 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
1038 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
1039 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
1040 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
1041 RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
1042 RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
1043 RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
1044 RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
1045 int actual_len;
1046 struct stasis_cache_update *actual_update;
1047
1048 switch (cmd) {
1049 case TEST_INIT:
1050 info->name = __func__;
1051 info->category = test_category;
1052 info->summary = "Test passing messages through cache topic unscathed.";
1053 info->description = "Test passing messages through cache topic unscathed.";
1054 return AST_TEST_NOT_RUN;
1055 case TEST_EXECUTE:
1056 break;
1057 }
1058
1059 ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
1060 ast_test_validate(test, NULL != cache_type);
1061 topic = stasis_topic_create("SomeTopic");
1062 ast_test_validate(test, NULL != topic);
1063 cache = stasis_cache_create(cache_test_data_id);
1064 ast_test_validate(test, NULL != cache);
1065 caching_topic = stasis_caching_topic_create(topic, cache);
1066 ast_test_validate(test, NULL != caching_topic);
1067 consumer = consumer_create(1);
1068 ast_test_validate(test, NULL != consumer);
1069 sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
1070 ast_test_validate(test, NULL != sub);
1071 ao2_ref(consumer, +1);
1072
1073 test_message1_1 = cache_test_message_create(cache_type, "1", "1");
1074 ast_test_validate(test, NULL != test_message1_1);
1075 test_message2_1 = cache_test_message_create(cache_type, "2", "1");
1076 ast_test_validate(test, NULL != test_message2_1);
1077
1078 /* Post a couple of snapshots */
1079 stasis_publish(topic, test_message1_1);
1080 stasis_publish(topic, test_message2_1);
1081 actual_len = consumer_wait_for(consumer, 2);
1082 ast_test_validate(test, 2 == actual_len);
1083
1084 /* Check for new snapshot messages */
1085 ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(consumer->messages_rxed[0]));
1086 actual_update = stasis_message_data(consumer->messages_rxed[0]);
1087 ast_test_validate(test, NULL == actual_update->old_snapshot);
1088 ast_test_validate(test, test_message1_1 == actual_update->new_snapshot);
1089 ast_test_validate(test, test_message1_1 == stasis_cache_get(cache, cache_type, "1"));
1090 /* stasis_cache_get returned a ref, so unref test_message1_1 */
1091 ao2_ref(test_message1_1, -1);
1092
1093 ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(consumer->messages_rxed[1]));
1094 actual_update = stasis_message_data(consumer->messages_rxed[1]);
1095 ast_test_validate(test, NULL == actual_update->old_snapshot);
1096 ast_test_validate(test, test_message2_1 == actual_update->new_snapshot);
1097 ast_test_validate(test, test_message2_1 == stasis_cache_get(cache, cache_type, "2"));
1098 /* stasis_cache_get returned a ref, so unref test_message2_1 */
1099 ao2_ref(test_message2_1, -1);
1100
1101 /* Update snapshot 2 */
1102 test_message2_2 = cache_test_message_create(cache_type, "2", "2");
1103 ast_test_validate(test, NULL != test_message2_2);
1104 stasis_publish(topic, test_message2_2);
1105
1106 actual_len = consumer_wait_for(consumer, 3);
1107 ast_test_validate(test, 3 == actual_len);
1108
1109 actual_update = stasis_message_data(consumer->messages_rxed[2]);
1110 ast_test_validate(test, test_message2_1 == actual_update->old_snapshot);
1111 ast_test_validate(test, test_message2_2 == actual_update->new_snapshot);
1112 ast_test_validate(test, test_message2_2 == stasis_cache_get(cache, cache_type, "2"));
1113 /* stasis_cache_get returned a ref, so unref test_message2_2 */
1114 ao2_ref(test_message2_2, -1);
1115
1116 /* Clear snapshot 1 */
1117 test_message1_clear = stasis_cache_clear_create(test_message1_1);
1118 ast_test_validate(test, NULL != test_message1_clear);
1119 stasis_publish(topic, test_message1_clear);
1120
1121 actual_len = consumer_wait_for(consumer, 4);
1122 ast_test_validate(test, 4 == actual_len);
1123
1124 actual_update = stasis_message_data(consumer->messages_rxed[3]);
1125 ast_test_validate(test, test_message1_1 == actual_update->old_snapshot);
1126 ast_test_validate(test, NULL == actual_update->new_snapshot);
1127 ast_test_validate(test, NULL == stasis_cache_get(cache, cache_type, "1"));
1128
1129 return AST_TEST_PASS;
1130}
ao2_cleanup
#define ao2_cleanup(obj)
Definition astobj2.h:1934
ao2_ref
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition astobj2.h:459
stasis_cache_update_type
struct stasis_message_type * stasis_cache_update_type(void)
Message type for cache update messages.
sip_to_pjsip.info
info(msg)
Definition sip_to_pjsip.py:1313
cache
struct ao2_container * cache
Definition pbx_realtime.c:77
sub
static struct stasis_subscription * sub
Statsd channel stats. Exmaple of how to subscribe to Stasis events.
Definition res_chan_stats.c:44
NULL
#define NULL
Definition resample.c:96
stasis_cache_clear_create
struct stasis_message * stasis_cache_clear_create(struct stasis_message *message)
A message which instructs the caching topic to remove an entry from its cache.
Definition stasis_cache.c:778
stasis_caching_get_topic
struct stasis_topic * stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
Returns the topic of cached events from a caching topics.
Definition stasis_cache.c:85
stasis_caching_unsubscribe
struct stasis_caching_topic * stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
Unsubscribes a caching topic from its upstream topic.
Definition stasis_cache.c:119
stasis_topic_create
struct stasis_topic * stasis_topic_create(const char *name)
Create a new topic.
Definition stasis.c:684
stasis_caching_topic_create
struct stasis_caching_topic * stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
Create a topic which monitors and caches messages from another topic.
Definition stasis_cache.c:948
stasis_message_data
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
Definition stasis_message.c:195
stasis_message_type_create
enum stasis_message_type_result stasis_message_type_create(const char *name, struct stasis_message_vtable *vtable, struct stasis_message_type **result)
Create a new message type.
Definition stasis_message.c:56
stasis_cache_create
struct stasis_cache * stasis_cache_create(snapshot_get_id id_fn)
Create a cache.
Definition stasis_cache.c:360
stasis_unsubscribe
struct stasis_subscription * stasis_unsubscribe(struct stasis_subscription *subscription)
Cancel a subscription.
Definition stasis.c:1038
stasis_publish
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition stasis.c:1578
stasis_cache_get
struct stasis_message * stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
Retrieve an item from the cache for the ast_eid_default entity.
Definition stasis_cache.c:686
stasis_subscribe
#define stasis_subscribe(topic, callback, data)
Definition stasis.h:649
STASIS_MESSAGE_TYPE_SUCCESS
@ STASIS_MESSAGE_TYPE_SUCCESS
Definition stasis.h:287
consumer::messages_rxed
struct stasis_message ** messages_rxed
Definition test_stasis.c:164
stasis_cache_update
Cache update message.
Definition stasis.h:965
stasis_cache_update::old_snapshot
struct stasis_message * old_snapshot
Old value from the cache.
Definition stasis.h:969
stasis_cache_update::new_snapshot
struct stasis_message * new_snapshot
New value.
Definition stasis.h:971
stasis_cache
Definition stasis_cache.c:46
stasis_caching_topic
Definition stasis_cache.c:55
stasis_message_type
Definition stasis_message.c:38
stasis_message
Definition stasis_message.c:121
stasis_subscription
Definition stasis.c:747
stasis_topic
Definition stasis.c:438
TEST_INIT
@ TEST_INIT
Definition test.h:200
TEST_EXECUTE
@ TEST_EXECUTE
Definition test.h:201
AST_TEST_PASS
@ AST_TEST_PASS
Definition test.h:195
AST_TEST_NOT_RUN
@ AST_TEST_NOT_RUN
Definition test.h:194
consumer_create
static struct consumer * consumer_create(void)
Definition test_devicestate.c:400
consumer_exec
static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_message *message)
Definition test_stasis.c:204
cache_test_data_id
static const char * cache_test_data_id(struct stasis_message *message)
Definition test_stasis.c:901
test_category
#define test_category
Definition test_stasis.c:41
cache_test_message_create
static struct stasis_message * cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
Definition test_stasis.c:896
consumer_wait_for
static int consumer_wait_for(struct consumer *consumer, size_t expected_len)
Definition test_stasis.c:246
RAII_VAR
#define RAII_VAR(vartype, varname, initval, dtor)
Declare a variable that will call a destructor function when it goes out of scope.
Definition utils.h:978

References ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, cache_test_data_id(), cache_test_message_create(), consumer_create(), consumer_exec(), consumer_wait_for(), consumer::messages_rxed, stasis_cache_update::new_snapshot, NULL, stasis_cache_update::old_snapshot, RAII_VAR, stasis_cache_clear_create(), stasis_cache_create(), stasis_cache_get(), stasis_cache_update_type(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe(), stasis_message_data(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [2/27]

AST_TEST_DEFINE ( cache_dump  )

Definition at line 1132 of file test_stasis.c.

1133{
1134 RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
1135 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1136 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
1137 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
1138 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
1139 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
1140 RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
1141 RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
1142 RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
1143 RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
1144 RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
1145 int actual_len;
1146 struct ao2_iterator i;
1147 void *obj;
1148
1149 switch (cmd) {
1150 case TEST_INIT:
1151 info->name = __func__;
1152 info->category = test_category;
1153 info->summary = "Test cache dump routines.";
1154 info->description = "Test cache dump routines.";
1155 return AST_TEST_NOT_RUN;
1156 case TEST_EXECUTE:
1157 break;
1158 }
1159
1160 ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
1161 ast_test_validate(test, NULL != cache_type);
1162 topic = stasis_topic_create("SomeTopic");
1163 ast_test_validate(test, NULL != topic);
1164 cache = stasis_cache_create(cache_test_data_id);
1165 ast_test_validate(test, NULL != cache);
1166 caching_topic = stasis_caching_topic_create(topic, cache);
1167 ast_test_validate(test, NULL != caching_topic);
1168 consumer = consumer_create(1);
1169 ast_test_validate(test, NULL != consumer);
1170 sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
1171 ast_test_validate(test, NULL != sub);
1172 ao2_ref(consumer, +1);
1173
1174 test_message1_1 = cache_test_message_create(cache_type, "1", "1");
1175 ast_test_validate(test, NULL != test_message1_1);
1176 test_message2_1 = cache_test_message_create(cache_type, "2", "1");
1177 ast_test_validate(test, NULL != test_message2_1);
1178
1179 /* Post a couple of snapshots */
1180 stasis_publish(topic, test_message1_1);
1181 stasis_publish(topic, test_message2_1);
1182 actual_len = consumer_wait_for(consumer, 2);
1183 ast_test_validate(test, 2 == actual_len);
1184
1185 /* Check the cache */
1186 ao2_cleanup(cache_dump);
1187 cache_dump = stasis_cache_dump(cache, NULL);
1188 ast_test_validate(test, NULL != cache_dump);
1189 ast_test_validate(test, 2 == ao2_container_count(cache_dump));
1190 i = ao2_iterator_init(cache_dump, 0);
1191 while ((obj = ao2_iterator_next(&i))) {
1192 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1193 ast_test_validate(test, actual_cache_entry == test_message1_1 || actual_cache_entry == test_message2_1);
1194 }
1195 ao2_iterator_destroy(&i);
1196
1197 /* Update snapshot 2 */
1198 test_message2_2 = cache_test_message_create(cache_type, "2", "2");
1199 ast_test_validate(test, NULL != test_message2_2);
1200 stasis_publish(topic, test_message2_2);
1201
1202 actual_len = consumer_wait_for(consumer, 3);
1203 ast_test_validate(test, 3 == actual_len);
1204
1205 /* Check the cache */
1206 ao2_cleanup(cache_dump);
1207 cache_dump = stasis_cache_dump(cache, NULL);
1208 ast_test_validate(test, NULL != cache_dump);
1209 ast_test_validate(test, 2 == ao2_container_count(cache_dump));
1210 i = ao2_iterator_init(cache_dump, 0);
1211 while ((obj = ao2_iterator_next(&i))) {
1212 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1213 ast_test_validate(test, actual_cache_entry == test_message1_1 || actual_cache_entry == test_message2_2);
1214 }
1215 ao2_iterator_destroy(&i);
1216
1217 /* Clear snapshot 1 */
1218 test_message1_clear = stasis_cache_clear_create(test_message1_1);
1219 ast_test_validate(test, NULL != test_message1_clear);
1220 stasis_publish(topic, test_message1_clear);
1221
1222 actual_len = consumer_wait_for(consumer, 4);
1223 ast_test_validate(test, 4 == actual_len);
1224
1225 /* Check the cache */
1226 ao2_cleanup(cache_dump);
1227 cache_dump = stasis_cache_dump(cache, NULL);
1228 ast_test_validate(test, NULL != cache_dump);
1229 ast_test_validate(test, 1 == ao2_container_count(cache_dump));
1230 i = ao2_iterator_init(cache_dump, 0);
1231 while ((obj = ao2_iterator_next(&i))) {
1232 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1233 ast_test_validate(test, actual_cache_entry == test_message2_2);
1234 }
1235 ao2_iterator_destroy(&i);
1236
1237 /* Dump the cache to ensure that it has no subscription change items in it since those aren't cached */
1238 ao2_cleanup(cache_dump);
1239 cache_dump = stasis_cache_dump(cache, stasis_subscription_change_type());
1240 ast_test_validate(test, 0 == ao2_container_count(cache_dump));
1241
1242 return AST_TEST_PASS;
1243}
ao2_iterator_next
#define ao2_iterator_next(iter)
Definition astobj2.h:1911
ao2_container_count
int ao2_container_count(struct ao2_container *c)
Returns the number of elements in a container.
Definition astobj2_container.c:34
ao2_iterator_init
struct ao2_iterator ao2_iterator_init(struct ao2_container *c, int flags) attribute_warn_unused_result
Create an iterator for a container.
Definition astobj2_container.c:485
ao2_iterator_destroy
void ao2_iterator_destroy(struct ao2_iterator *iter)
Destroy a container iterator.
Definition astobj2_container.c:534
stasis_subscription_change_type
struct stasis_message_type * stasis_subscription_change_type(void)
Gets the message type for subscription change notices.
stasis_cache_dump
struct ao2_container * stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
Dump cached items to a subscription for the ast_eid_default entity.
Definition stasis_cache.c:736
ao2_container
Generic container type.
Definition astobj2_container_private.h:270
ao2_iterator
When we need to walk through a container, we use an ao2_iterator to keep track of the current positio...
Definition astobj2.h:1821

References ao2_cleanup, ao2_container_count(), ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, cache_test_data_id(), cache_test_message_create(), consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_cache_clear_create(), stasis_cache_create(), stasis_cache_dump(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_subscription_change_type(), stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [3/27]

AST_TEST_DEFINE ( cache_eid_aggregate  )

Definition at line 1245 of file test_stasis.c.

1246{
1247 RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
1248 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1249 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
1250 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
1251 RAII_VAR(struct consumer *, cache_consumer, NULL, ao2_cleanup);
1252 RAII_VAR(struct consumer *, topic_consumer, NULL, ao2_cleanup);
1253 RAII_VAR(struct stasis_subscription *, topic_sub, NULL, stasis_unsubscribe);
1254 RAII_VAR(struct stasis_subscription *, cache_sub, NULL, stasis_unsubscribe);
1255 RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
1256 RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
1257 RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
1258 RAII_VAR(struct stasis_message *, test_message2_3, NULL, ao2_cleanup);
1259 RAII_VAR(struct stasis_message *, test_message2_4, NULL, ao2_cleanup);
1260 RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
1261 RAII_VAR(struct stasis_message *, test_message2_clear, NULL, ao2_cleanup);
1262 RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
1263 int actual_len;
1264 struct ao2_iterator i;
1265 void *obj;
1266 struct ast_eid foreign_eid1;
1267 struct ast_eid foreign_eid2;
1268
1269 switch (cmd) {
1270 case TEST_INIT:
1271 info->name = __func__;
1272 info->category = test_category;
1273 info->summary = "Test cache eid and aggregate support.";
1274 info->description = "Test cache eid and aggregate support.";
1275 return AST_TEST_NOT_RUN;
1276 case TEST_EXECUTE:
1277 break;
1278 }
1279
1280 memset(&foreign_eid1, 0xAA, sizeof(foreign_eid1));
1281 memset(&foreign_eid2, 0xBB, sizeof(foreign_eid2));
1282
1283 ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
1284 ast_test_validate(test, NULL != cache_type);
1285
1286 topic = stasis_topic_create("SomeTopic");
1287 ast_test_validate(test, NULL != topic);
1288
1289 /* To consume events published to the topic. */
1290 topic_consumer = consumer_create(1);
1291 ast_test_validate(test, NULL != topic_consumer);
1292
1293 topic_sub = stasis_subscribe(topic, consumer_exec, topic_consumer);
1294 ast_test_validate(test, NULL != topic_sub);
1295 ao2_ref(topic_consumer, +1);
1296
1297 cache = stasis_cache_create_full(cache_test_data_id,
1298 cache_test_aggregate_calc_fn, cache_test_aggregate_publish_fn);
1299 ast_test_validate(test, NULL != cache);
1300
1301 caching_topic = stasis_caching_topic_create(topic, cache);
1302 ast_test_validate(test, NULL != caching_topic);
1303
1304 /* To consume update events published to the caching_topic. */
1305 cache_consumer = consumer_create(1);
1306 ast_test_validate(test, NULL != cache_consumer);
1307
1308 cache_sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, cache_consumer);
1309 ast_test_validate(test, NULL != cache_sub);
1310 ao2_ref(cache_consumer, +1);
1311
1312 /* Create test messages. */
1313 test_message1_1 = cache_test_message_create_full(cache_type, "1", "1", &ast_eid_default);
1314 ast_test_validate(test, NULL != test_message1_1);
1315 test_message2_1 = cache_test_message_create_full(cache_type, "2", "1", &ast_eid_default);
1316 ast_test_validate(test, NULL != test_message2_1);
1317 test_message2_2 = cache_test_message_create_full(cache_type, "2", "2", &foreign_eid1);
1318 ast_test_validate(test, NULL != test_message2_2);
1319 test_message2_3 = cache_test_message_create_full(cache_type, "2", "3", &foreign_eid2);
1320 ast_test_validate(test, NULL != test_message2_3);
1321 test_message2_4 = cache_test_message_create_full(cache_type, "2", "4", &foreign_eid2);
1322 ast_test_validate(test, NULL != test_message2_4);
1323
1324 /* Post some snapshots */
1325 stasis_publish(topic, test_message1_1);
1326 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", "1"));
1327 stasis_publish(topic, test_message2_1);
1328 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "1"));
1329 stasis_publish(topic, test_message2_2);
1330 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "3"));
1331
1332 actual_len = consumer_wait_for(cache_consumer, 6);
1333 ast_test_validate(test, 6 == actual_len);
1334 actual_len = consumer_wait_for(topic_consumer, 6);
1335 ast_test_validate(test, 6 == actual_len);
1336
1337 /* Check the cache */
1338 ao2_cleanup(cache_dump);
1339 cache_dump = stasis_cache_dump_all(cache, NULL);
1340 ast_test_validate(test, NULL != cache_dump);
1341 ast_test_validate(test, 3 == ao2_container_count(cache_dump));
1342 i = ao2_iterator_init(cache_dump, 0);
1343 while ((obj = ao2_iterator_next(&i))) {
1344 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1345
1346 ast_test_validate(test,
1347 actual_cache_entry == test_message1_1
1348 || actual_cache_entry == test_message2_1
1349 || actual_cache_entry == test_message2_2);
1350 }
1351 ao2_iterator_destroy(&i);
1352
1353 /* Check the local cached items */
1354 ao2_cleanup(cache_dump);
1355 cache_dump = stasis_cache_dump_by_eid(cache, NULL, &ast_eid_default);
1356 ast_test_validate(test, NULL != cache_dump);
1357 ast_test_validate(test, 2 == ao2_container_count(cache_dump));
1358 i = ao2_iterator_init(cache_dump, 0);
1359 while ((obj = ao2_iterator_next(&i))) {
1360 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1361
1362 ast_test_validate(test,
1363 actual_cache_entry == test_message1_1
1364 || actual_cache_entry == test_message2_1);
1365 }
1366 ao2_iterator_destroy(&i);
1367
1368 /* Post snapshot 2 from another eid. */
1369 stasis_publish(topic, test_message2_3);
1370 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "6"));
1371
1372 actual_len = consumer_wait_for(cache_consumer, 8);
1373 ast_test_validate(test, 8 == actual_len);
1374 actual_len = consumer_wait_for(topic_consumer, 8);
1375 ast_test_validate(test, 8 == actual_len);
1376
1377 /* Check the cache */
1378 ao2_cleanup(cache_dump);
1379 cache_dump = stasis_cache_dump_all(cache, NULL);
1380 ast_test_validate(test, NULL != cache_dump);
1381 ast_test_validate(test, 4 == ao2_container_count(cache_dump));
1382 i = ao2_iterator_init(cache_dump, 0);
1383 while ((obj = ao2_iterator_next(&i))) {
1384 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1385
1386 ast_test_validate(test,
1387 actual_cache_entry == test_message1_1
1388 || actual_cache_entry == test_message2_1
1389 || actual_cache_entry == test_message2_2
1390 || actual_cache_entry == test_message2_3);
1391 }
1392 ao2_iterator_destroy(&i);
1393
1394 /* Check the remote cached items */
1395 ao2_cleanup(cache_dump);
1396 cache_dump = stasis_cache_dump_by_eid(cache, NULL, &foreign_eid1);
1397 ast_test_validate(test, NULL != cache_dump);
1398 ast_test_validate(test, 1 == ao2_container_count(cache_dump));
1399 i = ao2_iterator_init(cache_dump, 0);
1400 while ((obj = ao2_iterator_next(&i))) {
1401 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1402
1403 ast_test_validate(test, actual_cache_entry == test_message2_2);
1404 }
1405 ao2_iterator_destroy(&i);
1406
1407 /* Post snapshot 2 from a repeated eid. */
1408 stasis_publish(topic, test_message2_4);
1409 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "7"));
1410
1411 actual_len = consumer_wait_for(cache_consumer, 10);
1412 ast_test_validate(test, 10 == actual_len);
1413 actual_len = consumer_wait_for(topic_consumer, 10);
1414 ast_test_validate(test, 10 == actual_len);
1415
1416 /* Check the cache */
1417 ao2_cleanup(cache_dump);
1418 cache_dump = stasis_cache_dump_all(cache, NULL);
1419 ast_test_validate(test, NULL != cache_dump);
1420 ast_test_validate(test, 4 == ao2_container_count(cache_dump));
1421 i = ao2_iterator_init(cache_dump, 0);
1422 while ((obj = ao2_iterator_next(&i))) {
1423 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1424
1425 ast_test_validate(test,
1426 actual_cache_entry == test_message1_1
1427 || actual_cache_entry == test_message2_1
1428 || actual_cache_entry == test_message2_2
1429 || actual_cache_entry == test_message2_4);
1430 }
1431 ao2_iterator_destroy(&i);
1432
1433 /* Check all snapshot 2 cache entries. */
1434 ao2_cleanup(cache_dump);
1435 cache_dump = stasis_cache_get_all(cache, cache_type, "2");
1436 ast_test_validate(test, NULL != cache_dump);
1437 ast_test_validate(test, 3 == ao2_container_count(cache_dump));
1438 i = ao2_iterator_init(cache_dump, 0);
1439 while ((obj = ao2_iterator_next(&i))) {
1440 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1441
1442 ast_test_validate(test,
1443 actual_cache_entry == test_message2_1
1444 || actual_cache_entry == test_message2_2
1445 || actual_cache_entry == test_message2_4);
1446 }
1447 ao2_iterator_destroy(&i);
1448
1449 /* Clear snapshot 1 */
1450 test_message1_clear = stasis_cache_clear_create(test_message1_1);
1451 ast_test_validate(test, NULL != test_message1_clear);
1452 stasis_publish(topic, test_message1_clear);
1453 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", NULL));
1454
1455 actual_len = consumer_wait_for(cache_consumer, 12);
1456 ast_test_validate(test, 12 == actual_len);
1457 actual_len = consumer_wait_for(topic_consumer, 11);
1458 ast_test_validate(test, 11 == actual_len);
1459
1460 /* Check the cache */
1461 ao2_cleanup(cache_dump);
1462 cache_dump = stasis_cache_dump_all(cache, NULL);
1463 ast_test_validate(test, NULL != cache_dump);
1464 ast_test_validate(test, 3 == ao2_container_count(cache_dump));
1465 i = ao2_iterator_init(cache_dump, 0);
1466 while ((obj = ao2_iterator_next(&i))) {
1467 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1468
1469 ast_test_validate(test,
1470 actual_cache_entry == test_message2_1
1471 || actual_cache_entry == test_message2_2
1472 || actual_cache_entry == test_message2_4);
1473 }
1474 ao2_iterator_destroy(&i);
1475
1476 /* Clear snapshot 2 from a remote eid */
1477 test_message2_clear = stasis_cache_clear_create(test_message2_2);
1478 ast_test_validate(test, NULL != test_message2_clear);
1479 stasis_publish(topic, test_message2_clear);
1480 ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "5"));
1481
1482 actual_len = consumer_wait_for(cache_consumer, 14);
1483 ast_test_validate(test, 14 == actual_len);
1484 actual_len = consumer_wait_for(topic_consumer, 13);
1485 ast_test_validate(test, 13 == actual_len);
1486
1487 /* Check the cache */
1488 ao2_cleanup(cache_dump);
1489 cache_dump = stasis_cache_dump_all(cache, NULL);
1490 ast_test_validate(test, NULL != cache_dump);
1491 ast_test_validate(test, 2 == ao2_container_count(cache_dump));
1492 i = ao2_iterator_init(cache_dump, 0);
1493 while ((obj = ao2_iterator_next(&i))) {
1494 RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
1495
1496 ast_test_validate(test,
1497 actual_cache_entry == test_message2_1
1498 || actual_cache_entry == test_message2_4);
1499 }
1500 ao2_iterator_destroy(&i);
1501
1502 return AST_TEST_PASS;
1503}
stasis_cache_get_all
struct ao2_container * stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
Retrieve all matching entity items from the cache.
Definition stasis_cache.c:587
stasis_cache_dump_all
struct ao2_container * stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
Dump all entity items from the cache to a subscription.
Definition stasis_cache.c:757
stasis_cache_dump_by_eid
struct ao2_container * stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
Dump cached items to a subscription for a specific entity.
Definition stasis_cache.c:718
stasis_cache_create_full
struct stasis_cache * stasis_cache_create_full(snapshot_get_id id_fn, cache_aggregate_calc_fn aggregate_calc_fn, cache_aggregate_publish_fn aggregate_publish_fn)
Create a cache.
Definition stasis_cache.c:334
ast_eid
An Entity ID is essentially a MAC address, brief and unique.
Definition utils.h:850
cache_test_aggregate_calc_fn
static struct stasis_message * cache_test_aggregate_calc_fn(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
Definition test_stasis.c:911
cache_test_message_create_full
static struct stasis_message * cache_test_message_create_full(struct stasis_message_type *type, const char *name, const char *value, struct ast_eid *eid)
Definition test_stasis.c:875
cache_test_aggregate_publish_fn
static void cache_test_aggregate_publish_fn(struct stasis_topic *topic, struct stasis_message *aggregate)
Definition test_stasis.c:965
check_cache_aggregate
static int check_cache_aggregate(struct stasis_cache *cache, struct stasis_message_type *cache_type, const char *id, const char *value)
Definition test_stasis.c:970
ast_eid_default
struct ast_eid ast_eid_default
Global EID.
Definition options.c:94

References ao2_cleanup, ao2_container_count(), ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_eid_default, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, cache_test_aggregate_calc_fn(), cache_test_aggregate_publish_fn(), cache_test_data_id(), cache_test_message_create_full(), check_cache_aggregate(), consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_cache_clear_create(), stasis_cache_create_full(), stasis_cache_dump_all(), stasis_cache_dump_by_eid(), stasis_cache_get_all(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [4/27]

AST_TEST_DEFINE ( cache_filter  )

Definition at line 986 of file test_stasis.c.

987{
988 RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
989 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
990 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
991 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
992 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
993 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
994 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
995 int actual_len;
996
997 switch (cmd) {
998 case TEST_INIT:
999 info->name = __func__;
1000 info->category = test_category;
1001 info->summary = "Test caching topics only forward cache_update messages.";
1002 info->description = "Test caching topics only forward cache_update messages.";
1003 return AST_TEST_NOT_RUN;
1004 case TEST_EXECUTE:
1005 break;
1006 }
1007
1008 ast_test_validate(test, stasis_message_type_create("NonCacheable", NULL, &non_cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
1009 ast_test_validate(test, NULL != non_cache_type);
1010 topic = stasis_topic_create("SomeTopic");
1011 ast_test_validate(test, NULL != topic);
1012 cache = stasis_cache_create(cache_test_data_id);
1013 ast_test_validate(test, NULL != cache);
1014 caching_topic = stasis_caching_topic_create(topic, cache);
1015 ast_test_validate(test, NULL != caching_topic);
1016 consumer = consumer_create(1);
1017 ast_test_validate(test, NULL != consumer);
1018 sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
1019 ast_test_validate(test, NULL != sub);
1020 ao2_ref(consumer, +1);
1021
1022 test_message = cache_test_message_create(non_cache_type, "1", "1");
1023 ast_test_validate(test, NULL != test_message);
1024
1025 stasis_publish(topic, test_message);
1026
1027 actual_len = consumer_should_stay(consumer, 0);
1028 ast_test_validate(test, 0 == actual_len);
1029
1030 return AST_TEST_PASS;
1031}
consumer_should_stay
static int consumer_should_stay(struct consumer *consumer, size_t expected_len)
Definition test_stasis.c:288

References ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, cache_test_data_id(), cache_test_message_create(), consumer_create(), consumer_exec(), consumer_should_stay(), NULL, RAII_VAR, stasis_cache_create(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [5/27]

AST_TEST_DEFINE ( caching_dtor_order  )

Definition at line 2006 of file test_stasis.c.

2007{
2008 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
2009 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
2010 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL,
2011 stasis_caching_unsubscribe);
2012 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
2013
2014 switch (cmd) {
2015 case TEST_INIT:
2016 info->name = __func__;
2017 info->category = test_category;
2018 info->summary = "Test that destruction order doesn't bomb stuff";
2019 info->description = "Test that destruction order doesn't bomb stuff";
2020 return AST_TEST_NOT_RUN;
2021 case TEST_EXECUTE:
2022 break;
2023 }
2024
2025 cache = stasis_cache_create(noop_get_id);
2026 ast_test_validate(test, NULL != cache);
2027
2028 topic = stasis_topic_create("test-topic");
2029 ast_test_validate(test, NULL != topic);
2030
2031 caching_topic = stasis_caching_topic_create(topic, cache);
2032 ast_test_validate(test, NULL != caching_topic);
2033
2034 sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), noop,
2035 NULL);
2036 ast_test_validate(test, NULL != sub);
2037
2038 /* With any luck, this won't completely blow everything up */
2039 ao2_cleanup(cache);
2040 ao2_cleanup(topic);
2041 stasis_caching_unsubscribe(caching_topic);
2042 stasis_unsubscribe(sub);
2043
2044 /* These refs were cleaned up manually */
2045 cache = NULL;
2046 topic = NULL;
2047 caching_topic = NULL;
2048 sub = NULL;
2049
2050 return AST_TEST_PASS;
2051}
noop_get_id
static const char * noop_get_id(struct stasis_message *message)
Definition test_stasis.c:2001
noop
static void noop(void *data, struct stasis_subscription *sub, struct stasis_message *message)
Definition test_stasis.c:1962

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, noop(), noop_get_id(), NULL, RAII_VAR, stasis_cache_create(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [6/27]

AST_TEST_DEFINE ( combo_filters  )

Definition at line 2373 of file test_stasis.c.

2374{
2375 RAII_VAR(struct cts *, cts, NULL, ao2_cleanup);
2376 RAII_VAR(struct test_message_types *, types, NULL, ao2_cleanup);
2377 int ix = 0;
2378
2379 switch (cmd) {
2380 case TEST_INIT:
2381 info->name = __func__;
2382 info->category = test_category "filtering/";
2383 info->summary = "Test message filtering by type and formatter";
2384 info->description = "Test message filtering by type and formatter";
2385 return AST_TEST_NOT_RUN;
2386 case TEST_EXECUTE:
2387 break;
2388 }
2389
2390 types = create_message_types(test);
2391 ast_test_validate(test, NULL != types);
2392
2393 cts = create_cts(test);
2394 ast_test_validate(test, NULL != cts);
2395
2396 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->type1) == 0);
2397 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->type2) == 0);
2398 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->change) == 0);
2399 ast_test_validate(test, stasis_subscription_set_filter(cts->sub, STASIS_SUBSCRIPTION_FILTER_SELECTIVE) == 0);
2400 stasis_subscription_accept_formatters(cts->sub,
2401 STASIS_SUBSCRIPTION_FORMATTER_AMI | STASIS_SUBSCRIPTION_FORMATTER_JSON);
2402
2403 /* We should get these */
2404 ast_test_validate(test, send_msg(test, cts, types->type1, "Pass"));
2405 ast_test_validate(test, send_msg(test, cts, types->type2, "Pass"));
2406 ast_test_validate(test, send_msg(test, cts, types->ami, "Pass"));
2407 ast_test_validate(test, send_msg(test, cts, types->amievent, "Pass"));
2408 ast_test_validate(test, send_msg(test, cts, types->json, "Pass"));
2409
2410 /* ... but not these */
2411 ast_test_validate(test, send_msg(test, cts, types->type3, "FAIL"));
2412 ast_test_validate(test, send_msg(test, cts, types->event, "FAIL"));
2413
2414 /* Wait for change(subscribe) and the "Pass" messages */
2415 consumer_wait_for(cts->consumer, 6);
2416
2417 stasis_unsubscribe(cts->sub);
2418 cts->sub = NULL;
2419 consumer_wait_for_completion(cts->consumer);
2420
2421 dump_consumer(test, cts);
2422
2423 ast_test_validate(test, 1 == cts->consumer->complete);
2424 ast_test_validate(test, 7 == cts->consumer->messages_rxed_len);
2425 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Subscribe"));
2426 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->type1, "Pass"));
2427 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->type2, "Pass"));
2428 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->ami, "Pass"));
2429 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->amievent, "Pass"));
2430 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->json, "Pass"));
2431 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Unsubscribe"));
2432
2433 return AST_TEST_PASS;
2434}
STASIS_SUBSCRIPTION_FILTER_SELECTIVE
@ STASIS_SUBSCRIPTION_FILTER_SELECTIVE
Definition stasis.h:297
stasis_subscription_accept_message_type
int stasis_subscription_accept_message_type(struct stasis_subscription *subscription, const struct stasis_message_type *type)
Indicate to a subscription that we are interested in a message type.
Definition stasis.c:1090
stasis_subscription_set_filter
int stasis_subscription_set_filter(struct stasis_subscription *subscription, enum stasis_subscription_message_filter filter)
Set the message type filtering level on a subscription.
Definition stasis.c:1144
STASIS_SUBSCRIPTION_FORMATTER_AMI
@ STASIS_SUBSCRIPTION_FORMATTER_AMI
Definition stasis.h:311
STASIS_SUBSCRIPTION_FORMATTER_JSON
@ STASIS_SUBSCRIPTION_FORMATTER_JSON
Definition stasis.h:310
stasis_subscription_accept_formatters
void stasis_subscription_accept_formatters(struct stasis_subscription *subscription, enum stasis_subscription_message_formatters formatters)
Indicate to a subscription that we are interested in messages with one or more formatters.
Definition stasis.c:1160
consumer::messages_rxed_len
size_t messages_rxed_len
Definition test_stasis.c:165
cts::sub
struct stasis_subscription * sub
Definition test_stasis.c:2142
cts::consumer
struct consumer * consumer
Definition test_stasis.c:2140
test_message_types
Definition test_stasis.c:2053
create_message_types
static struct test_message_types * create_message_types(struct ast_test *test)
Definition test_stasis.c:2080
consumer_wait_for_completion
static int consumer_wait_for_completion(struct consumer *consumer)
Definition test_stasis.c:267
send_msg
static int send_msg(struct ast_test *test, struct cts *cts, struct stasis_message_type *msg_type, const char *data)
Definition test_stasis.c:2208
is_msg
static int is_msg(struct stasis_message *msg, struct stasis_message_type *mtype, const char *data)
Definition test_stasis.c:2178
create_cts
static struct cts * create_cts(struct ast_test *test)
Definition test_stasis.c:2154
dump_consumer
static void dump_consumer(struct ast_test *test, struct cts *cts)
Definition test_stasis.c:2193

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer::complete, cts::consumer, consumer_wait_for(), consumer_wait_for_completion(), create_cts(), create_message_types(), dump_consumer(), is_msg(), consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, send_msg(), stasis_subscription_accept_formatters(), stasis_subscription_accept_message_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, STASIS_SUBSCRIPTION_FORMATTER_AMI, STASIS_SUBSCRIPTION_FORMATTER_JSON, stasis_subscription_set_filter(), stasis_unsubscribe(), cts::sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [7/27]

AST_TEST_DEFINE ( dtor_order  )

Definition at line 1968 of file test_stasis.c.

1969{
1970 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1971 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
1972
1973 switch (cmd) {
1974 case TEST_INIT:
1975 info->name = __func__;
1976 info->category = test_category;
1977 info->summary = "Test that destruction order doesn't bomb stuff";
1978 info->description = "Test that destruction order doesn't bomb stuff";
1979 return AST_TEST_NOT_RUN;
1980 case TEST_EXECUTE:
1981 break;
1982 }
1983
1984 topic = stasis_topic_create("test-topic");
1985 ast_test_validate(test, NULL != topic);
1986
1987 sub = stasis_subscribe(topic, noop, NULL);
1988 ast_test_validate(test, NULL != sub);
1989
1990 /* With any luck, this won't completely blow everything up */
1991 ao2_cleanup(topic);
1992 stasis_unsubscribe(sub);
1993
1994 /* These refs were cleaned up manually */
1995 topic = NULL;
1996 sub = NULL;
1997
1998 return AST_TEST_PASS;
1999}

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, noop(), NULL, RAII_VAR, stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [8/27]

AST_TEST_DEFINE ( formatter_filters  )

Definition at line 2301 of file test_stasis.c.

2302{
2303 RAII_VAR(struct cts *, cts, NULL, ao2_cleanup);
2304 RAII_VAR(struct test_message_types *, types, NULL, ao2_cleanup) ;
2305 int ix = 0;
2306
2307 switch (cmd) {
2308 case TEST_INIT:
2309 info->name = __func__;
2310 info->category = test_category "filtering/";
2311 info->summary = "Test message filtering by formatter";
2312 info->description = "Test message filtering by formatter";
2313 return AST_TEST_NOT_RUN;
2314 case TEST_EXECUTE:
2315 break;
2316 }
2317
2318 types = create_message_types(test);
2319 ast_test_validate(test, NULL != types);
2320
2321 cts = create_cts(test);
2322 ast_test_validate(test, NULL != cts);
2323
2324 stasis_subscription_accept_formatters(cts->sub,
2325 STASIS_SUBSCRIPTION_FORMATTER_AMI | STASIS_SUBSCRIPTION_FORMATTER_JSON);
2326
2327 /* We should get these */
2328 ast_test_validate(test, send_msg(test, cts, types->ami, "Pass"));
2329 ast_test_validate(test, send_msg(test, cts, types->json, "Pass"));
2330 ast_test_validate(test, send_msg(test, cts, types->amievent, "Pass"));
2331
2332 /* ... but not these */
2333 ast_test_validate(test, send_msg(test, cts, types->none, "FAIL"));
2334 ast_test_validate(test, send_msg(test, cts, types->event, "FAIL"));
2335 ast_test_validate(test, send_msg(test, cts, types->type1, "FAIL"));
2336
2337 /* Wait for change(subscribe) and the "Pass" messages */
2338 consumer_wait_for(cts->consumer, 4);
2339
2340 /* Change the subscription to accept only event formatters */
2341 stasis_subscription_accept_formatters(cts->sub, STASIS_SUBSCRIPTION_FORMATTER_EVENT);
2342
2343 /* We should NOT get these now */
2344 ast_test_validate(test, send_msg(test, cts, types->ami, "FAIL"));
2345 ast_test_validate(test, send_msg(test, cts, types->json, "FAIL"));
2346 /* ... but we should still get this one */
2347 ast_test_validate(test, send_msg(test, cts, types->amievent, "Pass2"));
2348 /* ... and this one should be new */
2349 ast_test_validate(test, send_msg(test, cts, types->event, "Pass"));
2350
2351 /* We should now have a second amievent */
2352 consumer_wait_for(cts->consumer, 6);
2353
2354 stasis_unsubscribe(cts->sub);
2355 cts->sub = NULL;
2356 consumer_wait_for_completion(cts->consumer);
2357
2358 dump_consumer(test, cts);
2359
2360 ast_test_validate(test, 1 == cts->consumer->complete);
2361 ast_test_validate(test, 7 == cts->consumer->messages_rxed_len);
2362 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Subscribe"));
2363 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->ami, "Pass"));
2364 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->json, "Pass"));
2365 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->amievent, "Pass"));
2366 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->amievent, "Pass2"));
2367 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->event, "Pass"));
2368 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Unsubscribe"));
2369
2370 return AST_TEST_PASS;
2371}
STASIS_SUBSCRIPTION_FORMATTER_EVENT
@ STASIS_SUBSCRIPTION_FORMATTER_EVENT
Definition stasis.h:312

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer::complete, cts::consumer, consumer_wait_for(), consumer_wait_for_completion(), create_cts(), create_message_types(), dump_consumer(), is_msg(), consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, send_msg(), stasis_subscription_accept_formatters(), STASIS_SUBSCRIPTION_FORMATTER_AMI, STASIS_SUBSCRIPTION_FORMATTER_EVENT, STASIS_SUBSCRIPTION_FORMATTER_JSON, stasis_unsubscribe(), cts::sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [9/27]

AST_TEST_DEFINE ( forward  )

Definition at line 613 of file test_stasis.c.

614{
615 RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
616 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
617
618 RAII_VAR(struct consumer *, parent_consumer, NULL, ao2_cleanup);
619 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
620
621 RAII_VAR(struct stasis_forward *, forward_sub, NULL, stasis_forward_cancel);
622 RAII_VAR(struct stasis_subscription *, parent_sub, NULL, stasis_unsubscribe);
623 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
624
625 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
626 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
627 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
628 int actual_len;
629
630 switch (cmd) {
631 case TEST_INIT:
632 info->name = __func__;
633 info->category = test_category;
634 info->summary = "Test sending events to a parent topic";
635 info->description = "Test sending events to a parent topic.\n"
636 "This test creates three topics (one parent, two children)\n"
637 "and publishes a message to one child, and verifies it's\n"
638 "only seen by that child and the parent";
639 return AST_TEST_NOT_RUN;
640 case TEST_EXECUTE:
641 break;
642 }
643
644 parent_topic = stasis_topic_create("ParentTestTopic");
645 ast_test_validate(test, NULL != parent_topic);
646 topic = stasis_topic_create("TestTopic");
647 ast_test_validate(test, NULL != topic);
648
649 forward_sub = stasis_forward_all(topic, parent_topic);
650 ast_test_validate(test, NULL != forward_sub);
651
652 parent_consumer = consumer_create(1);
653 ast_test_validate(test, NULL != parent_consumer);
654 consumer = consumer_create(1);
655 ast_test_validate(test, NULL != consumer);
656
657 parent_sub = stasis_subscribe(parent_topic, consumer_exec, parent_consumer);
658 ast_test_validate(test, NULL != parent_sub);
659 ao2_ref(parent_consumer, +1);
660 sub = stasis_subscribe(topic, consumer_exec, consumer);
661 ast_test_validate(test, NULL != sub);
662 ao2_ref(consumer, +1);
663
664 test_data = ao2_alloc(1, NULL);
665 ast_test_validate(test, NULL != test_data);
666 ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
667 test_message = stasis_message_create(test_message_type, test_data);
668
669 stasis_publish(topic, test_message);
670
671 actual_len = consumer_wait_for(consumer, 1);
672 ast_test_validate(test, 1 == actual_len);
673 actual_len = consumer_wait_for(parent_consumer, 1);
674 ast_test_validate(test, 1 == actual_len);
675
676 return AST_TEST_PASS;
677}
ao2_alloc
#define ao2_alloc(data_size, destructor_fn)
Definition astobj2.h:409
stasis_forward_cancel
struct stasis_forward * stasis_forward_cancel(struct stasis_forward *forward)
Definition stasis.c:1615
stasis_message_create
struct stasis_message * stasis_message_create(struct stasis_message_type *type, void *data)
Create a new message.
Definition stasis_message.c:174
stasis_forward_all
struct stasis_forward * stasis_forward_all(struct stasis_topic *from_topic, struct stasis_topic *to_topic)
Create a subscription which forwards all messages from one topic to another.
Definition stasis.c:1645
stasis_forward
Forwarding information.
Definition stasis.c:1598
test_data
Sorcery object created based on backend data.
Definition test_res_pjsip_scheduler.c:45

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_forward_all(), stasis_forward_cancel(), stasis_message_create(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, TEST_INIT, and stasis_subscription_change::topic.

◆ AST_TEST_DEFINE() [10/27]

AST_TEST_DEFINE ( interleaving  )

Definition at line 679 of file test_stasis.c.

680{
681 RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
682 RAII_VAR(struct stasis_topic *, topic1, NULL, ao2_cleanup);
683 RAII_VAR(struct stasis_topic *, topic2, NULL, ao2_cleanup);
684
685 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
686
687 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
688
689 RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
690 RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
691 RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
692
693 RAII_VAR(struct stasis_forward *, forward_sub1, NULL, stasis_forward_cancel);
694 RAII_VAR(struct stasis_forward *, forward_sub2, NULL, stasis_forward_cancel);
695 RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
696
697 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
698
699 int actual_len;
700
701 switch (cmd) {
702 case TEST_INIT:
703 info->name = __func__;
704 info->category = test_category;
705 info->summary = "Test sending interleaved events to a parent topic";
706 info->description = "Test sending events to a parent topic.\n"
707 "This test creates three topics (one parent, two children)\n"
708 "and publishes messages alternately between the children.\n"
709 "It verifies that the messages are received in the expected\n"
710 "order.";
711 return AST_TEST_NOT_RUN;
712 case TEST_EXECUTE:
713 break;
714 }
715
716 ast_test_validate(test, stasis_message_type_create("test", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
717 ast_test_validate(test, NULL != test_message_type);
718
719 test_data = ao2_alloc(1, NULL);
720 ast_test_validate(test, NULL != test_data);
721
722 test_message1 = stasis_message_create(test_message_type, test_data);
723 ast_test_validate(test, NULL != test_message1);
724 test_message2 = stasis_message_create(test_message_type, test_data);
725 ast_test_validate(test, NULL != test_message2);
726 test_message3 = stasis_message_create(test_message_type, test_data);
727 ast_test_validate(test, NULL != test_message3);
728
729 parent_topic = stasis_topic_create("ParentTestTopic");
730 ast_test_validate(test, NULL != parent_topic);
731 topic1 = stasis_topic_create("Topic1");
732 ast_test_validate(test, NULL != topic1);
733 topic2 = stasis_topic_create("Topic2");
734 ast_test_validate(test, NULL != topic2);
735
736 forward_sub1 = stasis_forward_all(topic1, parent_topic);
737 ast_test_validate(test, NULL != forward_sub1);
738 forward_sub2 = stasis_forward_all(topic2, parent_topic);
739 ast_test_validate(test, NULL != forward_sub2);
740
741 consumer = consumer_create(1);
742 ast_test_validate(test, NULL != consumer);
743
744 sub = stasis_subscribe(parent_topic, consumer_exec, consumer);
745 ast_test_validate(test, NULL != sub);
746 ao2_ref(consumer, +1);
747
748 stasis_publish(topic1, test_message1);
749 stasis_publish(topic2, test_message2);
750 stasis_publish(topic1, test_message3);
751
752 actual_len = consumer_wait_for(consumer, 3);
753 ast_test_validate(test, 3 == actual_len);
754
755 ast_test_validate(test, test_message1 == consumer->messages_rxed[0]);
756 ast_test_validate(test, test_message2 == consumer->messages_rxed[1]);
757 ast_test_validate(test, test_message3 == consumer->messages_rxed[2]);
758
759 return AST_TEST_PASS;
760}

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), consumer::messages_rxed, NULL, RAII_VAR, stasis_forward_all(), stasis_forward_cancel(), stasis_message_create(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [11/27]

AST_TEST_DEFINE ( message  )

Definition at line 99 of file test_stasis.c.

100{
101 RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
102 RAII_VAR(struct stasis_message *, uut1, NULL, ao2_cleanup);
103 RAII_VAR(struct stasis_message *, uut2, NULL, ao2_cleanup);
104 RAII_VAR(char *, data, NULL, ao2_cleanup);
105 char *expected = "SomeData";
106 struct timeval expected_timestamp;
107 struct timeval time_diff;
108 struct ast_eid foreign_eid;
109
110 switch (cmd) {
111 case TEST_INIT:
112 info->name = __func__;
113 info->category = test_category;
114 info->summary = "Test basic message functions";
115 info->description = "Test basic message functions";
116 return AST_TEST_NOT_RUN;
117 case TEST_EXECUTE:
118 break;
119 }
120
121
122 memset(&foreign_eid, 0xFF, sizeof(foreign_eid));
123
124 ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
125
126 ast_test_validate(test, NULL == stasis_message_create_full(NULL, NULL, NULL));
127 ast_test_validate(test, NULL == stasis_message_create_full(type, NULL, NULL));
128
129 data = ao2_alloc(strlen(expected) + 1, NULL);
130 strcpy(data, expected);/* Safe */
131 expected_timestamp = ast_tvnow();
132 uut1 = stasis_message_create_full(type, data, &foreign_eid);
133 uut2 = stasis_message_create_full(type, data, NULL);
134
135 ast_test_validate(test, NULL != uut1);
136 ast_test_validate(test, NULL != uut2);
137 ast_test_validate(test, type == stasis_message_type(uut1));
138 ast_test_validate(test, type == stasis_message_type(uut2));
139 ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut1)));
140 ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut2)));
141 ast_test_validate(test, NULL != stasis_message_eid(uut1));
142 ast_test_validate(test, NULL == stasis_message_eid(uut2));
143 ast_test_validate(test, !ast_eid_cmp(&foreign_eid, stasis_message_eid(uut1)));
144
145 ast_test_validate(test, 3 == ao2_ref(data, 0)); /* uut1 and uut2 have ref to data */
146
147 time_diff = ast_tvsub(*stasis_message_timestamp(uut1), expected_timestamp);
148 /* 10ms is certainly long enough for the two calls to complete */
149 ast_test_validate(test, time_diff.tv_sec == 0);
150 ast_test_validate(test, time_diff.tv_usec < 10000);
151
152 ao2_ref(uut1, -1);
153 uut1 = NULL;
154 ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut1 unreffed data */
155 ao2_ref(uut2, -1);
156 uut2 = NULL;
157 ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut2 unreffed data */
158
159 return AST_TEST_PASS;
160}
type
static const char type[]
Definition chan_ooh323.c:109
stasis_message_eid
const struct ast_eid * stasis_message_eid(const struct stasis_message *msg)
Get the entity id for a stasis_message.
Definition stasis_message.c:179
stasis_message_create_full
struct stasis_message * stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid)
Create a new message for an entity.
Definition stasis_message.c:140
stasis_message_timestamp
const struct timeval * stasis_message_timestamp(const struct stasis_message *msg)
Get the time when a message was created.
Definition stasis_message.c:203
ast_tvsub
struct timeval ast_tvsub(struct timeval a, struct timeval b)
Returns the difference of two timevals a - b.
Definition extconf.c:2295
ast_tvnow
struct timeval ast_tvnow(void)
Returns current timeval. Meant to replace calls to gettimeofday().
Definition time.h:159
ast_eid_cmp
int ast_eid_cmp(const struct ast_eid *eid1, const struct ast_eid *eid2)
Compare two EIDs.
Definition utils.c:3130

References ao2_alloc, ao2_cleanup, ao2_ref, ast_eid_cmp(), AST_TEST_NOT_RUN, AST_TEST_PASS, ast_tvnow(), ast_tvsub(), NULL, RAII_VAR, stasis_message_create_full(), stasis_message_data(), stasis_message_eid(), stasis_message_timestamp(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, TEST_INIT, and type.

◆ AST_TEST_DEFINE() [12/27]

AST_TEST_DEFINE ( message_type  )

Definition at line 77 of file test_stasis.c.

78{
79 RAII_VAR(struct stasis_message_type *, uut, NULL, ao2_cleanup);
80
81 switch (cmd) {
82 case TEST_INIT:
83 info->name = __func__;
84 info->category = test_category;
85 info->summary = "Test basic message_type functions";
86 info->description = "Test basic message_type functions";
87 return AST_TEST_NOT_RUN;
88 case TEST_EXECUTE:
89 break;
90 }
91
92 ast_test_validate(test, stasis_message_type_create(NULL, NULL, NULL) == STASIS_MESSAGE_TYPE_ERROR);
93 ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &uut) == STASIS_MESSAGE_TYPE_SUCCESS);
94 ast_test_validate(test, 0 == strcmp(stasis_message_type_name(uut), "SomeMessage"));
95
96 return AST_TEST_PASS;
97}
stasis_message_type_name
const char * stasis_message_type_name(const struct stasis_message_type *type)
Gets the name of a given message type.
Definition stasis_message.c:99
STASIS_MESSAGE_TYPE_ERROR
@ STASIS_MESSAGE_TYPE_ERROR
Definition stasis.h:286

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, NULL, RAII_VAR, stasis_message_type_create(), STASIS_MESSAGE_TYPE_ERROR, stasis_message_type_name(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [13/27]

AST_TEST_DEFINE ( no_to_ami  )

Definition at line 1892 of file test_stasis.c.

1893{
1894 RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
1895 RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
1896 RAII_VAR(char *, data, NULL, ao2_cleanup);
1897 RAII_VAR(struct ast_manager_event_blob *, actual, NULL, ao2_cleanup);
1898 char *expected = "SomeData";
1899
1900 switch (cmd) {
1901 case TEST_INIT:
1902 info->name = __func__;
1903 info->category = test_category;
1904 info->summary = "Test message to_ami function when NULL";
1905 info->description = "Test message to_ami function when NULL";
1906 return AST_TEST_NOT_RUN;
1907 case TEST_EXECUTE:
1908 break;
1909 }
1910
1911 /* Test NULL */
1912 actual = stasis_message_to_ami(NULL);
1913 ast_test_validate(test, NULL == actual);
1914
1915 /* Test message with NULL to_ami function */
1916 ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
1917
1918 data = ao2_alloc(strlen(expected) + 1, NULL);
1919 strcpy(data, expected);
1920 uut = stasis_message_create(type, data);
1921 ast_test_validate(test, NULL != uut);
1922
1923 actual = stasis_message_to_ami(uut);
1924 ast_test_validate(test, NULL == actual);
1925
1926 return AST_TEST_PASS;
1927}
stasis_message_to_ami
struct ast_manager_event_blob * stasis_message_to_ami(struct stasis_message *msg)
Build the AMI representation of the message.
Definition stasis_message.c:224
ast_manager_event_blob
Struct containing info for an AMI event to send out.
Definition manager.h:503

References ao2_alloc, ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, stasis_message::data, NULL, RAII_VAR, stasis_message_create(), stasis_message_to_ami(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, TEST_INIT, and type.

◆ AST_TEST_DEFINE() [14/27]

AST_TEST_DEFINE ( no_to_json  )

Definition at line 1821 of file test_stasis.c.

1822{
1823 RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
1824 RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
1825 RAII_VAR(char *, data, NULL, ao2_cleanup);
1826 RAII_VAR(struct ast_json *, actual, NULL, ast_json_unref);
1827 char *expected = "SomeData";
1828
1829 switch (cmd) {
1830 case TEST_INIT:
1831 info->name = __func__;
1832 info->category = test_category;
1833 info->summary = "Test message to_json function";
1834 info->description = "Test message to_json function";
1835 return AST_TEST_NOT_RUN;
1836 case TEST_EXECUTE:
1837 break;
1838 }
1839
1840 /* Test NULL */
1841 actual = stasis_message_to_json(NULL, NULL);
1842 ast_test_validate(test, NULL == actual);
1843
1844 /* Test message with NULL to_json function */
1845 ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
1846
1847 data = ao2_alloc(strlen(expected) + 1, NULL);
1848 strcpy(data, expected);
1849 uut = stasis_message_create(type, data);
1850 ast_test_validate(test, NULL != uut);
1851
1852 actual = stasis_message_to_json(uut, NULL);
1853 ast_test_validate(test, NULL == actual);
1854
1855 return AST_TEST_PASS;
1856}
ast_json_unref
void ast_json_unref(struct ast_json *value)
Decrease refcount on value. If refcount reaches zero, value is freed.
Definition json.c:73
stasis_message_to_json
struct ast_json * stasis_message_to_json(struct stasis_message *msg, struct stasis_message_sanitizer *sanitize)
Build the JSON representation of the message.
Definition stasis_message.c:229
ast_json
Abstract JSON element (object, array, string, int, ...).

References ao2_alloc, ao2_cleanup, ast_json_unref(), AST_TEST_NOT_RUN, AST_TEST_PASS, stasis_message::data, NULL, RAII_VAR, stasis_message_create(), stasis_message_to_json(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, TEST_INIT, and type.

◆ AST_TEST_DEFINE() [15/27]

AST_TEST_DEFINE ( publish  )

Definition at line 424 of file test_stasis.c.

425{
426 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
427 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
428 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
429 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
430 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
431 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
432 int actual_len;
433 const char *actual;
434
435 switch (cmd) {
436 case TEST_INIT:
437 info->name = __func__;
438 info->category = test_category;
439 info->summary = "Test publishing";
440 info->description = "Test publishing";
441 return AST_TEST_NOT_RUN;
442 case TEST_EXECUTE:
443 break;
444 }
445
446 topic = stasis_topic_create("TestTopic");
447 ast_test_validate(test, NULL != topic);
448
449 consumer = consumer_create(1);
450 ast_test_validate(test, NULL != consumer);
451
452 uut = stasis_subscribe(topic, consumer_exec, consumer);
453 ast_test_validate(test, NULL != uut);
454 ao2_ref(consumer, +1);
455
456 test_data = ao2_alloc(1, NULL);
457 ast_test_validate(test, NULL != test_data);
458 ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
459 test_message = stasis_message_create(test_message_type, test_data);
460
461 stasis_publish(topic, test_message);
462
463 actual_len = consumer_wait_for(consumer, 1);
464 ast_test_validate(test, 1 == actual_len);
465 actual = stasis_message_data(consumer->messages_rxed[0]);
466 ast_test_validate(test, test_data == actual);
467
468 return AST_TEST_PASS;
469}

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), consumer::messages_rxed, NULL, RAII_VAR, stasis_message_create(), stasis_message_data(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, and stasis_subscription_change::topic.

◆ AST_TEST_DEFINE() [16/27]

AST_TEST_DEFINE ( publish_pool  )

Definition at line 518 of file test_stasis.c.

519{
520 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
521 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
522 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
523 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
524 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
525 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
526 int actual_len;
527 const char *actual;
528
529 switch (cmd) {
530 case TEST_INIT:
531 info->name = __func__;
532 info->category = test_category;
533 info->summary = "Test publishing with a taskpool";
534 info->description = "Test publishing to a subscriber whose\n"
535 "subscription dictates messages are received through a\n"
536 "taskpool.";
537 return AST_TEST_NOT_RUN;
538 case TEST_EXECUTE:
539 break;
540 }
541
542 topic = stasis_topic_create("TestTopic");
543 ast_test_validate(test, NULL != topic);
544
545 consumer = consumer_create(1);
546 ast_test_validate(test, NULL != consumer);
547
548 uut = stasis_subscribe_pool(topic, consumer_exec, consumer);
549 ast_test_validate(test, NULL != uut);
550 ao2_ref(consumer, +1);
551
552 test_data = ao2_alloc(1, NULL);
553 ast_test_validate(test, NULL != test_data);
554 ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
555 test_message = stasis_message_create(test_message_type, test_data);
556
557 stasis_publish(topic, test_message);
558
559 actual_len = consumer_wait_for(consumer, 1);
560 ast_test_validate(test, 1 == actual_len);
561 actual = stasis_message_data(consumer->messages_rxed[0]);
562 ast_test_validate(test, test_data == actual);
563
564 return AST_TEST_PASS;
565}
stasis_subscribe_pool
#define stasis_subscribe_pool(topic, callback, data)
Definition stasis.h:680

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), consumer::messages_rxed, NULL, RAII_VAR, stasis_message_create(), stasis_message_data(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe_pool, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, and stasis_subscription_change::topic.

◆ AST_TEST_DEFINE() [17/27]

AST_TEST_DEFINE ( publish_sync  )

Definition at line 471 of file test_stasis.c.

472{
473 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
474 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
475 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
476 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
477 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
478 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
479 int actual_len;
480 const char *actual;
481
482 switch (cmd) {
483 case TEST_INIT:
484 info->name = __func__;
485 info->category = test_category;
486 info->summary = "Test synchronous publishing";
487 info->description = "Test synchronous publishing";
488 return AST_TEST_NOT_RUN;
489 case TEST_EXECUTE:
490 break;
491 }
492
493 topic = stasis_topic_create("TestTopic");
494 ast_test_validate(test, NULL != topic);
495
496 consumer = consumer_create(1);
497 ast_test_validate(test, NULL != consumer);
498
499 uut = stasis_subscribe(topic, consumer_exec_sync, consumer);
500 ast_test_validate(test, NULL != uut);
501 ao2_ref(consumer, +1);
502
503 test_data = ao2_alloc(1, NULL);
504 ast_test_validate(test, NULL != test_data);
505 ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
506 test_message = stasis_message_create(test_message_type, test_data);
507
508 stasis_publish_sync(uut, test_message);
509
510 actual_len = consumer->messages_rxed_len;
511 ast_test_validate(test, 1 == actual_len);
512 actual = stasis_message_data(consumer->messages_rxed[0]);
513 ast_test_validate(test, test_data == actual);
514
515 return AST_TEST_PASS;
516}
stasis_publish_sync
void stasis_publish_sync(struct stasis_subscription *sub, struct stasis_message *message)
Publish a message to a topic's subscribers, synchronizing on the specified subscriber.
Definition stasis.c:1583
consumer_exec_sync
static void consumer_exec_sync(void *data, struct stasis_subscription *sub, struct stasis_message *message)
Definition test_stasis.c:226

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec_sync(), consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, stasis_message_create(), stasis_message_data(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish_sync(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, and stasis_subscription_change::topic.

◆ AST_TEST_DEFINE() [18/27]

AST_TEST_DEFINE ( router  )

Definition at line 1505 of file test_stasis.c.

1506{
1507 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1508 RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
1509 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
1510 RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
1511 RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
1512 RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
1513 RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
1514 RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
1515 RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
1516 RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
1517 RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
1518 RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
1519 int actual_len, ret;
1520 struct stasis_message *actual;
1521
1522 switch (cmd) {
1523 case TEST_INIT:
1524 info->name = __func__;
1525 info->category = test_category;
1526 info->summary = "Test simple message routing";
1527 info->description = "Test simple message routing";
1528 return AST_TEST_NOT_RUN;
1529 case TEST_EXECUTE:
1530 break;
1531 }
1532
1533 topic = stasis_topic_create("TestTopic");
1534 ast_test_validate(test, NULL != topic);
1535
1536 consumer1 = consumer_create(1);
1537 ast_test_validate(test, NULL != consumer1);
1538 consumer2 = consumer_create(1);
1539 ast_test_validate(test, NULL != consumer2);
1540 consumer3 = consumer_create(1);
1541 ast_test_validate(test, NULL != consumer3);
1542
1543 ast_test_validate(test, stasis_message_type_create("TestMessage1", NULL, &test_message_type1) == STASIS_MESSAGE_TYPE_SUCCESS);
1544 ast_test_validate(test, NULL != test_message_type1);
1545 ast_test_validate(test, stasis_message_type_create("TestMessage2", NULL, &test_message_type2) == STASIS_MESSAGE_TYPE_SUCCESS);
1546 ast_test_validate(test, NULL != test_message_type2);
1547 ast_test_validate(test, stasis_message_type_create("TestMessage3", NULL, &test_message_type3) == STASIS_MESSAGE_TYPE_SUCCESS);
1548 ast_test_validate(test, NULL != test_message_type3);
1549
1550 uut = stasis_message_router_create(topic);
1551 ast_test_validate(test, NULL != uut);
1552
1553 ret = stasis_message_router_add(
1554 uut, test_message_type1, consumer_exec, consumer1);
1555 ast_test_validate(test, 0 == ret);
1556 ao2_ref(consumer1, +1);
1557 ret = stasis_message_router_add(
1558 uut, test_message_type2, consumer_exec, consumer2);
1559 ast_test_validate(test, 0 == ret);
1560 ao2_ref(consumer2, +1);
1561 ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
1562 ast_test_validate(test, 0 == ret);
1563 ao2_ref(consumer3, +1);
1564
1565 test_data = ao2_alloc(1, NULL);
1566 ast_test_validate(test, NULL != test_data);
1567 test_message1 = stasis_message_create(test_message_type1, test_data);
1568 ast_test_validate(test, NULL != test_message1);
1569 test_message2 = stasis_message_create(test_message_type2, test_data);
1570 ast_test_validate(test, NULL != test_message2);
1571 test_message3 = stasis_message_create(test_message_type3, test_data);
1572 ast_test_validate(test, NULL != test_message3);
1573
1574 stasis_publish(topic, test_message1);
1575 stasis_publish(topic, test_message2);
1576 stasis_publish(topic, test_message3);
1577
1578 actual_len = consumer_wait_for(consumer1, 1);
1579 ast_test_validate(test, 1 == actual_len);
1580 actual_len = consumer_wait_for(consumer2, 1);
1581 ast_test_validate(test, 1 == actual_len);
1582 actual_len = consumer_wait_for(consumer3, 1);
1583 ast_test_validate(test, 1 == actual_len);
1584
1585 actual = consumer1->messages_rxed[0];
1586 ast_test_validate(test, test_message1 == actual);
1587
1588 actual = consumer2->messages_rxed[0];
1589 ast_test_validate(test, test_message2 == actual);
1590
1591 actual = consumer3->messages_rxed[0];
1592 ast_test_validate(test, test_message3 == actual);
1593
1594 /* consumer1 and consumer2 do not get the final message. */
1595 ao2_cleanup(consumer1);
1596 ao2_cleanup(consumer2);
1597
1598 return AST_TEST_PASS;
1599}
stasis_message_router_create
#define stasis_message_router_create(topic)
Create a new message router object.
Definition stasis_message_router.h:58
stasis_message_router_add
int stasis_message_router_add(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
Add a route to a message router.
Definition stasis_message_router.c:308
stasis_message_router_unsubscribe_and_join
void stasis_message_router_unsubscribe_and_join(struct stasis_message_router *router)
Unsubscribe the router from the upstream topic, blocking until the final message has been processed.
Definition stasis_message_router.c:267
stasis_message_router_set_default
int stasis_message_router_set_default(struct stasis_message_router *router, stasis_subscription_cb callback, void *data)
Sets the default route of a router.
Definition stasis_message_router.c:385
stasis_message_router
Definition stasis_message_router.c:129

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_message_create(), stasis_message_router_add(), stasis_message_router_create, stasis_message_router_set_default(), stasis_message_router_unsubscribe_and_join(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_topic_create(), test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [19/27]

AST_TEST_DEFINE ( router_cache_updates  )

Definition at line 1710 of file test_stasis.c.

1711{
1712 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1713 RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
1714 RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe_and_join);
1715 RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
1716 RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
1717 RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
1718 RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
1719 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
1720 RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
1721 RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
1722 RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
1723 RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
1724 RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
1725 RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
1726 RAII_VAR(struct stasis_message *, message1, NULL, ao2_cleanup);
1727 RAII_VAR(struct stasis_message *, message2, NULL, ao2_cleanup);
1728 struct stasis_cache_update *update;
1729 int actual_len, ret;
1730 struct stasis_message *actual;
1731
1732 switch (cmd) {
1733 case TEST_INIT:
1734 info->name = __func__;
1735 info->category = test_category;
1736 info->summary = "Test special handling cache_update messages";
1737 info->description = "Test special handling cache_update messages";
1738 return AST_TEST_NOT_RUN;
1739 case TEST_EXECUTE:
1740 break;
1741 }
1742
1743 topic = stasis_topic_create("TestTopic");
1744 ast_test_validate(test, NULL != topic);
1745
1746 cache = stasis_cache_create(cache_simple);
1747 ast_test_validate(test, NULL != cache);
1748 caching_topic = stasis_caching_topic_create(topic, cache);
1749 ast_test_validate(test, NULL != caching_topic);
1750
1751 consumer1 = consumer_create(1);
1752 ast_test_validate(test, NULL != consumer1);
1753 consumer2 = consumer_create(1);
1754 ast_test_validate(test, NULL != consumer2);
1755 consumer3 = consumer_create(1);
1756 ast_test_validate(test, NULL != consumer3);
1757
1758 ast_test_validate(test, stasis_message_type_create("Cache1", NULL, &test_message_type1) == STASIS_MESSAGE_TYPE_SUCCESS);
1759 ast_test_validate(test, NULL != test_message_type1);
1760 ast_test_validate(test, stasis_message_type_create("Cache2", NULL, &test_message_type2) == STASIS_MESSAGE_TYPE_SUCCESS);
1761 ast_test_validate(test, NULL != test_message_type2);
1762 ast_test_validate(test, stasis_message_type_create("NonCache", NULL, &test_message_type3) == STASIS_MESSAGE_TYPE_SUCCESS);
1763 ast_test_validate(test, NULL != test_message_type3);
1764
1765 uut = stasis_message_router_create(
1766 stasis_caching_get_topic(caching_topic));
1767 ast_test_validate(test, NULL != uut);
1768
1769 ret = stasis_message_router_add_cache_update(
1770 uut, test_message_type1, consumer_exec, consumer1);
1771 ast_test_validate(test, 0 == ret);
1772 ao2_ref(consumer1, +1);
1773 ret = stasis_message_router_add(
1774 uut, stasis_cache_update_type(), consumer_exec, consumer2);
1775 ast_test_validate(test, 0 == ret);
1776 ao2_ref(consumer2, +1);
1777 ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
1778 ast_test_validate(test, 0 == ret);
1779 ao2_ref(consumer3, +1);
1780
1781 test_data = ao2_alloc(1, NULL);
1782 ast_test_validate(test, NULL != test_data);
1783 test_message1 = stasis_message_create(test_message_type1, test_data);
1784 ast_test_validate(test, NULL != test_message1);
1785 test_message2 = stasis_message_create(test_message_type2, test_data);
1786 ast_test_validate(test, NULL != test_message2);
1787 test_message3 = stasis_message_create(test_message_type3, test_data);
1788 ast_test_validate(test, NULL != test_message3);
1789
1790 stasis_publish(topic, test_message1);
1791 stasis_publish(topic, test_message2);
1792 stasis_publish(topic, test_message3);
1793
1794 actual_len = consumer_wait_for(consumer1, 1);
1795 ast_test_validate(test, 1 == actual_len);
1796 actual_len = consumer_wait_for(consumer2, 1);
1797 ast_test_validate(test, 1 == actual_len);
1798 /* Uncacheable message should not be passed through */
1799 actual_len = consumer_should_stay(consumer3, 0);
1800 ast_test_validate(test, 0 == actual_len);
1801
1802 actual = consumer1->messages_rxed[0];
1803 ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
1804 update = stasis_message_data(actual);
1805 ast_test_validate(test, test_message_type1 == update->type);
1806 ast_test_validate(test, test_message1 == update->new_snapshot);
1807
1808 actual = consumer2->messages_rxed[0];
1809 ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
1810 update = stasis_message_data(actual);
1811 ast_test_validate(test, test_message_type2 == update->type);
1812 ast_test_validate(test, test_message2 == update->new_snapshot);
1813
1814 /* consumer1 and consumer2 do not get the final message. */
1815 ao2_cleanup(consumer1);
1816 ao2_cleanup(consumer2);
1817
1818 return AST_TEST_PASS;
1819}
update
static void update(int code_size, int y, int wi, int fi, int dq, int sr, int dqsez, struct g726_state *state_ptr)
Definition codec_g726.c:367
stasis_caching_unsubscribe_and_join
struct stasis_caching_topic * stasis_caching_unsubscribe_and_join(struct stasis_caching_topic *caching_topic)
Unsubscribes a caching topic from its upstream topic, blocking until all messages have been forwarded...
Definition stasis_cache.c:146
stasis_message_router_add_cache_update
int stasis_message_router_add_cache_update(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
Add a route for stasis_cache_update messages to a message router.
Definition stasis_message_router.c:334
cache_simple
static const char * cache_simple(struct stasis_message *message)
Definition test_stasis.c:1699

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, cache, cache_simple(), consumer_create(), consumer_exec(), consumer_should_stay(), consumer_wait_for(), NULL, RAII_VAR, stasis_cache_create(), stasis_cache_update_type(), stasis_caching_get_topic(), stasis_caching_topic_create(), stasis_caching_unsubscribe_and_join(), stasis_message_create(), stasis_message_data(), stasis_message_router_add(), stasis_message_router_add_cache_update(), stasis_message_router_create, stasis_message_router_set_default(), stasis_message_router_unsubscribe_and_join(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_topic_create(), test_category, TEST_EXECUTE, TEST_INIT, and update().

◆ AST_TEST_DEFINE() [20/27]

AST_TEST_DEFINE ( router_pool  )

Definition at line 1601 of file test_stasis.c.

1602{
1603 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
1604 RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
1605 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
1606 RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
1607 RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
1608 RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
1609 RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
1610 RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
1611 RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
1612 RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
1613 RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
1614 RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
1615 int actual_len, ret;
1616 struct stasis_message *actual;
1617
1618 switch (cmd) {
1619 case TEST_INIT:
1620 info->name = __func__;
1621 info->category = test_category;
1622 info->summary = "Test message routing via taskpool";
1623 info->description = "Test simple message routing when\n"
1624 "the subscriptions dictate usage of the Stasis\n"
1625 "taskpool.";
1626 return AST_TEST_NOT_RUN;
1627 case TEST_EXECUTE:
1628 break;
1629 }
1630
1631 topic = stasis_topic_create("TestTopic");
1632 ast_test_validate(test, NULL != topic);
1633
1634 consumer1 = consumer_create(1);
1635 ast_test_validate(test, NULL != consumer1);
1636 consumer2 = consumer_create(1);
1637 ast_test_validate(test, NULL != consumer2);
1638 consumer3 = consumer_create(1);
1639 ast_test_validate(test, NULL != consumer3);
1640
1641 ast_test_validate(test, stasis_message_type_create("TestMessage1", NULL, &test_message_type1) == STASIS_MESSAGE_TYPE_SUCCESS);
1642 ast_test_validate(test, NULL != test_message_type1);
1643 ast_test_validate(test, stasis_message_type_create("TestMessage2", NULL, &test_message_type2) == STASIS_MESSAGE_TYPE_SUCCESS);
1644 ast_test_validate(test, NULL != test_message_type2);
1645 ast_test_validate(test, stasis_message_type_create("TestMessage3", NULL, &test_message_type3) == STASIS_MESSAGE_TYPE_SUCCESS);
1646 ast_test_validate(test, NULL != test_message_type3);
1647
1648 uut = stasis_message_router_create_pool(topic);
1649 ast_test_validate(test, NULL != uut);
1650
1651 ret = stasis_message_router_add(
1652 uut, test_message_type1, consumer_exec, consumer1);
1653 ast_test_validate(test, 0 == ret);
1654 ao2_ref(consumer1, +1);
1655 ret = stasis_message_router_add(
1656 uut, test_message_type2, consumer_exec, consumer2);
1657 ast_test_validate(test, 0 == ret);
1658 ao2_ref(consumer2, +1);
1659 ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
1660 ast_test_validate(test, 0 == ret);
1661 ao2_ref(consumer3, +1);
1662
1663 test_data = ao2_alloc(1, NULL);
1664 ast_test_validate(test, NULL != test_data);
1665 test_message1 = stasis_message_create(test_message_type1, test_data);
1666 ast_test_validate(test, NULL != test_message1);
1667 test_message2 = stasis_message_create(test_message_type2, test_data);
1668 ast_test_validate(test, NULL != test_message2);
1669 test_message3 = stasis_message_create(test_message_type3, test_data);
1670 ast_test_validate(test, NULL != test_message3);
1671
1672 stasis_publish(topic, test_message1);
1673 stasis_publish(topic, test_message2);
1674 stasis_publish(topic, test_message3);
1675
1676 actual_len = consumer_wait_for(consumer1, 1);
1677 ast_test_validate(test, 1 == actual_len);
1678 actual_len = consumer_wait_for(consumer2, 1);
1679 ast_test_validate(test, 1 == actual_len);
1680 actual_len = consumer_wait_for(consumer3, 1);
1681 ast_test_validate(test, 1 == actual_len);
1682
1683 actual = consumer1->messages_rxed[0];
1684 ast_test_validate(test, test_message1 == actual);
1685
1686 actual = consumer2->messages_rxed[0];
1687 ast_test_validate(test, test_message2 == actual);
1688
1689 actual = consumer3->messages_rxed[0];
1690 ast_test_validate(test, test_message3 == actual);
1691
1692 /* consumer1 and consumer2 do not get the final message. */
1693 ao2_cleanup(consumer1);
1694 ao2_cleanup(consumer2);
1695
1696 return AST_TEST_PASS;
1697}
stasis_message_router_create_pool
#define stasis_message_router_create_pool(topic)
Create a new message router object.
Definition stasis_message_router.h:75

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_message_create(), stasis_message_router_add(), stasis_message_router_create_pool, stasis_message_router_set_default(), stasis_message_router_unsubscribe_and_join(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_topic_create(), test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [21/27]

AST_TEST_DEFINE ( subscription_interleaving  )

Definition at line 762 of file test_stasis.c.

763{
764 RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
765 RAII_VAR(struct stasis_topic *, topic1, NULL, ao2_cleanup);
766 RAII_VAR(struct stasis_topic *, topic2, NULL, ao2_cleanup);
767
768 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
769
770 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
771
772 RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
773 RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
774 RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
775
776 RAII_VAR(struct stasis_forward *, forward_sub1, NULL, stasis_forward_cancel);
777 RAII_VAR(struct stasis_forward *, forward_sub2, NULL, stasis_forward_cancel);
778 RAII_VAR(struct stasis_subscription *, sub1, NULL, stasis_unsubscribe);
779 RAII_VAR(struct stasis_subscription *, sub2, NULL, stasis_unsubscribe);
780
781 RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
782 RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
783
784 int actual_len;
785
786 switch (cmd) {
787 case TEST_INIT:
788 info->name = __func__;
789 info->category = test_category;
790 info->summary = "Test sending interleaved events to a parent topic with different subscribers";
791 info->description = "Test sending events to a parent topic.\n"
792 "This test creates three topics (one parent, two children)\n"
793 "and publishes messages alternately between the children.\n"
794 "It verifies that the messages are received in the expected\n"
795 "order, for different subscription types: one with a dedicated\n"
796 "thread, the other on the Stasis taskpool.";
797 return AST_TEST_NOT_RUN;
798 case TEST_EXECUTE:
799 break;
800 }
801
802 ast_test_validate(test, stasis_message_type_create("test", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
803 ast_test_validate(test, NULL != test_message_type);
804
805 test_data = ao2_alloc(1, NULL);
806 ast_test_validate(test, NULL != test_data);
807
808 test_message1 = stasis_message_create(test_message_type, test_data);
809 ast_test_validate(test, NULL != test_message1);
810 test_message2 = stasis_message_create(test_message_type, test_data);
811 ast_test_validate(test, NULL != test_message2);
812 test_message3 = stasis_message_create(test_message_type, test_data);
813 ast_test_validate(test, NULL != test_message3);
814
815 parent_topic = stasis_topic_create("ParentTestTopic");
816 ast_test_validate(test, NULL != parent_topic);
817 topic1 = stasis_topic_create("Topic1");
818 ast_test_validate(test, NULL != topic1);
819 topic2 = stasis_topic_create("Topic2");
820 ast_test_validate(test, NULL != topic2);
821
822 forward_sub1 = stasis_forward_all(topic1, parent_topic);
823 ast_test_validate(test, NULL != forward_sub1);
824 forward_sub2 = stasis_forward_all(topic2, parent_topic);
825 ast_test_validate(test, NULL != forward_sub2);
826
827 consumer1 = consumer_create(1);
828 ast_test_validate(test, NULL != consumer1);
829
830 consumer2 = consumer_create(1);
831 ast_test_validate(test, NULL != consumer2);
832
833 sub1 = stasis_subscribe(parent_topic, consumer_exec, consumer1);
834 ast_test_validate(test, NULL != sub1);
835 ao2_ref(consumer1, +1);
836
837 sub2 = stasis_subscribe_pool(parent_topic, consumer_exec, consumer2);
838 ast_test_validate(test, NULL != sub2);
839 ao2_ref(consumer2, +1);
840
841 stasis_publish(topic1, test_message1);
842 stasis_publish(topic2, test_message2);
843 stasis_publish(topic1, test_message3);
844
845 actual_len = consumer_wait_for(consumer1, 3);
846 ast_test_validate(test, 3 == actual_len);
847
848 actual_len = consumer_wait_for(consumer2, 3);
849 ast_test_validate(test, 3 == actual_len);
850
851 ast_test_validate(test, test_message1 == consumer1->messages_rxed[0]);
852 ast_test_validate(test, test_message2 == consumer1->messages_rxed[1]);
853 ast_test_validate(test, test_message3 == consumer1->messages_rxed[2]);
854
855 ast_test_validate(test, test_message1 == consumer2->messages_rxed[0]);
856 ast_test_validate(test, test_message2 == consumer2->messages_rxed[1]);
857 ast_test_validate(test, test_message3 == consumer2->messages_rxed[2]);
858
859 return AST_TEST_PASS;
860}

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for(), NULL, RAII_VAR, stasis_forward_all(), stasis_forward_cancel(), stasis_message_create(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_subscribe_pool, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [22/27]

AST_TEST_DEFINE ( subscription_messages  )

Definition at line 314 of file test_stasis.c.

315{
316 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
317 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
318 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
319 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
320 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
321 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
322 RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
323 int complete;
324 struct stasis_subscription_change *change;
325
326 switch (cmd) {
327 case TEST_INIT:
328 info->name = __func__;
329 info->category = test_category;
330 info->summary = "Test subscribe/unsubscribe messages";
331 info->description = "Test subscribe/unsubscribe messages";
332 return AST_TEST_NOT_RUN;
333 case TEST_EXECUTE:
334 break;
335 }
336
337 topic = stasis_topic_create("TestTopic");
338 ast_test_validate(test, NULL != topic);
339
340 consumer = consumer_create(0);
341 ast_test_validate(test, NULL != consumer);
342
343 uut = stasis_subscribe(topic, consumer_exec, consumer);
344 ast_test_validate(test, NULL != uut);
345 ao2_ref(consumer, +1);
346 expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
347
348 uut = stasis_unsubscribe(uut);
349 complete = consumer_wait_for_completion(consumer);
350 ast_test_validate(test, 1 == complete);
351
352 ast_test_validate(test, 2 == consumer->messages_rxed_len);
353 ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[0]));
354 ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[1]));
355
356 change = stasis_message_data(consumer->messages_rxed[0]);
357 ast_test_validate(test, topic == change->topic);
358 ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
359 ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
360
361 change = stasis_message_data(consumer->messages_rxed[1]);
362 ast_test_validate(test, topic == change->topic);
363 ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
364 ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
365
366 return AST_TEST_PASS;
367}
ast_free
#define ast_free(a)
Definition astmm.h:180
ast_strdup
#define ast_strdup(str)
A wrapper for strdup()
Definition astmm.h:241
stasis_subscription_uniqueid
const char * stasis_subscription_uniqueid(const struct stasis_subscription *sub)
Get the unique ID for the subscription.
Definition stasis.c:1236
stasis_subscription_change
Holds details about changes to subscriptions for the specified topic.
Definition stasis.h:890
stasis_subscription_change::topic
struct stasis_topic * topic
Definition stasis.h:891
stasis_subscription_change::description
char description[0]
Definition stasis.h:893
stasis_subscription_change::uniqueid
char * uniqueid
Definition stasis.h:892

References ao2_cleanup, ao2_ref, ast_free, ast_strdup, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for_completion(), stasis_subscription_change::description, consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, stasis_message_data(), stasis_subscribe, stasis_subscription_change_type(), stasis_subscription_uniqueid(), stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, stasis_subscription_change::topic, and stasis_subscription_change::uniqueid.

◆ AST_TEST_DEFINE() [23/27]

AST_TEST_DEFINE ( subscription_pool_messages  )

Definition at line 369 of file test_stasis.c.

370{
371 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
372 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
373 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
374 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
375 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
376 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
377 RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
378 int complete;
379 struct stasis_subscription_change *change;
380
381 switch (cmd) {
382 case TEST_INIT:
383 info->name = __func__;
384 info->category = test_category;
385 info->summary = "Test subscribe/unsubscribe messages using a taskpool subscription";
386 info->description = "Test subscribe/unsubscribe messages using a taskpool subscription";
387 return AST_TEST_NOT_RUN;
388 case TEST_EXECUTE:
389 break;
390 }
391
392 topic = stasis_topic_create("TestTopic");
393 ast_test_validate(test, NULL != topic);
394
395 consumer = consumer_create(0);
396 ast_test_validate(test, NULL != consumer);
397
398 uut = stasis_subscribe_pool(topic, consumer_exec, consumer);
399 ast_test_validate(test, NULL != uut);
400 ao2_ref(consumer, +1);
401 expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
402
403 uut = stasis_unsubscribe(uut);
404 complete = consumer_wait_for_completion(consumer);
405 ast_test_validate(test, 1 == complete);
406
407 ast_test_validate(test, 2 == consumer->messages_rxed_len);
408 ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[0]));
409 ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[1]));
410
411 change = stasis_message_data(consumer->messages_rxed[0]);
412 ast_test_validate(test, topic == change->topic);
413 ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
414 ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
415
416 change = stasis_message_data(consumer->messages_rxed[1]);
417 ast_test_validate(test, topic == change->topic);
418 ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
419 ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
420
421 return AST_TEST_PASS;
422}

References ao2_cleanup, ao2_ref, ast_free, ast_strdup, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_wait_for_completion(), stasis_subscription_change::description, consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, stasis_message_data(), stasis_subscribe_pool, stasis_subscription_change_type(), stasis_subscription_uniqueid(), stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, stasis_subscription_change::topic, and stasis_subscription_change::uniqueid.

◆ AST_TEST_DEFINE() [24/27]

AST_TEST_DEFINE ( to_ami  )

Definition at line 1929 of file test_stasis.c.

1930{
1931 RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
1932 RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
1933 RAII_VAR(char *, data, NULL, ao2_cleanup);
1934 RAII_VAR(struct ast_manager_event_blob *, actual, NULL, ao2_cleanup);
1935 const char *expected_text = "SomeData";
1936 const char *expected = "Message: SomeData\r\n";
1937
1938 switch (cmd) {
1939 case TEST_INIT:
1940 info->name = __func__;
1941 info->category = test_category;
1942 info->summary = "Test message to_ami function";
1943 info->description = "Test message to_ami function";
1944 return AST_TEST_NOT_RUN;
1945 case TEST_EXECUTE:
1946 break;
1947 }
1948
1949 ast_test_validate(test, stasis_message_type_create("SomeMessage", &fake_vtable, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
1950
1951 data = ao2_alloc(strlen(expected_text) + 1, NULL);
1952 strcpy(data, expected_text);
1953 uut = stasis_message_create(type, data);
1954 ast_test_validate(test, NULL != uut);
1955
1956 actual = stasis_message_to_ami(uut);
1957 ast_test_validate(test, strcmp(expected, actual->extra_fields) == 0);
1958
1959 return AST_TEST_PASS;
1960}
fake_vtable
static struct stasis_message_vtable fake_vtable
Definition test_stasis.c:72

References ao2_alloc, ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, stasis_message::data, fake_vtable, NULL, RAII_VAR, stasis_message_create(), stasis_message_to_ami(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, TEST_INIT, and type.

◆ AST_TEST_DEFINE() [25/27]

AST_TEST_DEFINE ( to_json  )

Definition at line 1858 of file test_stasis.c.

1859{
1860 RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
1861 RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
1862 RAII_VAR(char *, data, NULL, ao2_cleanup);
1863 RAII_VAR(struct ast_json *, actual, NULL, ast_json_unref);
1864 const char *expected_text = "SomeData";
1865 RAII_VAR(struct ast_json *, expected, NULL, ast_json_unref);
1866
1867 switch (cmd) {
1868 case TEST_INIT:
1869 info->name = __func__;
1870 info->category = test_category;
1871 info->summary = "Test message to_json function when NULL";
1872 info->description = "Test message to_json function when NULL";
1873 return AST_TEST_NOT_RUN;
1874 case TEST_EXECUTE:
1875 break;
1876 }
1877
1878 ast_test_validate(test, stasis_message_type_create("SomeMessage", &fake_vtable, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
1879
1880 data = ao2_alloc(strlen(expected_text) + 1, NULL);
1881 strcpy(data, expected_text);
1882 uut = stasis_message_create(type, data);
1883 ast_test_validate(test, NULL != uut);
1884
1885 expected = ast_json_string_create(expected_text);
1886 actual = stasis_message_to_json(uut, NULL);
1887 ast_test_validate(test, ast_json_equal(expected, actual));
1888
1889 return AST_TEST_PASS;
1890}
ast_json_string_create
struct ast_json * ast_json_string_create(const char *value)
Construct a JSON string from value.
Definition json.c:278
ast_json_equal
int ast_json_equal(const struct ast_json *lhs, const struct ast_json *rhs)
Compare two JSON objects.
Definition json.c:357

References ao2_alloc, ao2_cleanup, ast_json_equal(), ast_json_string_create(), ast_json_unref(), AST_TEST_NOT_RUN, AST_TEST_PASS, stasis_message::data, fake_vtable, NULL, RAII_VAR, stasis_message_create(), stasis_message_to_json(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, test_category, TEST_EXECUTE, TEST_INIT, and type.

◆ AST_TEST_DEFINE() [26/27]

AST_TEST_DEFINE ( type_filters  )

Definition at line 2234 of file test_stasis.c.

2235{
2236 RAII_VAR(struct cts *, cts, NULL, ao2_cleanup);
2237 RAII_VAR(struct test_message_types *, types, NULL, ao2_cleanup);
2238 int ix = 0;
2239
2240 switch (cmd) {
2241 case TEST_INIT:
2242 info->name = __func__;
2243 info->category = test_category "filtering/";
2244 info->summary = "Test message filtering by type";
2245 info->description = "Test message filtering by type";
2246 return AST_TEST_NOT_RUN;
2247 case TEST_EXECUTE:
2248 break;
2249 }
2250
2251 types = create_message_types(test);
2252 ast_test_validate(test, NULL != types);
2253
2254 cts = create_cts(test);
2255 ast_test_validate(test, NULL != cts);
2256
2257 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->type1) == 0);
2258 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->type2) == 0);
2259 ast_test_validate(test, stasis_subscription_accept_message_type(cts->sub, types->change) == 0);
2260 ast_test_validate(test, stasis_subscription_set_filter(cts->sub, STASIS_SUBSCRIPTION_FILTER_SELECTIVE) == 0);
2261
2262 /* We should get these */
2263 ast_test_validate(test, send_msg(test, cts, types->type1, "Pass"));
2264 ast_test_validate(test, send_msg(test, cts, types->type2, "Pass"));
2265 /* ... but not this one */
2266 ast_test_validate(test, send_msg(test, cts, types->type3, "FAIL"));
2267
2268 /* Wait for change(subscribe) and "Pass" messages */
2269 consumer_wait_for(cts->consumer, 3);
2270
2271 /* Remove type 1 */
2272 ast_test_validate(test, stasis_subscription_decline_message_type(cts->sub, types->type1) == 0);
2273
2274 /* We should now NOT get this one */
2275 ast_test_validate(test, send_msg(test, cts, types->type1, "FAIL"));
2276 /* We should get this one (again) */
2277 ast_test_validate(test, send_msg(test, cts, types->type2, "Pass2"));
2278 /* We still should NOT get this one */
2279 ast_test_validate(test, send_msg(test, cts, types->type3, "FAIL"));
2280
2281 /* We should now have a second type2 */
2282 consumer_wait_for(cts->consumer, 4);
2283
2284 stasis_unsubscribe(cts->sub);
2285 cts->sub = NULL;
2286 consumer_wait_for_completion(cts->consumer);
2287
2288 dump_consumer(test, cts);
2289
2290 ast_test_validate(test, 1 == cts->consumer->complete);
2291 ast_test_validate(test, 5 == cts->consumer->messages_rxed_len);
2292 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Subscribe"));
2293 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->type1, "Pass"));
2294 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->type2, "Pass"));
2295 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->type2, "Pass2"));
2296 ast_test_validate(test, is_msg(cts->consumer->messages_rxed[ix++], types->change, "Unsubscribe"));
2297
2298 return AST_TEST_PASS;
2299}
stasis_subscription_decline_message_type
int stasis_subscription_decline_message_type(struct stasis_subscription *subscription, const struct stasis_message_type *type)
Indicate to a subscription that we are not interested in a message type.
Definition stasis.c:1120

References ao2_cleanup, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer::complete, cts::consumer, consumer_wait_for(), consumer_wait_for_completion(), create_cts(), create_message_types(), dump_consumer(), is_msg(), consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, send_msg(), stasis_subscription_accept_message_type(), stasis_subscription_decline_message_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_set_filter(), stasis_unsubscribe(), cts::sub, test_category, TEST_EXECUTE, and TEST_INIT.

◆ AST_TEST_DEFINE() [27/27]

AST_TEST_DEFINE ( unsubscribe_stops_messages  )

Definition at line 567 of file test_stasis.c.

568{
569 RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
570 RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
571 RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
572 RAII_VAR(char *, test_data, NULL, ao2_cleanup);
573 RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
574 RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
575 int actual_len;
576
577 switch (cmd) {
578 case TEST_INIT:
579 info->name = __func__;
580 info->category = test_category;
581 info->summary = "Test simple subscriptions";
582 info->description = "Test simple subscriptions";
583 return AST_TEST_NOT_RUN;
584 case TEST_EXECUTE:
585 break;
586 }
587
588 topic = stasis_topic_create("TestTopic");
589 ast_test_validate(test, NULL != topic);
590
591 consumer = consumer_create(1);
592 ast_test_validate(test, NULL != consumer);
593
594 uut = stasis_subscribe(topic, consumer_exec, consumer);
595 ast_test_validate(test, NULL != uut);
596 ao2_ref(consumer, +1);
597
598 uut = stasis_unsubscribe(uut);
599
600 test_data = ao2_alloc(1, NULL);
601 ast_test_validate(test, NULL != test_data);
602 ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
603 test_message = stasis_message_create(test_message_type, test_data);
604
605 stasis_publish(topic, test_message);
606
607 actual_len = consumer_should_stay(consumer, 0);
608 ast_test_validate(test, 0 == actual_len);
609
610 return AST_TEST_PASS;
611}

References ao2_alloc, ao2_cleanup, ao2_ref, AST_TEST_NOT_RUN, AST_TEST_PASS, consumer_create(), consumer_exec(), consumer_should_stay(), NULL, RAII_VAR, stasis_message_create(), stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_publish(), stasis_subscribe, stasis_topic_create(), stasis_unsubscribe(), test_category, TEST_EXECUTE, TEST_INIT, and stasis_subscription_change::topic.

◆ cache_simple()

static const char * cache_simple ( struct stasis_message *  message)
static

Definition at line 1699 of file test_stasis.c.

1700{
1701 const char *type_name =
1702 stasis_message_type_name(stasis_message_type(message));
1703 if (!ast_begins_with(type_name, "Cache")) {
1704 return NULL;
1705 }
1706
1707 return "cached";
1708}
ast_begins_with
static int force_inline attribute_pure ast_begins_with(const char *str, const char *prefix)
Checks whether a string begins with another.
Definition strings.h:97

References ast_begins_with(), NULL, and stasis_message_type_name().

Referenced by AST_TEST_DEFINE().

◆ cache_test_aggregate_calc_fn()

static struct stasis_message * cache_test_aggregate_calc_fn ( struct stasis_cache_entry *  entry,
struct stasis_message *  new_snapshot 
)
static

Definition at line 911 of file test_stasis.c.

912{
913 struct stasis_message *aggregate_snapshot;
914 struct stasis_message *snapshot;
915 struct stasis_message_type *type = NULL;
916 struct cache_test_data *test_data = NULL;
917 int idx;
918 int accumulated = 0;
919 char aggregate_str[30];
920
921 /* Accumulate the aggregate value. */
922 snapshot = stasis_cache_entry_get_local(entry);
923 if (snapshot) {
924 type = stasis_message_type(snapshot);
925 test_data = stasis_message_data(snapshot);
926 accumulated += atoi(test_data->value);
927 }
928 for (idx = 0; ; ++idx) {
929 snapshot = stasis_cache_entry_get_remote(entry, idx);
930 if (!snapshot) {
931 break;
932 }
933
934 type = stasis_message_type(snapshot);
935 test_data = stasis_message_data(snapshot);
936 accumulated += atoi(test_data->value);
937 }
938
939 if (!test_data) {
940 /* There are no test entries cached. Delete the aggregate. */
941 return NULL;
942 }
943
944 snapshot = stasis_cache_entry_get_aggregate(entry);
945 if (snapshot) {
946 type = stasis_message_type(snapshot);
947 test_data = stasis_message_data(snapshot);
948 if (accumulated == atoi(test_data->value)) {
949 /* Aggregate test entry did not change. */
950 return ao2_bump(snapshot);
951 }
952 }
953
954 snprintf(aggregate_str, sizeof(aggregate_str), "%d", accumulated);
955 aggregate_snapshot = cache_test_message_create_full(type, test_data->id, aggregate_str, NULL);
956 if (!aggregate_snapshot) {
957 /* Bummer. We have to keep the old aggregate snapshot. */
958 ast_log(LOG_ERROR, "Could not create aggregate snapshot.\n");
959 return ao2_bump(snapshot);
960 }
961
962 return aggregate_snapshot;
963}
ast_log
#define ast_log
Definition astobj2.c:42
ao2_bump
#define ao2_bump(obj)
Bump refcount on an AO2 object by one, returning the object.
Definition astobj2.h:480
LOG_ERROR
#define LOG_ERROR
Definition include/asterisk/logger.h:291
stasis_cache_entry_get_aggregate
struct stasis_message * stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
Get the aggregate cache entry snapshot.
Definition stasis_cache.c:365
stasis_cache_entry_get_local
struct stasis_message * stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
Get the local entity's cache entry snapshot.
Definition stasis_cache.c:370
stasis_cache_entry_get_remote
struct stasis_message * stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx)
Get a remote entity's cache entry snapshot by index.
Definition stasis_cache.c:375
cache_test_data
Definition test_stasis.c:862

References ao2_bump, ast_log, cache_test_message_create_full(), LOG_ERROR, NULL, stasis_cache_entry_get_aggregate(), stasis_cache_entry_get_local(), stasis_cache_entry_get_remote(), stasis_message_data(), and type.

Referenced by AST_TEST_DEFINE().

◆ cache_test_aggregate_publish_fn()

static void cache_test_aggregate_publish_fn ( struct stasis_topic *  topic,
struct stasis_message *  aggregate 
)
static

Definition at line 965 of file test_stasis.c.

966{
967 stasis_publish(topic, aggregate);
968}

References stasis_publish().

Referenced by AST_TEST_DEFINE().

◆ cache_test_data_dtor()

static void cache_test_data_dtor ( void *  obj)
static

Definition at line 867 of file test_stasis.c.

868{
869 struct cache_test_data *data = obj;
870
871 ast_free(data->id);
872 ast_free(data->value);
873}
cache_test_data::value
char * value
Definition test_stasis.c:864
cache_test_data::id
char * id
Definition test_stasis.c:863

References ast_free, cache_test_data::id, and cache_test_data::value.

Referenced by cache_test_message_create_full().

◆ cache_test_data_id()

static const char * cache_test_data_id ( struct stasis_message *  message)
static

Definition at line 901 of file test_stasis.c.

902{
903 struct cache_test_data *cachable = stasis_message_data(message);
904
905 if (0 != strcmp("Cacheable", stasis_message_type_name(stasis_message_type(message)))) {
906 return NULL;
907 }
908 return cachable->id;
909}

References cache_test_data::id, NULL, stasis_message_data(), and stasis_message_type_name().

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ cache_test_message_create()

static struct stasis_message * cache_test_message_create ( struct stasis_message_type *  type,
const char *  name,
const char *  value 
)
static

Definition at line 896 of file test_stasis.c.

897{
898 return cache_test_message_create_full(type, name, value, &ast_eid_default);
899}
name
static const char name[]
Definition format_mp3.c:68
value
int value
Definition syslog.c:37

References ast_eid_default, cache_test_message_create_full(), name, type, and value.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ cache_test_message_create_full()

static struct stasis_message * cache_test_message_create_full ( struct stasis_message_type *  type,
const char *  name,
const char *  value,
struct ast_eid *  eid 
)
static

Definition at line 875 of file test_stasis.c.

876{
877 RAII_VAR(struct cache_test_data *, data, NULL, ao2_cleanup);
878
879 data = ao2_alloc(sizeof(*data), cache_test_data_dtor);
880 if (data == NULL) {
881 return NULL;
882 }
883
884 ast_assert(name != NULL);
885 ast_assert(value != NULL);
886
887 data->id = ast_strdup(name);
888 data->value = ast_strdup(value);
889 if (!data->id || !data->value) {
890 return NULL;
891 }
892
893 return stasis_message_create_full(type, data, eid);
894}
cache_test_data_dtor
static void cache_test_data_dtor(void *obj)
Definition test_stasis.c:867
ast_assert
#define ast_assert(a)
Definition utils.h:776

References ao2_alloc, ao2_cleanup, ast_assert, ast_strdup, cache_test_data_dtor(), stasis_message::data, stasis_message::eid, name, NULL, RAII_VAR, stasis_message_create_full(), type, and value.

Referenced by AST_TEST_DEFINE(), cache_test_aggregate_calc_fn(), and cache_test_message_create().

◆ check_cache_aggregate()

static int check_cache_aggregate ( struct stasis_cache *  cache,
struct stasis_message_type *  cache_type,
const char *  id,
const char *  value 
)
static

Definition at line 970 of file test_stasis.c.

971{
972 RAII_VAR(struct stasis_message *, aggregate, NULL, ao2_cleanup);
973 struct cache_test_data *test_data;
974
975 aggregate = stasis_cache_get_by_eid(cache, cache_type, id, NULL);
976 if (!aggregate) {
977 /* No aggregate, return true if given no value. */
978 return !value;
979 }
980
981 /* Return true if the given value matches the aggregate value. */
982 test_data = stasis_message_data(aggregate);
983 return value && !strcmp(value, test_data->value);
984}
stasis_cache_get_by_eid
struct stasis_message * stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid)
Retrieve an item from the cache for a specific entity.
Definition stasis_cache.c:659

References ao2_cleanup, cache, NULL, RAII_VAR, stasis_cache_get_by_eid(), stasis_message_data(), and value.

Referenced by AST_TEST_DEFINE().

◆ consumer_create()

static struct consumer * consumer_create ( int  ignore_subscriptions)
static

Definition at line 183 of file test_stasis.c.

184{
185 struct consumer *consumer;
186
187 consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
188 if (!consumer) {
189 return NULL;
190 }
191
192 consumer->ignore_subscriptions = ignore_subscriptions;
193 consumer->messages_rxed = ast_malloc(sizeof(*consumer->messages_rxed));
194 if (!consumer->messages_rxed) {
195 ao2_cleanup(consumer);
196 return NULL;
197 }
198
199 ast_cond_init(&consumer->out, NULL);
200
201 return consumer;
202}
ast_malloc
#define ast_malloc(len)
A wrapper for malloc()
Definition astmm.h:191
ast_cond_init
#define ast_cond_init(cond, attr)
Definition lock.h:208
consumer::out
ast_cond_t out
Definition test_devicestate.c:377
consumer::ignore_subscriptions
int ignore_subscriptions
Definition test_stasis.c:166
consumer_dtor
static void consumer_dtor(void *obj)
Definition test_stasis.c:170

References ao2_alloc, ao2_cleanup, ast_cond_init, ast_malloc, consumer_dtor(), consumer::ignore_subscriptions, consumer::messages_rxed, NULL, and consumer::out.

◆ consumer_dtor()

static void consumer_dtor ( void *  obj)
static

Definition at line 170 of file test_stasis.c.

171{
172 struct consumer *consumer = obj;
173
174 ast_cond_destroy(&consumer->out);
175
176 while (consumer->messages_rxed_len > 0) {
177 ao2_cleanup(consumer->messages_rxed[--consumer->messages_rxed_len]);
178 }
179 ast_free(consumer->messages_rxed);
180 consumer->messages_rxed = NULL;
181}
ast_cond_destroy
#define ast_cond_destroy(cond)
Definition lock.h:209

References ao2_cleanup, ast_cond_destroy, ast_free, consumer::messages_rxed, consumer::messages_rxed_len, NULL, and consumer::out.

Referenced by consumer_create().

◆ consumer_exec()

static void consumer_exec ( void *  data,
struct stasis_subscription *  sub,
struct stasis_message *  message 
)
static

Definition at line 204 of file test_stasis.c.

205{
206 struct consumer *consumer = data;
207 RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
208 SCOPED_AO2LOCK(lock, consumer);
209
210 if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change_type()) {
211 ++consumer->messages_rxed_len;
212 consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
213 ast_assert(consumer->messages_rxed != NULL);
214 consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
215 ao2_ref(message, +1);
216 }
217
218 if (stasis_subscription_final_message(sub, message)) {
219 consumer->complete = 1;
220 consumer_needs_cleanup = consumer;
221 }
222
223 ast_cond_signal(&consumer->out);
224}
lock
ast_mutex_t lock
Definition app_sla.c:337
ast_realloc
#define ast_realloc(p, len)
A wrapper for realloc()
Definition astmm.h:226
SCOPED_AO2LOCK
#define SCOPED_AO2LOCK(varname, obj)
scoped lock specialization for ao2 mutexes.
Definition lock.h:611
ast_cond_signal
#define ast_cond_signal(cond)
Definition lock.h:210
stasis_subscription_final_message
int stasis_subscription_final_message(struct stasis_subscription *sub, struct stasis_message *msg)
Determine whether a message is the final message to be received on a subscription.
Definition stasis.c:1241

References ao2_cleanup, ao2_ref, ast_assert, ast_cond_signal, ast_realloc, consumer::complete, consumer::ignore_subscriptions, lock, consumer::messages_rxed, consumer::messages_rxed_len, NULL, consumer::out, RAII_VAR, SCOPED_AO2LOCK, stasis_subscription_change_type(), stasis_subscription_final_message(), and sub.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and create_cts().

◆ consumer_exec_sync()

static void consumer_exec_sync ( void *  data,
struct stasis_subscription *  sub,
struct stasis_message *  message 
)
static

Definition at line 226 of file test_stasis.c.

227{
228 struct consumer *consumer = data;
229 RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
230 SCOPED_AO2LOCK(lock, consumer);
231
232 if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change_type()) {
233 ++consumer->messages_rxed_len;
234 consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
235 ast_assert(consumer->messages_rxed != NULL);
236 consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
237 ao2_ref(message, +1);
238 }
239
240 if (stasis_subscription_final_message(sub, message)) {
241 consumer->complete = 1;
242 consumer_needs_cleanup = consumer;
243 }
244}

References ao2_cleanup, ao2_ref, ast_assert, ast_realloc, consumer::complete, consumer::ignore_subscriptions, lock, consumer::messages_rxed, consumer::messages_rxed_len, NULL, RAII_VAR, SCOPED_AO2LOCK, stasis_subscription_change_type(), stasis_subscription_final_message(), and sub.

Referenced by AST_TEST_DEFINE().

◆ consumer_should_stay()

static int consumer_should_stay ( struct consumer *  consumer,
size_t  expected_len 
)
static

Definition at line 288 of file test_stasis.c.

289{
290 struct timeval start = ast_tvnow();
291 struct timeval diff = {
292 .tv_sec = 0,
293 .tv_usec = 100000 /* wait for 100ms */
294 };
295 struct timeval end_tv = ast_tvadd(start, diff);
296 struct timespec end = {
297 .tv_sec = end_tv.tv_sec,
298 .tv_nsec = end_tv.tv_usec * 1000
299 };
300
301 SCOPED_AO2LOCK(lock, consumer);
302
303 while (consumer->messages_rxed_len == expected_len) {
304 int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
305
306 if (r == ETIMEDOUT) {
307 break;
308 }
309 ast_assert(r == 0); /* Not expecting any other types of errors */
310 }
311 return consumer->messages_rxed_len;
312}
ao2_object_get_lockaddr
void * ao2_object_get_lockaddr(void *obj)
Return the mutex lock address of an object.
Definition astobj2.c:476
end
char * end
Definition eagi_proxy.c:73
ast_cond_timedwait
#define ast_cond_timedwait(cond, mutex, time)
Definition lock.h:213
ast_tvadd
struct timeval ast_tvadd(struct timeval a, struct timeval b)
Returns the sum of two timevals a + b.
Definition extconf.c:2280

References ao2_object_get_lockaddr(), ast_assert, ast_cond_timedwait, ast_tvadd(), ast_tvnow(), end, lock, consumer::messages_rxed_len, consumer::out, and SCOPED_AO2LOCK.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ consumer_wait_for()

static int consumer_wait_for ( struct consumer *  consumer,
size_t  expected_len 
)
static

Definition at line 246 of file test_stasis.c.

247{
248 struct timeval start = ast_tvnow();
249 struct timespec end = {
250 .tv_sec = start.tv_sec + 30,
251 .tv_nsec = start.tv_usec * 1000
252 };
253
254 SCOPED_AO2LOCK(lock, consumer);
255
256 while (consumer->messages_rxed_len < expected_len) {
257 int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
258
259 if (r == ETIMEDOUT) {
260 break;
261 }
262 ast_assert(r == 0); /* Not expecting any other types of errors */
263 }
264 return consumer->messages_rxed_len;
265}

References ao2_object_get_lockaddr(), ast_assert, ast_cond_timedwait, ast_tvnow(), end, lock, consumer::messages_rxed_len, consumer::out, and SCOPED_AO2LOCK.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ consumer_wait_for_completion()

static int consumer_wait_for_completion ( struct consumer *  consumer)
static

Definition at line 267 of file test_stasis.c.

268{
269 struct timeval start = ast_tvnow();
270 struct timespec end = {
271 .tv_sec = start.tv_sec + 3,
272 .tv_nsec = start.tv_usec * 1000
273 };
274
275 SCOPED_AO2LOCK(lock, consumer);
276
277 while (!consumer->complete) {
278 int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
279
280 if (r == ETIMEDOUT) {
281 break;
282 }
283 ast_assert(r == 0); /* Not expecting any other types of errors */
284 }
285 return consumer->complete;
286}

References ao2_object_get_lockaddr(), ast_assert, ast_cond_timedwait, ast_tvnow(), consumer::complete, end, lock, consumer::out, and SCOPED_AO2LOCK.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ create_cts()

static struct cts * create_cts ( struct ast_test *  test)
static

Definition at line 2154 of file test_stasis.c.

2155{
2156 struct cts *cts = ao2_alloc(sizeof(*cts), destroy_cts);
2157 enum ast_test_result_state __attribute__ ((unused)) rc;
2158
2159 ast_test_validate_cleanup(test, cts, rc, cleanup);
2160
2161 cts->topic = stasis_topic_create("TestTopic");
2162 ast_test_validate_cleanup(test, NULL != cts->topic, rc, cleanup);
2163
2164 cts->consumer = consumer_create(0);
2165 ast_test_validate_cleanup(test, NULL != cts->consumer, rc, cleanup);
2166
2167 ao2_ref(cts->consumer, +1);
2168 cts->sub = stasis_subscribe(cts->topic, consumer_exec, cts->consumer);
2169 ast_test_validate_cleanup(test, NULL != cts->sub, rc, cleanup);
2170
2171 return cts;
2172
2173cleanup:
2174 ao2_cleanup(cts);
2175 return NULL;
2176}
cleanup
static void cleanup(void)
Clean up any old apps that we don't need any more.
Definition res_stasis.c:327
cts::topic
struct stasis_topic * topic
Definition test_stasis.c:2141
ast_test_result_state
ast_test_result_state
Definition test.h:193
destroy_cts
static void destroy_cts(void *obj)
Definition test_stasis.c:2145

References ao2_alloc, ao2_cleanup, ao2_ref, cleanup(), cts::consumer, consumer_create(), consumer_exec(), destroy_cts(), NULL, stasis_subscribe, stasis_topic_create(), cts::sub, and cts::topic.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ create_message_types()

static struct test_message_types * create_message_types ( struct ast_test *  test)
static

Definition at line 2080 of file test_stasis.c.

2081{
2082 struct stasis_message_vtable vtable = { 0 };
2083 struct test_message_types *types;
2084 enum ast_test_result_state __attribute__ ((unused)) rc;
2085
2086 types = ao2_alloc(sizeof(*types), destroy_message_types);
2087 if (!types) {
2088 return NULL;
2089 }
2090
2091 ast_test_validate_cleanup(test,
2092 stasis_message_type_create("TestMessageNONE", &vtable, &types->none) == STASIS_MESSAGE_TYPE_SUCCESS,
2093 rc, cleanup);
2094
2095 vtable.to_ami = fake_ami;
2096 ast_test_validate_cleanup(test,
2097 stasis_message_type_create("TestMessageAMI", &vtable, &types->ami) == STASIS_MESSAGE_TYPE_SUCCESS,
2098 rc, cleanup);
2099
2100 vtable.to_ami = NULL;
2101 vtable.to_json = fake_json;
2102 ast_test_validate_cleanup(test,
2103 stasis_message_type_create("TestMessageJSON", &vtable, &types->json) == STASIS_MESSAGE_TYPE_SUCCESS,
2104 rc, cleanup);
2105
2106 vtable.to_ami = NULL;
2107 vtable.to_json = NULL;
2108 vtable.to_event = fake_event;
2109 ast_test_validate_cleanup(test,
2110 stasis_message_type_create("TestMessageEVENT", &vtable, &types->event) == STASIS_MESSAGE_TYPE_SUCCESS,
2111 rc, cleanup);
2112
2113 vtable.to_ami = fake_ami;
2114 ast_test_validate_cleanup(test,
2115 stasis_message_type_create("TestMessageAMIEVENT", &vtable, &types->amievent) == STASIS_MESSAGE_TYPE_SUCCESS,
2116 rc, cleanup);
2117
2118 ast_test_validate_cleanup(test,
2119 stasis_message_type_create("TestMessageType1", NULL, &types->type1) == STASIS_MESSAGE_TYPE_SUCCESS,
2120 rc, cleanup);
2121
2122 ast_test_validate_cleanup(test,
2123 stasis_message_type_create("TestMessageType2", NULL, &types->type2) == STASIS_MESSAGE_TYPE_SUCCESS,
2124 rc, cleanup);
2125
2126 ast_test_validate_cleanup(test,
2127 stasis_message_type_create("TestMessageType3", NULL, &types->type3) == STASIS_MESSAGE_TYPE_SUCCESS,
2128 rc, cleanup);
2129
2130 types->change = stasis_subscription_change_type();
2131
2132 return types;
2133
2134cleanup:
2135 ao2_cleanup(types);
2136 return NULL;
2137}
stasis_message_vtable
Virtual table providing methods for messages.
Definition stasis.h:239
stasis_message_vtable::to_ami
struct ast_manager_event_blob *(* to_ami)(struct stasis_message *message)
Build the AMI representation of the message.
Definition stasis.h:264
stasis_message_vtable::to_json
struct ast_json *(* to_json)(struct stasis_message *message, const struct stasis_message_sanitizer *sanitize)
Build the JSON representation of the message.
Definition stasis.h:252
stasis_message_vtable::to_event
struct ast_event *(* to_event)(struct stasis_message *message)
Build the ast_event representation of the message.
Definition stasis.h:278
test_message_types::change
struct stasis_message_type * change
Definition test_stasis.c:2062
test_message_types::event
struct stasis_message_type * event
Definition test_stasis.c:2057
test_message_types::type3
struct stasis_message_type * type3
Definition test_stasis.c:2061
test_message_types::ami
struct stasis_message_type * ami
Definition test_stasis.c:2055
test_message_types::json
struct stasis_message_type * json
Definition test_stasis.c:2056
test_message_types::amievent
struct stasis_message_type * amievent
Definition test_stasis.c:2058
test_message_types::type2
struct stasis_message_type * type2
Definition test_stasis.c:2060
test_message_types::none
struct stasis_message_type * none
Definition test_stasis.c:2054
test_message_types::type1
struct stasis_message_type * type1
Definition test_stasis.c:2059
fake_event
static struct ast_event * fake_event(struct stasis_message *message)
Definition test_stasis.c:43
fake_ami
static struct ast_manager_event_blob * fake_ami(struct stasis_message *message)
Definition test_stasis.c:56
destroy_message_types
static void destroy_message_types(void *obj)
Definition test_stasis.c:2065
fake_json
static struct ast_json * fake_json(struct stasis_message *message, const struct stasis_message_sanitizer *sanitize)
Definition test_stasis.c:49

References test_message_types::ami, test_message_types::amievent, ao2_alloc, ao2_cleanup, test_message_types::change, cleanup(), destroy_message_types(), test_message_types::event, fake_ami(), fake_event(), fake_json(), test_message_types::json, test_message_types::none, NULL, stasis_message_type_create(), STASIS_MESSAGE_TYPE_SUCCESS, stasis_subscription_change_type(), stasis_message_vtable::to_ami, stasis_message_vtable::to_event, stasis_message_vtable::to_json, test_message_types::type1, test_message_types::type2, and test_message_types::type3.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ destroy_cts()

static void destroy_cts ( void *  obj)
static

Definition at line 2145 of file test_stasis.c.

2146{
2147 struct cts *c = obj;
2148
2149 stasis_unsubscribe(c->sub);
2150 ao2_cleanup(c->topic);
2151 ao2_cleanup(c->consumer);
2152}
c
static struct test_val c
Definition test_linkedlists.c:48

References ao2_cleanup, c, and stasis_unsubscribe().

Referenced by create_cts().

◆ destroy_message_types()

static void destroy_message_types ( void *  obj)
static

Definition at line 2065 of file test_stasis.c.

2066{
2067 struct test_message_types *types = obj;
2068
2069 ao2_cleanup(types->none);
2070 ao2_cleanup(types->ami);
2071 ao2_cleanup(types->json);
2072 ao2_cleanup(types->event);
2073 ao2_cleanup(types->amievent);
2074 ao2_cleanup(types->type1);
2075 ao2_cleanup(types->type2);
2076 ao2_cleanup(types->type3);
2077 /* N.B. Don't cleanup types->change! */
2078}

References test_message_types::ami, test_message_types::amievent, ao2_cleanup, test_message_types::event, test_message_types::json, test_message_types::none, test_message_types::type1, test_message_types::type2, and test_message_types::type3.

Referenced by create_message_types().

◆ dump_consumer()

static void dump_consumer ( struct ast_test *  test,
struct cts *  cts 
)
static

Definition at line 2193 of file test_stasis.c.

2194{
2195 int i;
2196 struct stasis_subscription_change *data;
2197
2198 ast_test_status_update(test, "Messages received: %zu Final? %s\n", cts->consumer->messages_rxed_len,
2199 cts->consumer->complete ? "yes" : "no");
2200 for (i = 0; i < cts->consumer->messages_rxed_len; i++) {
2201 data = stasis_message_data(cts->consumer->messages_rxed[i]);
2202 ast_test_status_update(test, "Message type received: %s %s\n",
2203 stasis_message_type_name(stasis_message_type(cts->consumer->messages_rxed[i])),
2204 data && !ast_strlen_zero(data->description) ? data->description : "no data");
2205 }
2206}
ast_strlen_zero
static force_inline int attribute_pure ast_strlen_zero(const char *s)
Definition strings.h:65
ast_test_status_update
#define ast_test_status_update(a, b, c...)
Definition test.h:129

References ast_strlen_zero(), ast_test_status_update, consumer::complete, cts::consumer, stasis_subscription_change::description, consumer::messages_rxed, consumer::messages_rxed_len, stasis_message_data(), and stasis_message_type_name().

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ fake_ami()

static struct ast_manager_event_blob * fake_ami ( struct stasis_message *  message)
static

Definition at line 56 of file test_stasis.c.

57{
58 RAII_VAR(struct ast_manager_event_blob *, res, NULL, ao2_cleanup);
59 const char *text = stasis_message_data(message);
60
61 res = ast_manager_event_blob_create(EVENT_FLAG_TEST, "FakeMI",
62 "Message: %s\r\n", text);
63
64 if (res == NULL) {
65 return NULL;
66 }
67
68 ao2_ref(res, +1);
69 return res;
70}
text
char * text
Definition app_queue.c:1768
EVENT_FLAG_TEST
#define EVENT_FLAG_TEST
Definition manager.h:92
ast_manager_event_blob_create
struct ast_manager_event_blob * ast_manager_event_blob_create(int event_flags, const char *manager_event, const char *extra_fields_fmt,...)
Construct a ast_manager_event_blob.
Definition manager.c:10144

References ao2_cleanup, ao2_ref, ast_manager_event_blob_create(), EVENT_FLAG_TEST, NULL, RAII_VAR, stasis_message_data(), and text.

Referenced by create_message_types().

◆ fake_event()

static struct ast_event * fake_event ( struct stasis_message *  message)
static

Definition at line 43 of file test_stasis.c.

44{
45 return ast_event_new(AST_EVENT_CUSTOM,
46 AST_EVENT_IE_DESCRIPTION, AST_EVENT_IE_PLTYPE_STR, "Dummy", AST_EVENT_IE_END);
47}
ast_event_new
struct ast_event * ast_event_new(enum ast_event_type event_type,...)
Create a new event.
Definition event.c:403
AST_EVENT_IE_END
@ AST_EVENT_IE_END
Definition event_defs.h:70
AST_EVENT_IE_DESCRIPTION
@ AST_EVENT_IE_DESCRIPTION
Description Used by: AST_EVENT_SUB, AST_EVENT_UNSUB Payload type: STR.
Definition event_defs.h:265
AST_EVENT_CUSTOM
@ AST_EVENT_CUSTOM
Definition event_defs.h:36
AST_EVENT_IE_PLTYPE_STR
@ AST_EVENT_IE_PLTYPE_STR
Definition event_defs.h:335

References AST_EVENT_CUSTOM, AST_EVENT_IE_DESCRIPTION, AST_EVENT_IE_END, AST_EVENT_IE_PLTYPE_STR, and ast_event_new().

Referenced by create_message_types().

◆ fake_json()

static struct ast_json * fake_json ( struct stasis_message *  message,
const struct stasis_message_sanitizer *  sanitize 
)
static

Definition at line 49 of file test_stasis.c.

50{
51 const char *text = stasis_message_data(message);
52
53 return ast_json_string_create(text);
54}

References ast_json_string_create(), stasis_message_data(), and text.

Referenced by create_message_types().

◆ is_msg()

static int is_msg ( struct stasis_message *  msg,
struct stasis_message_type *  mtype,
const char *  data 
)
static

Definition at line 2178 of file test_stasis.c.

2179{
2180 struct stasis_subscription_change *msg_data = stasis_message_data(msg);
2181
2182 if (stasis_message_type(msg) != mtype) {
2183 return 0;
2184 }
2185
2186 if (data) {
2187 return (strcmp(data, msg_data->description) == 0);
2188 }
2189
2190 return 1;
2191}
mtype
const char * mtype
Definition http.c:151
msg_data
Definition main/message.c:246

References mtype, and stasis_message_data().

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ load_module()

static int load_module ( void  )
static

Definition at line 2468 of file test_stasis.c.

2469{
2470 AST_TEST_REGISTER(message_type);
2471 AST_TEST_REGISTER(message);
2472 AST_TEST_REGISTER(subscription_messages);
2473 AST_TEST_REGISTER(subscription_pool_messages);
2474 AST_TEST_REGISTER(publish);
2475 AST_TEST_REGISTER(publish_sync);
2476 AST_TEST_REGISTER(publish_pool);
2477 AST_TEST_REGISTER(unsubscribe_stops_messages);
2478 AST_TEST_REGISTER(forward);
2479 AST_TEST_REGISTER(cache_filter);
2480 AST_TEST_REGISTER(cache);
2481 AST_TEST_REGISTER(cache_dump);
2482 AST_TEST_REGISTER(cache_eid_aggregate);
2483 AST_TEST_REGISTER(router);
2484 AST_TEST_REGISTER(router_pool);
2485 AST_TEST_REGISTER(router_cache_updates);
2486 AST_TEST_REGISTER(interleaving);
2487 AST_TEST_REGISTER(subscription_interleaving);
2488 AST_TEST_REGISTER(no_to_json);
2489 AST_TEST_REGISTER(to_json);
2490 AST_TEST_REGISTER(no_to_ami);
2491 AST_TEST_REGISTER(to_ami);
2492 AST_TEST_REGISTER(dtor_order);
2493 AST_TEST_REGISTER(caching_dtor_order);
2494 AST_TEST_REGISTER(type_filters);
2495 AST_TEST_REGISTER(formatter_filters);
2496 AST_TEST_REGISTER(combo_filters);
2497 return AST_MODULE_LOAD_SUCCESS;
2498}
AST_MODULE_LOAD_SUCCESS
@ AST_MODULE_LOAD_SUCCESS
Definition module.h:70
router
static struct stasis_message_router * router
Definition res_chan_stats.c:46
publish
unsigned char publish
Definition res_corosync.c:241
to_ami
static void to_ami(struct ast_sip_subscription *sub, struct ast_str **buf)
Definition res_pjsip_exten_state.c:574
AST_TEST_REGISTER
#define AST_TEST_REGISTER(cb)
Definition test.h:127

References AST_MODULE_LOAD_SUCCESS, AST_TEST_REGISTER, cache, publish, router, and to_ami().

◆ noop()

static void noop ( void *  data,
struct stasis_subscription *  sub,
struct stasis_message *  message 
)
static

Definition at line 1962 of file test_stasis.c.

1964{
1965 /* no-op */
1966}

Referenced by AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ noop_get_id()

static const char * noop_get_id ( struct stasis_message *  message)
static

Definition at line 2001 of file test_stasis.c.

2002{
2003 return NULL;
2004}

References NULL.

Referenced by AST_TEST_DEFINE().

◆ send_msg()

static int send_msg ( struct ast_test *  test,
struct cts *  cts,
struct stasis_message_type *  msg_type,
const char *  data 
)
static

Definition at line 2208 of file test_stasis.c.

2210{
2211 struct stasis_message *msg;
2212 struct stasis_subscription_change *test_data =
2213 ao2_alloc(sizeof(*test_data) + (data ? strlen(data) : strlen("no data")) + 1, NULL);
2214
2215 if (!test_data) {
2216 return 0;
2217 }
2218 strcpy(test_data->description, S_OR(data, "no data")); /* Safe */
2219
2220 msg = stasis_message_create(msg_type, test_data);
2221 ao2_ref(test_data, -1);
2222 if (!msg) {
2223 ast_test_status_update(test, "Unable to create %s message\n",
2224 stasis_message_type_name(msg_type));
2225 return 0;
2226 }
2227
2228 stasis_publish(cts->topic, msg);
2229 ao2_ref(msg, -1);
2230
2231 return 1;
2232}
S_OR
#define S_OR(a, b)
returns the equivalent of logic or for strings: first one if not empty, otherwise second one.
Definition strings.h:80

References ao2_alloc, ao2_ref, ast_test_status_update, NULL, S_OR, stasis_message_create(), stasis_message_type_name(), stasis_publish(), and cts::topic.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

◆ unload_module()

static int unload_module ( void  )
static

Definition at line 2436 of file test_stasis.c.

2437{
2438 AST_TEST_UNREGISTER(message_type);
2439 AST_TEST_UNREGISTER(message);
2440 AST_TEST_UNREGISTER(subscription_messages);
2441 AST_TEST_UNREGISTER(subscription_pool_messages);
2442 AST_TEST_UNREGISTER(publish);
2443 AST_TEST_UNREGISTER(publish_sync);
2444 AST_TEST_UNREGISTER(publish_pool);
2445 AST_TEST_UNREGISTER(unsubscribe_stops_messages);
2446 AST_TEST_UNREGISTER(forward);
2447 AST_TEST_UNREGISTER(cache_filter);
2448 AST_TEST_UNREGISTER(cache);
2449 AST_TEST_UNREGISTER(cache_dump);
2450 AST_TEST_UNREGISTER(cache_eid_aggregate);
2451 AST_TEST_UNREGISTER(router);
2452 AST_TEST_UNREGISTER(router_pool);
2453 AST_TEST_UNREGISTER(router_cache_updates);
2454 AST_TEST_UNREGISTER(interleaving);
2455 AST_TEST_UNREGISTER(subscription_interleaving);
2456 AST_TEST_UNREGISTER(no_to_json);
2457 AST_TEST_UNREGISTER(to_json);
2458 AST_TEST_UNREGISTER(no_to_ami);
2459 AST_TEST_UNREGISTER(to_ami);
2460 AST_TEST_UNREGISTER(dtor_order);
2461 AST_TEST_UNREGISTER(caching_dtor_order);
2462 AST_TEST_UNREGISTER(type_filters);
2463 AST_TEST_UNREGISTER(formatter_filters);
2464 AST_TEST_UNREGISTER(combo_filters);
2465 return 0;
2466}
AST_TEST_UNREGISTER
#define AST_TEST_UNREGISTER(cb)
Definition test.h:128

References AST_TEST_UNREGISTER, cache, publish, router, and to_ami().

Variable Documentation

◆ __mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = 0 , .description = "Stasis testing" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module }
static

Definition at line 2504 of file test_stasis.c.

◆ ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info
static

Definition at line 2504 of file test_stasis.c.

◆ fake_vtable

struct stasis_message_vtable fake_vtable
static
Initial value:
= {
.to_json = fake_json,
.to_ami = fake_ami
}

Definition at line 72 of file test_stasis.c.

72 {
73 .to_json = fake_json,
74 .to_ami = fake_ami
75};

Referenced by AST_TEST_DEFINE(), and AST_TEST_DEFINE().

Generated on Wed Oct 15 2025 20:04:37 for Asterisk - The Open Source Telephony Project by doxygen 1.9.8