diff options
author | David M. Lee <dlee@digium.com> | 2013-03-15 17:35:16 +0000 |
---|---|---|
committer | David M. Lee <dlee@digium.com> | 2013-03-15 17:35:16 +0000 |
commit | 49e3489cac88927c87ce2b9c3e69a51aacf0bdc5 (patch) | |
tree | 501c46235f3b993322cb174f01d7190fc3b2fbd9 | |
parent | 641fc7ea54618828843e742f17c4d3415c1c5e26 (diff) |
A simplistic router for stasis_message's.
Often times, when subscribing to a topic, one wants to handle
different message types differently. While one could cascade if/else
statements through the subscription handler, it is much cleaner to
specify a different callback for each message type. The
stasis_message_router is here to help!
A stasis_message_router is constructed for a particular stasis_topic,
which is subscribes to. Call stasis_message_router_unsubscribe() to
cancel that subscription.
Once constructed, routes can be added using
stasis_message_router_add() (or stasis_message_router_set_default()
for any messages not handled by other routes). There may be only one
route per stasis_message_type. The route's callback is invoked just as
if it were a callback for a subscription; but it only gets called for
messages of the specified type.
(issue ASTERISK-20887)
Review: https://reviewboard.asterisk.org/r/2390/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@383242 65c4cc65-6c06-0410-ace0-fbb531ad65f3
-rw-r--r-- | include/asterisk/stasis_message_router.h | 89 | ||||
-rw-r--r-- | main/stasis.c | 8 | ||||
-rw-r--r-- | main/stasis_cache.c | 2 | ||||
-rw-r--r-- | main/stasis_message.c | 4 | ||||
-rw-r--r-- | main/stasis_message_router.c | 244 | ||||
-rw-r--r-- | tests/test_stasis.c | 147 |
6 files changed, 487 insertions, 7 deletions
diff --git a/include/asterisk/stasis_message_router.h b/include/asterisk/stasis_message_router.h new file mode 100644 index 000000000..42770d293 --- /dev/null +++ b/include/asterisk/stasis_message_router.h @@ -0,0 +1,89 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2013, Digium, Inc. + * + * David M. Lee, II <dlee@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +#ifndef _ASTERISK_STASIS_MESSAGE_ROUTER_H +#define _ASTERISK_STASIS_MESSAGE_ROUTER_H + +/*! + * \brief A simplistic router for \ref stasis_message's. + * + * Often times, when subscribing to a topic, one wants to handle different + * message types differently. While one could cascade if/else statements through + * the subscription handler, it is much cleaner to specify a different callback + * for each message type. The \ref stasis_message_router is here to help! + * + * A \ref stasis_message_router is constructed for a particular \ref + * stasis_topic, which is subscribes to. Call + * stasis_message_router_unsubscribe() to cancel that subscription. + * + * Once constructed, routes can be added using stasis_message_router_add() (or + * stasis_message_router_set_default() for any messages not handled by other + * routes). There may be only one route per \ref stasis_message_type. The + * route's \a callback is invoked just as if it were a callback for a + * subscription; but it only gets called for messages of the specified type. + * + * \since 12 + */ + +#include "asterisk/stasis.h" + +/*! \brief Stasis message routing object */ +struct stasis_message_router; + +/*! + * \brief Create a new message router object. + * \param topic Topic to subscribe route to. + * \return New \ref stasis_message_router. + * \return \c NULL on error. + * \since 12 + */ +struct stasis_message_router *stasis_message_router_create( + struct stasis_topic *topic); + +/*! + * \brief Unsubscribe the router from the upstream topic. + * \param router Router to unsubscribe. + * \since 12 + */ +void stasis_message_router_unsubscribe(struct stasis_message_router *router); + +/*! + * \brief Add a route to a message router. + * \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. + * \param data Data pointer to pass to \a callback. + * \since 12 + */ +int stasis_message_router_add(struct stasis_message_router *router, + struct stasis_message_type *message_type, + stasis_subscription_cb callback, + void *data); + +/*! + * \brief Sets the default route of a router. + * \param router Router to set the default route of. + * \param callback Callback to forard messages which otherwise have no home. + * \param data Data pointer to pass to \a callback. + * \since 12 + */ +int stasis_message_router_set_default(struct stasis_message_router *router, + stasis_subscription_cb callback, + void *data); + +#endif /* _ASTERISK_STASIS_MESSAGE_ROUTER_H */ diff --git a/main/stasis.c b/main/stasis.c index 7f18bf5e1..a4d44b819 100644 --- a/main/stasis.c +++ b/main/stasis.c @@ -46,7 +46,7 @@ static struct ast_threadpool *pool; static struct stasis_message_type *__subscription_change_message_type; -/*! \private */ +/*! \internal */ struct stasis_topic { char *name; /*! Variable length array of the subscribers (raw pointer to avoid cyclic references) */ @@ -86,7 +86,7 @@ struct stasis_topic *stasis_topic_create(const char *name) } topic->num_subscribers_max = INITIAL_SUBSCRIBERS_MAX; - topic->subscribers = ast_calloc(topic->num_subscribers_max, sizeof(topic->subscribers)); + topic->subscribers = ast_calloc(topic->num_subscribers_max, sizeof(*topic->subscribers)); if (!topic->subscribers) { return NULL; } @@ -100,7 +100,7 @@ const char *stasis_topic_name(const struct stasis_topic *topic) return topic->name; } -/*! \private */ +/*! \internal */ struct stasis_subscription { /*! Unique ID for this subscription */ char *uniqueid; @@ -264,7 +264,7 @@ static int topic_add_subscription(struct stasis_topic *topic, struct stasis_subs } /*! - * \private + * \internal * \brief Information needed to dispatch a message to a subscription */ struct dispatch { diff --git a/main/stasis_cache.c b/main/stasis_cache.c index a82492870..85c823909 100644 --- a/main/stasis_cache.c +++ b/main/stasis_cache.c @@ -42,7 +42,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #define NUM_CACHE_BUCKETS 563 #endif -/*! \private */ +/*! \internal */ struct stasis_caching_topic { struct ao2_container *cache; struct stasis_topic *topic; diff --git a/main/stasis_message.c b/main/stasis_message.c index 8d397b935..8d2373f4d 100644 --- a/main/stasis_message.c +++ b/main/stasis_message.c @@ -35,7 +35,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/stasis.h" #include "asterisk/utils.h" -/*! \private */ +/*! \internal */ struct stasis_message_type { char *name; }; @@ -70,7 +70,7 @@ const char *stasis_message_type_name(const struct stasis_message_type *type) return type->name; } -/*! \private */ +/*! \internal */ struct stasis_message { /*! Time the message was created */ struct timeval timestamp; diff --git a/main/stasis_message_router.c b/main/stasis_message_router.c new file mode 100644 index 000000000..01ad88ae0 --- /dev/null +++ b/main/stasis_message_router.c @@ -0,0 +1,244 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2013, Digium, Inc. + * + * David M. Lee, II <dlee@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! \file + * + * \brief Stasis message router implementation. + * + * \author David M. Lee, II <dlee@digium.com> + */ + +/*** MODULEINFO + <support_level>core</support_level> + ***/ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$") + +#include "asterisk/astobj2.h" +#include "asterisk/stasis_message_router.h" + +#define INITIAL_ROUTES_MAX 8 + +/*! \internal */ +struct stasis_message_route { + /*! Message type handle by this route. */ + struct stasis_message_type *message_type; + /*! Callback function for incoming message processing. */ + stasis_subscription_cb callback; + /*! Data pointer to be handed to the callback. */ + void *data; +}; + +static void route_dtor(void *obj) +{ + struct stasis_message_route *route = obj; + + ao2_cleanup(route->message_type); + route->message_type = NULL; +} + +/*! \internal */ +struct stasis_message_router { + /*! Subscription to the upstream topic */ + struct stasis_subscription *subscription; + /*! Variable length array of the routes */ + struct stasis_message_route **routes; + /*! Route of last resort */ + struct stasis_message_route *default_route; + /*! Allocated length of the routes array */ + size_t num_routes_max; + /*! Current size of the routes array */ + size_t num_routes_current; +}; + +static void router_dtor(void *obj) +{ + struct stasis_message_router *router = obj; + size_t i; + + ast_assert(!stasis_subscription_is_subscribed(router->subscription)); + router->subscription = NULL; + for (i = 0; i < router->num_routes_current; ++i) { + ao2_cleanup(router->routes[i]); + router->routes[i] = NULL; + } + ast_free(router->routes); + router->routes = NULL; + ao2_cleanup(router->default_route); + router->default_route = NULL; +} + +static void router_dispatch(void *data, + struct stasis_subscription *sub, + struct stasis_topic *topic, + struct stasis_message *message) +{ + struct stasis_message_router *router = data; + RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup); + struct stasis_message_type *type = stasis_message_type(message); + size_t i; + + { + SCOPED_AO2LOCK(lock, router); + + /* We don't expect many message types, so a simple loop should + * be adequate, even if the complexity is O(n). Sorting the list + * would be an easy way to bring that down to O(log(n)). Using a + * hashtable/ao2_container could be even better. Just be sure to + * profile before you optimize! + */ + route = router->default_route; + for (i = 0; i < router->num_routes_current; ++i) { + if (router->routes[i]->message_type == type) { + route = router->routes[i]; + break; + } + } + + /* Ref the route before leaving the scoped lock */ + if (route) { + ao2_ref(route, +1); + } + } + + if (route) { + route->callback(route->data, sub, topic, message); + } + + if (stasis_subscription_final_message(sub, message)) { + ao2_cleanup(router); + return; + } + +} + +struct stasis_message_router *stasis_message_router_create( + struct stasis_topic *topic) +{ + RAII_VAR(struct stasis_message_router *, router, NULL, ao2_cleanup); + + router = ao2_alloc(sizeof(*router), router_dtor); + if (!router) { + return NULL; + } + + router->num_routes_max = INITIAL_ROUTES_MAX; + router->routes = ast_calloc(router->num_routes_max, + sizeof(*router->routes)); + if (!router->routes) { + return NULL; + } + + router->subscription = stasis_subscribe(topic, router_dispatch, router); + if (!router->subscription) { + return NULL; + } + + ao2_ref(router, +1); + return router; +} + +void stasis_message_router_unsubscribe(struct stasis_message_router *router) +{ + if (!router) { + return; + } + + stasis_unsubscribe(router->subscription); +} + +static struct stasis_message_route *route_create( + struct stasis_message_type *message_type, + stasis_subscription_cb callback, + void *data) +{ + RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup); + + route = ao2_alloc(sizeof(*route), route_dtor); + if (!route) { + return NULL; + } + + if (message_type) { + ao2_ref(message_type, +1); + } + route->message_type = message_type; + route->callback = callback; + route->data = data; + + ao2_ref(route, +1); + return route; +} + +static int add_route(struct stasis_message_router *router, + struct stasis_message_route *route) +{ + struct stasis_message_route **routes; + size_t i; + SCOPED_AO2LOCK(lock, router); + + /* Check for route conflicts */ + for (i = 0; i < router->num_routes_current; ++i) { + if (router->routes[i]->message_type == route->message_type) { + return -1; + } + } + + /* Increase list size, if needed */ + if (router->num_routes_current + 1 > router->num_routes_max) { + routes = realloc(router->routes, + 2 * router->num_routes_max * sizeof(*routes)); + if (!routes) { + return -1; + } + router->routes = routes; + router->num_routes_max *= 2; + } + + + ao2_ref(route, +1); + router->routes[router->num_routes_current++] = 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) +{ + RAII_VAR(struct stasis_message_route *, route, NULL, ao2_cleanup); + + route = route_create(message_type, callback, data); + if (!route) { + return -1; + } + + return add_route(router, route); +} + +int stasis_message_router_set_default(struct stasis_message_router *router, + stasis_subscription_cb callback, + void *data) +{ + SCOPED_AO2LOCK(lock, router); + ao2_cleanup(router->default_route); + router->default_route = route_create(NULL, callback, data); + return router->default_route ? 0 : -1; +} diff --git a/tests/test_stasis.c b/tests/test_stasis.c index fb81ecde3..4c5072330 100644 --- a/tests/test_stasis.c +++ b/tests/test_stasis.c @@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/astobj2.h" #include "asterisk/module.h" #include "asterisk/stasis.h" +#include "asterisk/stasis_message_router.h" #include "asterisk/test.h" static const char *test_category = "/stasis/core/"; @@ -674,6 +675,148 @@ AST_TEST_DEFINE(cache) return AST_TEST_PASS; } +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_type *, test_message_type, NULL, ao2_cleanup); + RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup); + RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup); + int ret; + + switch (cmd) { + case TEST_INIT: + info->name = __func__; + info->category = test_category; + info->summary = + "Multiple routes to the same message_type should fail"; + info->description = + "Multiple routes to the same message_type should fail"; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + topic = stasis_topic_create("TestTopic"); + ast_test_validate(test, NULL != topic); + + consumer1 = consumer_create(1); + ast_test_validate(test, NULL != consumer1); + consumer2 = consumer_create(1); + ast_test_validate(test, NULL != consumer2); + + test_message_type = stasis_message_type_create("TestMessage"); + ast_test_validate(test, NULL != test_message_type); + + uut = stasis_message_router_create(topic); + ast_test_validate(test, NULL != uut); + + ret = stasis_message_router_add( + uut, test_message_type, consumer_exec, consumer1); + ast_test_validate(test, 0 == ret); + ret = stasis_message_router_add( + uut, test_message_type, consumer_exec, consumer2); + ast_test_validate(test, 0 != ret); + + return AST_TEST_PASS; +} + +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(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); + RAII_VAR(struct stasis_message_type *, test_message_type3, 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 *, 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); + int actual_len, ret; + struct stasis_message *actual; + + switch (cmd) { + case TEST_INIT: + info->name = __func__; + info->category = test_category; + info->summary = "Test simple message routing"; + info->description = "Test simple message routing"; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + topic = stasis_topic_create("TestTopic"); + ast_test_validate(test, NULL != 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("TestMessage1"); + ast_test_validate(test, NULL != test_message_type1); + test_message_type2 = stasis_message_type_create("TestMessage2"); + ast_test_validate(test, NULL != test_message_type2); + test_message_type3 = stasis_message_type_create("TestMessage3"); + ast_test_validate(test, NULL != test_message_type3); + + uut = stasis_message_router_create(topic); + ast_test_validate(test, NULL != uut); + + ret = stasis_message_router_add( + uut, test_message_type1, consumer_exec, consumer1); + ast_test_validate(test, 0 == ret); + ao2_ref(consumer1, +1); + ret = stasis_message_router_add( + uut, test_message_type2, 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, test_message1 == actual); + + actual = consumer2->messages_rxed[0]; + ast_test_validate(test, test_message2 == actual); + + 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; +} + static int unload_module(void) { AST_TEST_UNREGISTER(message_type); @@ -684,6 +827,8 @@ static int unload_module(void) AST_TEST_UNREGISTER(forward); AST_TEST_UNREGISTER(cache_passthrough); AST_TEST_UNREGISTER(cache); + AST_TEST_UNREGISTER(route_conflicts); + AST_TEST_UNREGISTER(router); return 0; } @@ -697,6 +842,8 @@ static int load_module(void) AST_TEST_REGISTER(forward); AST_TEST_REGISTER(cache_passthrough); AST_TEST_REGISTER(cache); + AST_TEST_REGISTER(route_conflicts); + AST_TEST_REGISTER(router); return AST_MODULE_LOAD_SUCCESS; } |