Convert message_router routes to ao2. Add support for removal.
authorJason Parker <jparker@digium.com>
Thu, 6 Jun 2013 19:44:45 +0000 (19:44 +0000)
committerJason Parker <jparker@digium.com>
Thu, 6 Jun 2013 19:44:45 +0000 (19:44 +0000)
Review: https://reviewboard.asterisk.org/r/2591/

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@390698 65c4cc65-6c06-0410-ace0-fbb531ad65f3

include/asterisk/stasis_message_router.h
main/stasis_message_router.c

index e7d5a4c..81a636d 100644 (file)
@@ -47,9 +47,12 @@ 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(
@@ -59,6 +62,7 @@ struct stasis_message_router *stasis_message_router_create(
  * \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);
@@ -71,6 +75,7 @@ void stasis_message_router_unsubscribe(struct stasis_message_router *router);
  * vs. stasis_message_router_unsubscribe().
  *
  * \param router Router to unsubscribe.
+ *
  * \since 12
  */
 void stasis_message_router_unsubscribe_and_join(
@@ -80,6 +85,7 @@ void stasis_message_router_unsubscribe_and_join(
  * \brief Returns whether \a router has received its final message.
  *
  * \param router Router.
+ *
  * \return True (non-zero) if stasis_subscription_final_message() has been
  *         received.
  * \return False (zero) if waiting for the end.
@@ -88,10 +94,15 @@ int stasis_message_router_is_done(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.
+ *
+ * \retval 0 on success
+ * \retval -1 on failure
+ *
  * \since 12
  */
 int stasis_message_router_add(struct stasis_message_router *router,
@@ -100,10 +111,26 @@ int stasis_message_router_add(struct stasis_message_router *router,
                              void *data);
 
 /*!
+ * \brief Remove a route from a message router.
+ *
+ * \param router Router to remove the route from.
+ * \param message_type Type of message to route.
+ *
+ * \since 12
+ */
+void stasis_message_router_remove(struct stasis_message_router *router,
+                             struct stasis_message_type *message_type);
+
+/*!
  * \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.
+ *
+ * \retval 0 on success
+ * \retval -1 on failure
+ *
  * \since 12
  */
 int stasis_message_router_set_default(struct stasis_message_router *router,
index c7acca1..4e7bbb5 100644 (file)
@@ -34,8 +34,6 @@ 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. */
@@ -54,38 +52,53 @@ static void route_dtor(void *obj)
        route->message_type = NULL;
 }
 
+static int route_hash(const void *obj, const int flags)
+{
+       const struct stasis_message_route *route = obj;
+       const struct stasis_message_type *message_type = (flags & OBJ_KEY) ? obj : route->message_type;
+
+       return ast_str_hash(stasis_message_type_name(message_type));
+}
+
+static int route_cmp(void *obj, void *arg, int flags)
+{
+       const struct stasis_message_route *left = obj;
+       const struct stasis_message_route *right = arg;
+       const struct stasis_message_type *message_type = (flags & OBJ_KEY) ? arg : right->message_type;
+
+       return (left->message_type == message_type) ? CMP_MATCH | CMP_STOP : 0;
+}
+
 /*! \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;
+       /*! Subscribed routes */
+       struct ao2_container *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));
        ast_assert(stasis_subscription_is_done(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);
+
+       ao2_cleanup(router->routes);
        router->routes = NULL;
+
        ao2_cleanup(router->default_route);
        router->default_route = NULL;
 }
 
+static struct stasis_message_route *find_route(struct stasis_message_router *router, struct stasis_message_type *message_type)
+{
+       return ao2_find(router->routes, message_type, OBJ_KEY);
+}
+
 static void router_dispatch(void *data,
                            struct stasis_subscription *sub,
                            struct stasis_topic *topic,
@@ -95,29 +108,15 @@ static void router_dispatch(void *data,
        RAII_VAR(struct stasis_message_router *, router_needs_cleanup, NULL, ao2_cleanup);
        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;
+               if (!(route = find_route(router, type))) {
+                       if ((route = router->default_route)) {
+                               ao2_ref(route, +1);
                        }
                }
-
-               /* Ref the route before leaving the scoped lock */
-               if (route) {
-                       ao2_ref(route, +1);
-               }
        }
 
        if (route) {
@@ -141,10 +140,7 @@ struct stasis_message_router *stasis_message_router_create(
                return NULL;
        }
 
-       router->num_routes_max = INITIAL_ROUTES_MAX;
-       router->routes = ast_calloc(router->num_routes_max,
-                                   sizeof(*router->routes));
-       if (!router->routes) {
+       if (!(router->routes = ao2_container_alloc(7, route_hash, route_cmp))) {
                return NULL;
        }
 
@@ -212,31 +208,14 @@ static struct stasis_message_route *route_create(
 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);
+       RAII_VAR(struct stasis_message_route *, existing_route, NULL, ao2_cleanup);
 
-       /* 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;
+       if ((existing_route = find_route(router, route->message_type))) {
+               return -1;
        }
 
-
-       ao2_ref(route, +1);
-       router->routes[router->num_routes_current++] = route;
+       ao2_link(router->routes, route);
        return 0;
 }
 
@@ -255,6 +234,14 @@ int stasis_message_router_add(struct stasis_message_router *router,
        return add_route(router, route);
 }
 
+void stasis_message_router_remove(struct stasis_message_router *router,
+                             struct stasis_message_type *message_type)
+{
+       SCOPED_AO2LOCK(lock, router);
+
+       ao2_find(router->routes, message_type, OBJ_UNLINK | OBJ_NODATA | OBJ_KEY);
+}
+
 int stasis_message_router_set_default(struct stasis_message_router *router,
                                      stasis_subscription_cb callback,
                                      void *data)