[svn-commits] dlee: trunk r395118 - in /trunk: include/asterisk/ main/ tests/
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Tue Jul 23 08:39:55 CDT 2013
Author: dlee
Date: Tue Jul 23 08:39:50 2013
New Revision: 395118
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=395118
Log:
Fix bridge/channel AMI event ordering issues
The stasis_cache_update messages are somewhat cumbersome to handle
with the stasis_message_router. Since all updates have the same
message type, they are normally handled with the same route.
Since caching itself is a first class component of stasis-core, it
makes sense for the router to handle the cache update messages itself.
This patch adds stasis_message_router_add_cache_update() and
stasis_message_router_remove_cache_update() to handle the routing of
stasis_cache_update messages.
This patch also corrects an issue with manager_{bridging,channels}.c,
where events might be reordered. The reordering occurs because the
components use different message routers, which they needed because
they both needed to route cache update messages. They now both use
manager's router, and add cache routes for just the cache updates they
are interested in.
(closes issue ASTERISK-22038)
Review: https://reviewboard.asterisk.org/r/2677/
Modified:
trunk/include/asterisk/stasis_message_router.h
trunk/main/cdr.c
trunk/main/manager_bridging.c
trunk/main/manager_channels.c
trunk/main/stasis_message_router.c
trunk/tests/test_stasis.c
Modified: trunk/include/asterisk/stasis_message_router.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/stasis_message_router.h?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/include/asterisk/stasis_message_router.h (original)
+++ trunk/include/asterisk/stasis_message_router.h Tue Jul 23 08:39:50 2013
@@ -95,6 +95,11 @@
/*!
* \brief Add a route to a message router.
*
+ * A particular \a message_type may have at most one route per \a router. If
+ * you route \ref stasis_cache_update messages, the callback will only receive
+ * updates for types not handled by routes added with
+ * stasis_message_router_add_cache_update().
+ *
* \param router Router to add the route to.
* \param message_type Type of message to route.
* \param callback Callback to forard messages of \a message_type to.
@@ -106,9 +111,29 @@
* \since 12
*/
int stasis_message_router_add(struct stasis_message_router *router,
- struct stasis_message_type *message_type,
- stasis_subscription_cb callback,
- void *data);
+ struct stasis_message_type *message_type,
+ stasis_subscription_cb callback, void *data);
+
+/*!
+ * \brief Add a route for \ref stasis_cache_update messages to a message router.
+ *
+ * A particular \a message_type may have at most one cache route per \a router.
+ * These are distinct from regular routes, so one could have both a regular
+ * route and a cache route for the same \a message_type.
+ *
+ * \param router Router to add the route to.
+ * \param message_type Subtype of cache update to route.
+ * \param callback Callback to forard messages of \a message_type to.
+ * \param data Data pointer to pass to \a callback.
+ *
+ * \retval 0 on success
+ * \retval -1 on failure
+ *
+ * \since 12
+ */
+int stasis_message_router_add_cache_update(struct stasis_message_router *router,
+ struct stasis_message_type *message_type,
+ stasis_subscription_cb callback, void *data);
/*!
* \brief Remove a route from a message router.
@@ -119,7 +144,19 @@
* \since 12
*/
void stasis_message_router_remove(struct stasis_message_router *router,
- struct stasis_message_type *message_type);
+ struct stasis_message_type *message_type);
+
+/*!
+ * \brief Remove a cache route from a message router.
+ *
+ * \param router Router to remove the route from.
+ * \param message_type Type of message to route.
+ *
+ * \since 12
+ */
+void stasis_message_router_remove_cache_update(
+ struct stasis_message_router *router,
+ struct stasis_message_type *message_type);
/*!
* \brief Sets the default route of a router.
Modified: trunk/main/cdr.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/cdr.c?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/main/cdr.c (original)
+++ trunk/main/cdr.c Tue Jul 23 08:39:50 2013
@@ -1967,9 +1967,7 @@
struct cdr_object *it_cdr;
ast_assert(update != NULL);
- if (ast_channel_snapshot_type() != update->type) {
- return;
- }
+ ast_assert(ast_channel_snapshot_type() == update->type);
old_snapshot = stasis_message_data(update->old_snapshot);
new_snapshot = stasis_message_data(update->new_snapshot);
@@ -4024,7 +4022,7 @@
if (!stasis_router) {
return -1;
}
- stasis_message_router_add(stasis_router, stasis_cache_update_type(), handle_channel_cache_message, NULL);
+ stasis_message_router_add_cache_update(stasis_router, ast_channel_snapshot_type(), handle_channel_cache_message, NULL);
stasis_message_router_add(stasis_router, ast_channel_dial_type(), handle_dial_message, NULL);
stasis_message_router_add(stasis_router, ast_channel_entered_bridge_type(), handle_bridge_enter_message, NULL);
stasis_message_router_add(stasis_router, ast_channel_left_bridge_type(), handle_bridge_leave_message, NULL);
Modified: trunk/main/manager_bridging.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/manager_bridging.c?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/main/manager_bridging.c (original)
+++ trunk/main/manager_bridging.c Tue Jul 23 08:39:50 2013
@@ -196,9 +196,7 @@
update = stasis_message_data(message);
- if (ast_bridge_snapshot_type() != update->type) {
- return;
- }
+ ast_assert(ast_bridge_snapshot_type() == update->type);
old_snapshot = stasis_message_data(update->old_snapshot);
new_snapshot = stasis_message_data(update->new_snapshot);
@@ -495,35 +493,22 @@
return -1;
}
- /* BUGBUG - This should really route off of the manager_router, but
- * can't b/c manager_channels is already routing the
- * stasis_cache_update_type() messages. Having a separate router can
- * cause some message ordering issues with bridge and channel messages.
- */
- bridge_state_router = stasis_message_router_create(bridge_topic);
+ bridge_state_router = ast_manager_get_message_router();
if (!bridge_state_router) {
return -1;
}
+ ret |= stasis_message_router_add_cache_update(bridge_state_router,
+ ast_bridge_snapshot_type(), bridge_snapshot_update, NULL);
+
ret |= stasis_message_router_add(bridge_state_router,
- stasis_cache_update_type(),
- bridge_snapshot_update,
- NULL);
+ ast_bridge_merge_message_type(), bridge_merge_cb, NULL);
ret |= stasis_message_router_add(bridge_state_router,
- ast_bridge_merge_message_type(),
- bridge_merge_cb,
- NULL);
+ ast_channel_entered_bridge_type(), channel_enter_cb, NULL);
ret |= stasis_message_router_add(bridge_state_router,
- ast_channel_entered_bridge_type(),
- channel_enter_cb,
- NULL);
-
- ret |= stasis_message_router_add(bridge_state_router,
- ast_channel_left_bridge_type(),
- channel_leave_cb,
- NULL);
+ ast_channel_left_bridge_type(), channel_leave_cb, NULL);
ret |= ast_manager_register_xml_core("BridgeList", 0, manager_bridges_list);
ret |= ast_manager_register_xml_core("BridgeInfo", 0, manager_bridge_info);
Modified: trunk/main/manager_channels.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/manager_channels.c?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/main/manager_channels.c (original)
+++ trunk/main/manager_channels.c Tue Jul 23 08:39:50 2013
@@ -726,9 +726,7 @@
update = stasis_message_data(message);
- if (ast_channel_snapshot_type() != update->type) {
- return;
- }
+ ast_assert(ast_channel_snapshot_type() == update->type);
old_snapshot = stasis_message_data(update->old_snapshot);
new_snapshot = stasis_message_data(update->new_snapshot);
@@ -1283,85 +1281,57 @@
ast_register_atexit(manager_channels_shutdown);
- ret |= stasis_message_router_add(message_router,
- stasis_cache_update_type(),
- channel_snapshot_update,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_user_event_type(),
- channel_user_event_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_dtmf_begin_type(),
- channel_dtmf_begin_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_dtmf_end_type(),
- channel_dtmf_end_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_hangup_request_type(),
- channel_hangup_request_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_dial_type(),
- channel_dial_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_hold_type(),
- channel_hold_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_unhold_type(),
- channel_unhold_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_fax_type(),
- channel_fax_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_chanspy_start_type(),
- channel_chanspy_start_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_chanspy_stop_type(),
- channel_chanspy_stop_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_hangup_handler_type(),
- channel_hangup_handler_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_moh_start_type(),
- channel_moh_start_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_moh_stop_type(),
- channel_moh_stop_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_monitor_start_type(),
- channel_monitor_start_cb,
- NULL);
-
- ret |= stasis_message_router_add(message_router,
- ast_channel_monitor_stop_type(),
- channel_monitor_stop_cb,
- NULL);
+ ret |= stasis_message_router_add_cache_update(message_router,
+ ast_channel_snapshot_type(), channel_snapshot_update, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_user_event_type(), channel_user_event_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_dtmf_begin_type(), channel_dtmf_begin_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_dtmf_end_type(), channel_dtmf_end_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_hangup_request_type(), channel_hangup_request_cb,
+ NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_dial_type(), channel_dial_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_hold_type(), channel_hold_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_unhold_type(), channel_unhold_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_fax_type(), channel_fax_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_chanspy_start_type(), channel_chanspy_start_cb,
+ NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_chanspy_stop_type(), channel_chanspy_stop_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_hangup_handler_type(), channel_hangup_handler_cb,
+ NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_moh_start_type(), channel_moh_start_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_moh_stop_type(), channel_moh_stop_cb, NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_monitor_start_type(), channel_monitor_start_cb,
+ NULL);
+
+ ret |= stasis_message_router_add(message_router,
+ ast_channel_monitor_stop_type(), channel_monitor_stop_cb, NULL);
/* If somehow we failed to add any routes, just shut down the whole
* thing and fail it.
Modified: trunk/main/stasis_message_router.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_message_router.c?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/main/stasis_message_router.c (original)
+++ trunk/main/stasis_message_router.c Tue Jul 23 08:39:50 2013
@@ -33,6 +33,9 @@
#include "asterisk/astobj2.h"
#include "asterisk/stasis_message_router.h"
+
+/*! Number of hash buckets for the route table. Keep it prime! */
+#define ROUTE_TABLE_BUCKETS 7
/*! \internal */
struct stasis_message_route {
@@ -75,6 +78,8 @@
struct stasis_subscription *subscription;
/*! Subscribed routes */
struct ao2_container *routes;
+ /*! Subscribed routes for \ref stasi_cache_update messages */
+ struct ao2_container *cache_routes;
/*! Route of last resort */
struct stasis_message_route *default_route;
};
@@ -90,13 +95,46 @@
ao2_cleanup(router->routes);
router->routes = NULL;
+ ao2_cleanup(router->cache_routes);
+ router->cache_routes = NULL;
+
ao2_cleanup(router->default_route);
router->default_route = NULL;
}
-static struct stasis_message_route *find_route(struct stasis_message_router *router, struct stasis_message_type *message_type)
-{
- return ao2_find(router->routes, message_type, OBJ_KEY);
+static struct stasis_message_route *find_route(
+ struct stasis_message_router *router,
+ struct stasis_message *message)
+{
+ RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup);
+ struct stasis_message_type *type = stasis_message_type(message);
+ SCOPED_AO2LOCK(lock, router);
+
+ if (type == stasis_cache_update_type()) {
+ /* Find a cache route */
+ struct stasis_cache_update *update =
+ stasis_message_data(message);
+ route = ao2_find(router->cache_routes, update->type, OBJ_KEY);
+ }
+
+ if (route == NULL) {
+ /* Find a regular route */
+ route = ao2_find(router->routes, type, OBJ_KEY);
+ }
+
+ if (route == NULL) {
+ /* Maybe the default route, then? */
+ if ((route = router->default_route)) {
+ ao2_ref(route, +1);
+ }
+ }
+
+ if (route == NULL) {
+ return NULL;
+ }
+
+ ao2_ref(route, +1);
+ return route;
}
static void router_dispatch(void *data,
@@ -105,29 +143,18 @@
struct stasis_message *message)
{
struct stasis_message_router *router = data;
- RAII_VAR(struct stasis_message_router *, router_needs_cleanup, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup);
- struct stasis_message_type *type = stasis_message_type(message);
-
- {
- SCOPED_AO2LOCK(lock, router);
-
- if (!(route = find_route(router, type))) {
- if ((route = router->default_route)) {
- ao2_ref(route, +1);
- }
- }
- }
+ RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup);
+
+ route = find_route(router, message);
if (route) {
route->callback(route->data, sub, topic, message);
}
+
if (stasis_subscription_final_message(sub, message)) {
- router_needs_cleanup = router;
- return;
- }
-
+ ao2_cleanup(router);
+ }
}
struct stasis_message_router *stasis_message_router_create(
@@ -140,7 +167,15 @@
return NULL;
}
- if (!(router->routes = ao2_container_alloc(7, route_hash, route_cmp))) {
+ router->routes = ao2_container_alloc(ROUTE_TABLE_BUCKETS, route_hash,
+ route_cmp);
+ if (!router->routes) {
+ return NULL;
+ }
+
+ router->cache_routes = ao2_container_alloc(ROUTE_TABLE_BUCKETS,
+ route_hash, route_cmp);
+ if (!router->cache_routes) {
return NULL;
}
@@ -211,7 +246,10 @@
RAII_VAR(struct stasis_message_route *, existing_route, NULL, ao2_cleanup);
SCOPED_AO2LOCK(lock, router);
- if ((existing_route = find_route(router, route->message_type))) {
+ existing_route = ao2_find(router->routes, route->message_type, OBJ_KEY);
+
+ if (existing_route) {
+ ast_log(LOG_ERROR, "Cannot add route; route exists\n");
return -1;
}
@@ -219,10 +257,27 @@
return 0;
}
+static int add_cache_route(struct stasis_message_router *router,
+ struct stasis_message_route *route)
+{
+ RAII_VAR(struct stasis_message_route *, existing_route, NULL, ao2_cleanup);
+ SCOPED_AO2LOCK(lock, router);
+
+ existing_route = ao2_find(router->cache_routes, route->message_type,
+ OBJ_KEY);
+
+ if (existing_route) {
+ ast_log(LOG_ERROR, "Cannot add route; route exists\n");
+ return -1;
+ }
+
+ ao2_link(router->cache_routes, route);
+ return 0;
+}
+
int stasis_message_router_add(struct stasis_message_router *router,
- struct stasis_message_type *message_type,
- stasis_subscription_cb callback,
- void *data)
+ struct stasis_message_type *message_type,
+ stasis_subscription_cb callback, void *data)
{
RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup);
@@ -234,12 +289,37 @@
return add_route(router, route);
}
+int stasis_message_router_add_cache_update(struct stasis_message_router *router,
+ struct stasis_message_type *message_type,
+ stasis_subscription_cb callback, void *data)
+{
+ RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup);
+
+ route = route_create(message_type, callback, data);
+ if (!route) {
+ return -1;
+ }
+
+ return add_cache_route(router, route);
+}
+
void stasis_message_router_remove(struct stasis_message_router *router,
- struct stasis_message_type *message_type)
-{
- SCOPED_AO2LOCK(lock, router);
-
- ao2_find(router->routes, message_type, OBJ_UNLINK | OBJ_NODATA | OBJ_KEY);
+ struct stasis_message_type *message_type)
+{
+ SCOPED_AO2LOCK(lock, router);
+
+ ao2_find(router->routes, message_type,
+ OBJ_UNLINK | OBJ_NODATA | OBJ_KEY);
+}
+
+void stasis_message_router_remove_cache_update(
+ struct stasis_message_router *router,
+ struct stasis_message_type *message_type)
+{
+ SCOPED_AO2LOCK(lock, router);
+
+ ao2_find(router->cache_routes, message_type,
+ OBJ_UNLINK | OBJ_NODATA | OBJ_KEY);
}
int stasis_message_router_set_default(struct stasis_message_router *router,
Modified: trunk/tests/test_stasis.c
URL: http://svnview.digium.com/svn/asterisk/trunk/tests/test_stasis.c?view=diff&rev=395118&r1=395117&r2=395118
==============================================================================
--- trunk/tests/test_stasis.c (original)
+++ trunk/tests/test_stasis.c Tue Jul 23 08:39:50 2013
@@ -230,7 +230,7 @@
{
struct timeval start = ast_tvnow();
struct timespec end = {
- .tv_sec = start.tv_sec + 30,
+ .tv_sec = start.tv_sec + 3,
.tv_nsec = start.tv_usec * 1000
};
@@ -867,7 +867,7 @@
AST_TEST_DEFINE(route_conflicts)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
@@ -913,7 +913,7 @@
AST_TEST_DEFINE(router)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
@@ -1006,6 +1006,126 @@
return AST_TEST_PASS;
}
+static const char *cache_simple(struct stasis_message *message) {
+ const char *type_name =
+ stasis_message_type_name(stasis_message_type(message));
+ if (!ast_begins_with(type_name, "Cache")) {
+ return NULL;
+ }
+
+ return "cached";
+}
+
+AST_TEST_DEFINE(router_cache_updates)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
+ RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
+ RAII_VAR(char *, test_data, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, message1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, message2, NULL, ao2_cleanup);
+ struct stasis_cache_update *update;
+ int actual_len, ret;
+ struct stasis_message *actual;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary = "Test special handling cache_update messages";
+ info->description = "Test special handling cache_update messages";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ topic = stasis_topic_create("TestTopic");
+ ast_test_validate(test, NULL != topic);
+
+ caching_topic = stasis_caching_topic_create(topic, cache_simple);
+ ast_test_validate(test, NULL != caching_topic);
+
+ consumer1 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer1);
+ consumer2 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer2);
+ consumer3 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer3);
+
+ test_message_type1 = stasis_message_type_create("Cache1", NULL);
+ ast_test_validate(test, NULL != test_message_type1);
+ test_message_type2 = stasis_message_type_create("Cache2", NULL);
+ ast_test_validate(test, NULL != test_message_type2);
+ test_message_type3 = stasis_message_type_create("NonCache", NULL);
+ ast_test_validate(test, NULL != test_message_type3);
+
+ uut = stasis_message_router_create(
+ stasis_caching_get_topic(caching_topic));
+ ast_test_validate(test, NULL != uut);
+
+ ret = stasis_message_router_add_cache_update(
+ uut, test_message_type1, consumer_exec, consumer1);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer1, +1);
+ ret = stasis_message_router_add(
+ uut, stasis_cache_update_type(), consumer_exec, consumer2);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer2, +1);
+ ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer3, +1);
+
+ test_data = ao2_alloc(1, NULL);
+ ast_test_validate(test, NULL != test_data);
+ test_message1 = stasis_message_create(test_message_type1, test_data);
+ ast_test_validate(test, NULL != test_message1);
+ test_message2 = stasis_message_create(test_message_type2, test_data);
+ ast_test_validate(test, NULL != test_message2);
+ test_message3 = stasis_message_create(test_message_type3, test_data);
+ ast_test_validate(test, NULL != test_message3);
+
+ stasis_publish(topic, test_message1);
+ stasis_publish(topic, test_message2);
+ stasis_publish(topic, test_message3);
+
+ actual_len = consumer_wait_for(consumer1, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual_len = consumer_wait_for(consumer2, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual_len = consumer_wait_for(consumer3, 1);
+ ast_test_validate(test, 1 == actual_len);
+
+ actual = consumer1->messages_rxed[0];
+ ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
+ update = stasis_message_data(actual);
+ ast_test_validate(test, test_message_type1 == update->type);
+ ast_test_validate(test, test_message1 == update->new_snapshot);
+
+ actual = consumer2->messages_rxed[0];
+ ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
+ update = stasis_message_data(actual);
+ ast_test_validate(test, test_message_type2 == update->type);
+ ast_test_validate(test, test_message2 == update->new_snapshot);
+
+ actual = consumer3->messages_rxed[0];
+ ast_test_validate(test, test_message3 == actual);
+
+ /* consumer1 and consumer2 do not get the final message. */
+ ao2_cleanup(consumer1);
+ ao2_cleanup(consumer2);
+
+ return AST_TEST_PASS;
+}
+
AST_TEST_DEFINE(no_to_json)
{
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
@@ -1160,6 +1280,7 @@
AST_TEST_UNREGISTER(cache_dump);
AST_TEST_UNREGISTER(route_conflicts);
AST_TEST_UNREGISTER(router);
+ AST_TEST_UNREGISTER(router_cache_updates);
AST_TEST_UNREGISTER(interleaving);
AST_TEST_UNREGISTER(no_to_json);
AST_TEST_UNREGISTER(to_json);
@@ -1181,6 +1302,7 @@
AST_TEST_REGISTER(cache_dump);
AST_TEST_REGISTER(route_conflicts);
AST_TEST_REGISTER(router);
+ AST_TEST_REGISTER(router_cache_updates);
AST_TEST_REGISTER(interleaving);
AST_TEST_REGISTER(no_to_json);
AST_TEST_REGISTER(to_json);
More information about the svn-commits
mailing list