A simplistic router for stasis_message's.
authorDavid M. Lee <dlee@digium.com>
Fri, 15 Mar 2013 17:35:16 +0000 (17:35 +0000)
committerDavid M. Lee <dlee@digium.com>
Fri, 15 Mar 2013 17:35:16 +0000 (17:35 +0000)
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

include/asterisk/stasis_message_router.h [new file with mode: 0644]
main/stasis.c
main/stasis_cache.c
main/stasis_message.c
main/stasis_message_router.c [new file with mode: 0644]
tests/test_stasis.c

diff --git a/include/asterisk/stasis_message_router.h b/include/asterisk/stasis_message_router.h
new file mode 100644 (file)
index 0000000..42770d2
--- /dev/null
@@ -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 */
index 7f18bf5..a4d44b8 100644 (file)
@@ -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 {
index a824928..85c8239 100644 (file)
@@ -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;
index 8d397b9..8d2373f 100644 (file)
@@ -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 (file)
index 0000000..01ad88a
--- /dev/null
@@ -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;
+}
index fb81ecd..4c50723 100644 (file)
@@ -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;
 }