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

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;
+}