[asterisk-commits] dlee: branch dlee/stasis-router r383051 - in /team/dlee/stasis-router: includ...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Thu Mar 14 08:27:23 CDT 2013
Author: dlee
Date: Thu Mar 14 08:27:20 2013
New Revision: 383051
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=383051
Log:
Routing messages, so you don't have to.
Added:
team/dlee/stasis-router/include/asterisk/stasis_message_router.h (with props)
team/dlee/stasis-router/main/stasis_message_router.c (with props)
Modified:
team/dlee/stasis-router/main/stasis.c
team/dlee/stasis-router/tests/test_stasis.c
Added: team/dlee/stasis-router/include/asterisk/stasis_message_router.h
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-router/include/asterisk/stasis_message_router.h?view=auto&rev=383051
==============================================================================
--- team/dlee/stasis-router/include/asterisk/stasis_message_router.h (added)
+++ team/dlee/stasis-router/include/asterisk/stasis_message_router.h Thu Mar 14 08:27:20 2013
@@ -1,0 +1,82 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2013, Digium, Inc.
+ *
+ * David M. Lee, II <dlee at 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!
+ *
+ * Once constructed (using stasis_message_router_create()), the router can be
+ * subscribed to any number of \ref stasis_topic's.
+ *
+ * Individual message handlers may be added using stasis_message_router_add(). A
+ * default route may be added using stasis_message_router_add_default().
+ */
+
+#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 */
Propchange: team/dlee/stasis-router/include/asterisk/stasis_message_router.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: team/dlee/stasis-router/include/asterisk/stasis_message_router.h
------------------------------------------------------------------------------
svn:keywords = Author Date Id Rev URL
Propchange: team/dlee/stasis-router/include/asterisk/stasis_message_router.h
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: team/dlee/stasis-router/main/stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-router/main/stasis.c?view=diff&rev=383051&r1=383050&r2=383051
==============================================================================
--- team/dlee/stasis-router/main/stasis.c (original)
+++ team/dlee/stasis-router/main/stasis.c Thu Mar 14 08:27:20 2013
@@ -86,7 +86,7 @@
}
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;
}
Added: team/dlee/stasis-router/main/stasis_message_router.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-router/main/stasis_message_router.c?view=auto&rev=383051
==============================================================================
--- team/dlee/stasis-router/main/stasis_message_router.c (added)
+++ team/dlee/stasis-router/main/stasis_message_router.c Thu Mar 14 08:27:20 2013
@@ -1,0 +1,246 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2013, Digium, Inc.
+ *
+ * David M. Lee, II <dlee at 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 at 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
+
+/*! \private */
+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;
+}
+
+/*! \private */
+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;
+
+ if (stasis_subscription_final_message(sub, message)) {
+ ao2_lock(router);
+
+ for (i = 0; i < router->num_routes_current; ++i) {
+ router->routes[i]->callback(
+ router->routes[i]->data, sub, topic, message);
+ }
+
+ router->default_route->callback(
+ router->default_route->data, sub, topic, message);
+
+ ao2_unlock(router);
+
+ ao2_cleanup(router);
+ return;
+ }
+
+ {
+ 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.
+ */
+ 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);
+ }
+
+}
+
+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;
+ SCOPED_AO2LOCK(lock, router);
+
+ /* 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;
+}
Propchange: team/dlee/stasis-router/main/stasis_message_router.c
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: team/dlee/stasis-router/main/stasis_message_router.c
------------------------------------------------------------------------------
svn:keywords = Author Date Id Rev URL
Propchange: team/dlee/stasis-router/main/stasis_message_router.c
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: team/dlee/stasis-router/tests/test_stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-router/tests/test_stasis.c?view=diff&rev=383051&r1=383050&r2=383051
==============================================================================
--- team/dlee/stasis-router/tests/test_stasis.c (original)
+++ team/dlee/stasis-router/tests/test_stasis.c Thu Mar 14 08:27:20 2013
@@ -36,6 +36,7 @@
#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/";
@@ -676,6 +677,100 @@
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);
+ actual_len = consumer_wait_for(consumer1, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual = consumer1->messages_rxed[0];
+ ast_test_validate(test, test_message1 == actual);
+ ast_test_validate(test, 0 == consumer_should_stay(consumer2, 0));
+ ast_test_validate(test, 0 == consumer_should_stay(consumer3, 0));
+
+ stasis_publish(topic, test_message2);
+ actual_len = consumer_wait_for(consumer2, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual = consumer2->messages_rxed[0];
+ ast_test_validate(test, test_message2 == actual);
+ ast_test_validate(test, 1 == consumer_should_stay(consumer1, 1));
+ ast_test_validate(test, 0 == consumer_should_stay(consumer3, 0));
+
+ stasis_publish(topic, test_message3);
+ actual_len = consumer_wait_for(consumer3, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual = consumer3->messages_rxed[0];
+ ast_test_validate(test, test_message3 == actual);
+ ast_test_validate(test, 1 == consumer_should_stay(consumer1, 1));
+ ast_test_validate(test, 1 == consumer_should_stay(consumer2, 1));
+
+ return AST_TEST_PASS;
+}
+
static int unload_module(void)
{
AST_TEST_UNREGISTER(message_type);
@@ -686,6 +781,7 @@
AST_TEST_UNREGISTER(forward);
AST_TEST_UNREGISTER(cache_passthrough);
AST_TEST_UNREGISTER(cache);
+ AST_TEST_UNREGISTER(router);
return 0;
}
@@ -699,6 +795,7 @@
AST_TEST_REGISTER(forward);
AST_TEST_REGISTER(cache_passthrough);
AST_TEST_REGISTER(cache);
+ AST_TEST_REGISTER(router);
return AST_MODULE_LOAD_SUCCESS;
}
More information about the asterisk-commits
mailing list