From 49e3489cac88927c87ce2b9c3e69a51aacf0bdc5 Mon Sep 17 00:00:00 2001 From: "David M. Lee" Date: Fri, 15 Mar 2013 17:35:16 +0000 Subject: 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 --- main/stasis_message_router.c | 244 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 main/stasis_message_router.c (limited to 'main/stasis_message_router.c') 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 + * + * 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 + */ + +/*** MODULEINFO + core + ***/ + +#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; +} -- cgit v1.2.3