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 /main | |
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
Diffstat (limited to 'main')
-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 |
4 files changed, 251 insertions, 7 deletions
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; +} |