stasis_message_router.c File Reference

Stasis message router implementation. More...

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/stasis_message_router.h"
#include "asterisk/vector.h"

Include dependency graph for stasis_message_router.c:

Go to the source code of this file.

Data Structures
struct	stasis_message_route
struct	stasis_message_router

Macros
#define	ROUTE_TABLE_ELEM_CLEANUP(elem)   ao2_cleanup((elem).message_type)
	route_table vector element cleanup. More...
#define	ROUTE_TABLE_ELEM_CMP(elem, value)   ((elem).message_type == (value))
	route_table comparator for AST_VECTOR_REMOVE_CMP_UNORDERED() More...

Functions
struct stasis_message_router *	__stasis_message_router_create (struct stasis_topic *topic, const char *file, int lineno, const char *func)
	Create a new message router object. More...
struct stasis_message_router *	__stasis_message_router_create_pool (struct stasis_topic *topic, const char *file, int lineno, const char *func)
	Create a new message router object. More...
	AST_VECTOR (route_table, struct stasis_message_route)
static int	find_route (struct stasis_message_router *router, struct stasis_message *message, struct stasis_message_route *route_out)
static int	route_table_add (struct route_table *table, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
static void	route_table_dtor (struct route_table *table)
static struct stasis_message_route *	route_table_find (struct route_table *table, struct stasis_message_type *message_type)
static int	route_table_remove (struct route_table *table, struct stasis_message_type *message_type)
static void	router_dispatch (void *data, struct stasis_subscription *sub, struct stasis_message *message)
static void	router_dtor (void *obj)
void	stasis_message_router_accept_formatters (struct stasis_message_router *router, enum stasis_subscription_message_formatters formatters)
	Indicate to a message router that we are interested in messages with one or more formatters. More...
int	stasis_message_router_add (struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
	Add a route to a message router. More...
int	stasis_message_router_add_cache_update (struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
	Add a route for stasis_cache_update messages to a message router. More...
static struct stasis_message_router *	stasis_message_router_create_internal (struct stasis_topic *topic, int use_thread_pool, const char *file, int lineno, const char *func)
int	stasis_message_router_is_done (struct stasis_message_router *router)
	Returns whether router has received its final message. More...
void	stasis_message_router_publish_sync (struct stasis_message_router *router, struct stasis_message *message)
	Publish a message to a message router's subscription synchronously. More...
void	stasis_message_router_remove (struct stasis_message_router *router, struct stasis_message_type *message_type)
	Remove a route from a message router. More...
void	stasis_message_router_remove_cache_update (struct stasis_message_router *router, struct stasis_message_type *message_type)
	Remove a cache route from a message router. More...
int	stasis_message_router_set_congestion_limits (struct stasis_message_router *router, long low_water, long high_water)
	Set the high and low alert water marks of the stasis message router. More...
int	stasis_message_router_set_default (struct stasis_message_router *router, stasis_subscription_cb callback, void *data)
	Sets the default route of a router. More...
void	stasis_message_router_set_formatters_default (struct stasis_message_router *router, stasis_subscription_cb callback, void *data, enum stasis_subscription_message_formatters formatters)
	Sets the default route of a router with formatters. More...
void	stasis_message_router_unsubscribe (struct stasis_message_router *router)
	Unsubscribe the router from the upstream topic. More...
void	stasis_message_router_unsubscribe_and_join (struct stasis_message_router *router)
	Unsubscribe the router from the upstream topic, blocking until the final message has been processed. More...

Detailed Description

Stasis message router implementation.

Author

David M. Lee, II dlee@.nosp@m.digi.nosp@m.um.co.nosp@m.m

Definition in file stasis_message_router.c.

Macro Definition Documentation

ROUTE_TABLE_ELEM_CLEANUP

#define ROUTE_TABLE_ELEM_CLEANUP

(

elem

)

ao2_cleanup((elem).message_type)

route_table vector element cleanup.

Parameters

elem	Element to cleanup

Returns

Nothing

Definition at line 88 of file stasis_message_router.c.

Referenced by route_table_add(), route_table_dtor(), and route_table_remove().

ROUTE_TABLE_ELEM_CMP

#define ROUTE_TABLE_ELEM_CMP	(		elem,
			value
	)		((elem).message_type == (value))

route_table comparator for AST_VECTOR_REMOVE_CMP_UNORDERED()

Parameters

elem	Element to compare against
value	Value to compare with the vector element.

Returns

0 if element does not match.

Non-zero if element matches.

Definition at line 79 of file stasis_message_router.c.

Referenced by route_table_remove().

Function Documentation

__stasis_message_router_create()

struct stasis_message_router* __stasis_message_router_create	(	struct stasis_topic *	topic,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a new message router object.

Parameters

topic

Topic to subscribe route to.

Returns

New stasis_message_router.

NULL on error.

Since

12

Definition at line 246 of file stasis_message_router.c.

References stasis_message_router_create_internal().

{
   return stasis_message_router_create_internal(topic, 0, file, lineno, func);
}

__stasis_message_router_create_pool()

struct stasis_message_router* __stasis_message_router_create_pool	(	struct stasis_topic *	topic,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a new message router object.

The subscription created for this message router will dispatch callbacks on a thread pool.

Parameters

topic

Topic to subscribe route to.

Returns

New stasis_message_router.

NULL on error.

Since

12.8.0

Definition at line 252 of file stasis_message_router.c.

References stasis_message_router_create_internal().

{
   return stasis_message_router_create_internal(topic, 1, file, lineno, func);
}

AST_VECTOR()

AST_VECTOR	(	route_table	,
		struct stasis_message_route
	)

find_route()

static int find_route ( struct stasis_message_router *  router, struct stasis_message *  message, struct stasis_message_route *  route_out  )

static

Definition at line 155 of file stasis_message_router.c.

References ast_assert, stasis_message_router::cache_routes, stasis_message_route::callback, stasis_message_router::default_route, lock, NULL, route_table_find(), stasis_message_router::routes, SCOPED_AO2LOCK, stasis_cache_update_type(), stasis_message_data(), stasis_message_type(), type, stasis_cache_update::type, and update().

Referenced by router_dispatch().

{
   struct stasis_message_route *route = NULL;
   struct stasis_message_type *type = stasis_message_type(message);
   SCOPED_AO2LOCK(lock, router);

   ast_assert(route_out != NULL);

   if (type == stasis_cache_update_type()) {
      /* Find a cache route */
      struct stasis_cache_update *update =
         stasis_message_data(message);
      route = route_table_find(&router->cache_routes, update->type);
   }

   if (route == NULL) {
      /* Find a regular route */
      route = route_table_find(&router->routes, type);
   }

   if (route == NULL && router->default_route.callback) {
      /* Maybe the default route, then? */
      route = &router->default_route;
   }

   if (!route) {
      return -1;
   }

   *route_out = *route;
   return 0;
}

route_table_add()

static int route_table_add ( struct route_table *  table, struct stasis_message_type *  message_type, stasis_subscription_cb  callback, void *  data  )

static

Definition at line 97 of file stasis_message_router.c.

References ao2_bump, ast_assert, AST_VECTOR_APPEND, stasis_message_route::callback, stasis_message_route::data, stasis_message_route::message_type, NULL, ROUTE_TABLE_ELEM_CLEANUP, and route_table_find().

Referenced by stasis_message_router_add(), and stasis_message_router_add_cache_update().

{
   struct stasis_message_route route;
   int res;

   ast_assert(callback != NULL);
   ast_assert(route_table_find(table, message_type) == NULL);

   route.message_type = ao2_bump(message_type);
   route.callback = callback;
   route.data = data;

   res = AST_VECTOR_APPEND(table, route);
   if (res) {
      ROUTE_TABLE_ELEM_CLEANUP(route);
   }
   return res;
}

route_table_dtor()

static void route_table_dtor ( struct route_table *  table )

static

Definition at line 118 of file stasis_message_router.c.

References AST_VECTOR_FREE, AST_VECTOR_GET_ADDR, AST_VECTOR_SIZE, and ROUTE_TABLE_ELEM_CLEANUP.

Referenced by router_dtor().

{
   size_t idx;
   struct stasis_message_route *route;

   for (idx = 0; idx < AST_VECTOR_SIZE(table); ++idx) {
      route = AST_VECTOR_GET_ADDR(table, idx);
      ROUTE_TABLE_ELEM_CLEANUP(*route);
   }
   AST_VECTOR_FREE(table);
}

route_table_find()

static struct stasis_message_route* route_table_find ( struct route_table *  table, struct stasis_message_type *  message_type  )

static

Definition at line 48 of file stasis_message_router.c.

References AST_VECTOR_GET_ADDR, AST_VECTOR_SIZE, stasis_message_route::message_type, and NULL.

Referenced by find_route(), and route_table_add().

{
   size_t idx;
   struct stasis_message_route *route;

   /* While a linear search for routes may seem very inefficient, most
    * route tables have six routes or less. For such small data, it's
    * hard to beat a linear search. If we start having larger route
    * tables, then we can look into containers with more efficient
    * lookups.
    */
   for (idx = 0; idx < AST_VECTOR_SIZE(table); ++idx) {
      route = AST_VECTOR_GET_ADDR(table, idx);
      if (route->message_type == message_type) {
         return route;
      }
   }

   return NULL;
}

route_table_remove()

static int route_table_remove ( struct route_table *  table, struct stasis_message_type *  message_type  )

static

Definition at line 90 of file stasis_message_router.c.

References AST_VECTOR_REMOVE_CMP_UNORDERED, ROUTE_TABLE_ELEM_CLEANUP, and ROUTE_TABLE_ELEM_CMP.

Referenced by stasis_message_router_remove(), and stasis_message_router_remove_cache_update().

{
   return AST_VECTOR_REMOVE_CMP_UNORDERED(table, message_type, ROUTE_TABLE_ELEM_CMP,
      ROUTE_TABLE_ELEM_CLEANUP);
}

router_dispatch()

static void router_dispatch ( void *  data, struct stasis_subscription *  sub, struct stasis_message *  message  )

static

Definition at line 191 of file stasis_message_router.c.

References ao2_cleanup, stasis_message_route::callback, stasis_message_route::data, find_route(), router, and stasis_subscription_final_message().

Referenced by stasis_message_router_create_internal().

{
   struct stasis_message_router *router = data;
   struct stasis_message_route route;

   if (find_route(router, message, &route) == 0) {
      route.callback(route.data, sub, message);
   }

   if (stasis_subscription_final_message(sub, message)) {
      ao2_cleanup(router);
   }
}

router_dtor()

static void router_dtor ( void *  obj )

static

Definition at line 142 of file stasis_message_router.c.

References ast_assert, stasis_message_router::cache_routes, NULL, route_table_dtor(), router, stasis_message_router::routes, stasis_subscription_is_done(), stasis_subscription_is_subscribed(), and stasis_message_router::subscription.

Referenced by stasis_message_router_create_internal().

{
   struct stasis_message_router *router = obj;

   ast_assert(!stasis_subscription_is_subscribed(router->subscription));
   ast_assert(stasis_subscription_is_done(router->subscription));

   router->subscription = NULL;

   route_table_dtor(&router->routes);
   route_table_dtor(&router->cache_routes);
}

stasis_message_router_accept_formatters()

void stasis_message_router_accept_formatters	(	struct stasis_message_router *	router,
		enum stasis_subscription_message_formatters	formatters
	)

Indicate to a message router that we are interested in messages with one or more formatters.

The formatters are passed on to the underlying subscription.

Warning

With direct subscriptions, adding a formatter filter is an OR operation with any message type filters. In the current implementation of message router however, it's an AND operation. Even when setting a default route, the callback will only get messages that have the formatters provides in this call.

Parameters

router	Router to set the formatters of.
formatters	A bitmap of stasis_subscription_message_formatters we wish to receive.

Since

13.25.0

16.2.0

Definition at line 420 of file stasis_message_router.c.

References ast_assert, NULL, stasis_subscription_accept_formatters(), and stasis_message_router::subscription.

{
   ast_assert(router != NULL);

   stasis_subscription_accept_formatters(router->subscription, formatters);

   return;
}

stasis_message_router_add()

int stasis_message_router_add	(	struct stasis_message_router *	router,
		struct stasis_message_type *	message_type,
		stasis_subscription_cb	callback,
		void *	data
	)

Add a route to a message router.

A particular message_type may have at most one route per router. If you route stasis_cache_update messages, the callback will only receive updates for types not handled by routes added with stasis_message_router_add_cache_update().

Adding multiple routes for the same message type results in undefined behavior.

Parameters

router	Router to add the route to.
message_type	Type of message to route.
callback	Callback to forard messages of message_type to.
data	Data pointer to pass to callback.

Return values

0	on success
-1	on failure

Since

12

Definition at line 310 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, ast_assert, NULL, route_table_add(), stasis_message_router::routes, stasis_subscription_accept_message_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_set_filter(), and stasis_message_router::subscription.

Referenced by app_create(), AST_TEST_DEFINE(), create_routes(), endpoint_internal_create(), forwards_create_endpoint(), load_general_config(), load_module(), manager_bridging_init(), manager_channels_init(), manager_confbridge_init(), manager_endpoints_init(), manager_mwi_init(), manager_subscriptions_init(), meetme_stasis_init(), pjsip_outbound_registration_metrics_init(), and setup_stasis_subs().

{
   int res;

   ast_assert(router != NULL);

   if (!message_type) {
      /* Cannot route to NULL type. */
      return -1;
   }
   ao2_lock(router);
   res = route_table_add(&router->routes, message_type, callback, data);
   if (!res) {
      stasis_subscription_accept_message_type(router->subscription, message_type);
      /* Until a specific message type was added we would already drop the message, so being
       * selective now doesn't harm us. If we have a default route then we are already forced
       * to filter nothing and messages will come in regardless.
       */
      stasis_subscription_set_filter(router->subscription, STASIS_SUBSCRIPTION_FILTER_SELECTIVE);
   }
   ao2_unlock(router);
   return res;
}

stasis_message_router_add_cache_update()

int stasis_message_router_add_cache_update	(	struct stasis_message_router *	router,
		struct stasis_message_type *	message_type,
		stasis_subscription_cb	callback,
		void *	data
	)

Add a route for stasis_cache_update messages to a message router.

A particular message_type may have at most one cache route per router. These are distinct from regular routes, so one could have both a regular route and a cache route for the same message_type.

Adding multiple routes for the same message type results in undefined behavior.

Parameters

router	Router to add the route to.
message_type	Subtype of cache update to route.
callback	Callback to forard messages of message_type to.
data	Data pointer to pass to callback.

Return values

0	on success
-1	on failure

Since

12

Definition at line 336 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, ast_assert, stasis_message_router::cache_routes, NULL, route_table_add(), stasis_cache_update_type(), stasis_subscription_accept_message_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_set_filter(), and stasis_message_router::subscription.

Referenced by app_create(), and AST_TEST_DEFINE().

{
   int res;

   ast_assert(router != NULL);

   if (!message_type) {
      /* Cannot cache a route to NULL type. */
      return -1;
   }
   ao2_lock(router);
   res = route_table_add(&router->cache_routes, message_type, callback, data);
   if (!res) {
      stasis_subscription_accept_message_type(router->subscription, stasis_cache_update_type());
      stasis_subscription_set_filter(router->subscription, STASIS_SUBSCRIPTION_FILTER_SELECTIVE);
   }
   ao2_unlock(router);
   return res;
}

stasis_message_router_create_internal()

static struct stasis_message_router* stasis_message_router_create_internal ( struct stasis_topic *  topic, int  use_thread_pool, const char *  file, int  lineno, const char *  func  )

static

Definition at line 207 of file stasis_message_router.c.

References __stasis_subscribe(), __stasis_subscribe_pool(), ao2_ref, ao2_t_alloc, AST_VECTOR_INIT, stasis_message_router::cache_routes, NULL, router, router_dispatch(), router_dtor(), stasis_message_router::routes, stasis_subscription_accept_message_type(), stasis_subscription_change_type(), stasis_topic_name(), and stasis_message_router::subscription.

Referenced by __stasis_message_router_create(), and __stasis_message_router_create_pool().

{
   int res;
   struct stasis_message_router *router;

   router = ao2_t_alloc(sizeof(*router), router_dtor, stasis_topic_name(topic));
   if (!router) {
      return NULL;
   }

   res = 0;
   res |= AST_VECTOR_INIT(&router->routes, 0);
   res |= AST_VECTOR_INIT(&router->cache_routes, 0);
   if (res) {
      ao2_ref(router, -1);

      return NULL;
   }

   if (use_thread_pool) {
      router->subscription = __stasis_subscribe_pool(topic, router_dispatch, router, file, lineno, func);
   } else {
      router->subscription = __stasis_subscribe(topic, router_dispatch, router, file, lineno, func);
   }

   if (!router->subscription) {
      ao2_ref(router, -1);

      return NULL;
   }

   /* We need to receive subscription change messages so we know when our subscription goes away */
   stasis_subscription_accept_message_type(router->subscription, stasis_subscription_change_type());

   return router;
}

stasis_message_router_is_done()

int stasis_message_router_is_done

(

struct stasis_message_router *

router

)

Returns whether router has received its final message.

Parameters

router

Router.

Returns

True (non-zero) if stasis_subscription_final_message() has been received.

False (zero) if waiting for the end.

Definition at line 278 of file stasis_message_router.c.

References stasis_subscription_is_done(), and stasis_message_router::subscription.

Referenced by endpoint_dtor().

{
   if (!router) {
      /* Null router is about as done as you can get */
      return 1;
   }

   return stasis_subscription_is_done(router->subscription);
}

stasis_message_router_publish_sync()

void stasis_message_router_publish_sync	(	struct stasis_message_router *	router,
		struct stasis_message *	message
	)

Publish a message to a message router's subscription synchronously.

Parameters

router	Router
message	The Stasis Message Bus API message

This should be used when a message needs to be published synchronously to the underlying subscription created by a message router. This is analagous to stasis_publish_sync.

Note that the caller will be blocked until the thread servicing the message on the message router's subscription completes handling of the message.

Since

12.1.0

Definition at line 288 of file stasis_message_router.c.

References ao2_bump, ao2_cleanup, ast_assert, NULL, stasis_publish_sync(), and stasis_message_router::subscription.

Referenced by ast_cdr_engine_term(), cdr_prop_write(), cdr_read(), cdr_write(), forkcdr_exec(), and publish_app_cdr_message().

{
   ast_assert(router != NULL);

   ao2_bump(router);
   stasis_publish_sync(router->subscription, message);
   ao2_cleanup(router);
}

stasis_message_router_remove()

void stasis_message_router_remove	(	struct stasis_message_router *	router,
		struct stasis_message_type *	message_type
	)

Remove a route from a message router.

If a route is removed from another thread, there is no notification that all messages using this route have been processed. This typically means that the associated data pointer for this route must be kept until the route itself is disposed of.

Parameters

router	Router to remove the route from.
message_type	Type of message to route.

Since

12

Definition at line 358 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, ast_assert, NULL, route_table_remove(), and stasis_message_router::routes.

Referenced by cleanup_module(), load_general_config(), and unload_module().

{
   ast_assert(router != NULL);

   if (!message_type) {
      /* Cannot remove a NULL type. */
      return;
   }
   ao2_lock(router);
   route_table_remove(&router->routes, message_type);
   ao2_unlock(router);
}

stasis_message_router_remove_cache_update()

void stasis_message_router_remove_cache_update	(	struct stasis_message_router *	router,
		struct stasis_message_type *	message_type
	)

Remove a cache route from a message router.

If a route is removed from another thread, there is no notification that all messages using this route have been processed. This typically means that the associated data pointer for this route must be kept until the route itself is disposed of.

Parameters

router	Router to remove the route from.
message_type	Type of message to route.

Since

12

Definition at line 372 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, ast_assert, stasis_message_router::cache_routes, NULL, and route_table_remove().

{
   ast_assert(router != NULL);

   if (!message_type) {
      /* Cannot remove a NULL type. */
      return;
   }
   ao2_lock(router);
   route_table_remove(&router->cache_routes, message_type);
   ao2_unlock(router);
}

stasis_message_router_set_congestion_limits()

int stasis_message_router_set_congestion_limits	(	struct stasis_message_router *	router,
		long	low_water,
		long	high_water
	)

Set the high and low alert water marks of the stasis message router.

Since

13.10.0

Parameters

router	Pointer to a stasis message router
low_water	New queue low water mark. (-1 to set as 90% of high_water)
high_water	New queue high water mark.

Return values

0	on success.
-1	on error (water marks not changed).

Definition at line 298 of file stasis_message_router.c.

References stasis_subscription_set_congestion_limits(), and stasis_message_router::subscription.

Referenced by create_routes(), load_module(), and manager_subscriptions_init().

{
   int res = -1;

   if (router) {
      res = stasis_subscription_set_congestion_limits(router->subscription,
         low_water, high_water);
   }
   return res;
}

stasis_message_router_set_default()

int stasis_message_router_set_default	(	struct stasis_message_router *	router,
		stasis_subscription_cb	callback,
		void *	data
	)

Sets the default route of a router.

Parameters

router	Router to set the default route of.
callback	Callback to forward messages which otherwise have no home.
data	Data pointer to pass to callback.

Return values

0	on success
-1	on failure

Since

12

Note

Setting a default callback will automatically cause the underlying subscription to receive all messages and not be filtered. If filtering is desired then a specific route for each message type should be provided.

Definition at line 387 of file stasis_message_router.c.

References stasis_message_router_set_formatters_default(), and STASIS_SUBSCRIPTION_FORMATTER_NONE.

Referenced by AST_TEST_DEFINE(), load_module(), and setup_stasis_subs().

{
   stasis_message_router_set_formatters_default(router, callback, data, STASIS_SUBSCRIPTION_FORMATTER_NONE);

   /* While this implementation can never fail, it used to be able to */
   return 0;
}

stasis_message_router_set_formatters_default()

void stasis_message_router_set_formatters_default	(	struct stasis_message_router *	router,
		stasis_subscription_cb	callback,
		void *	data,
		enum stasis_subscription_message_formatters	formatters
	)

Sets the default route of a router with formatters.

Parameters

router	Router to set the default route of.
callback	Callback to forward messages which otherwise have no home.
data	Data pointer to pass to callback.
formatters	A bitmap of stasis_subscription_message_formatters we wish to receive.

Since

13.26.0

16.3.0

Note

If formatters are specified then the message router will remain in a selective filtering state. Any explicit routes will receive messages of their message type and the default callback will only receive messages that have one of the given formatters. Explicit routes will not be filtered according to the given formatters.

Definition at line 397 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, ast_assert, stasis_message_route::callback, stasis_message_route::data, stasis_message_router::default_route, NULL, stasis_subscription_accept_formatters(), STASIS_SUBSCRIPTION_FILTER_FORCED_NONE, STASIS_SUBSCRIPTION_FORMATTER_NONE, stasis_subscription_set_filter(), and stasis_message_router::subscription.

Referenced by app_create(), manager_subscriptions_init(), and stasis_message_router_set_default().

{
   ast_assert(router != NULL);
   ast_assert(callback != NULL);

   stasis_subscription_accept_formatters(router->subscription, formatters);

   ao2_lock(router);
   router->default_route.callback = callback;
   router->default_route.data = data;
   ao2_unlock(router);

   if (formatters == STASIS_SUBSCRIPTION_FORMATTER_NONE) {
      /* Formatters govern what messages the default callback get, so it is only if none is
       * specified that we accept all messages regardless.
       */
      stasis_subscription_set_filter(router->subscription, STASIS_SUBSCRIPTION_FILTER_FORCED_NONE);
   }
}

stasis_message_router_unsubscribe()

void stasis_message_router_unsubscribe

(

struct stasis_message_router *

router

)

Unsubscribe the router from the upstream topic.

Parameters

router

Router to unsubscribe.

Since

12

Definition at line 258 of file stasis_message_router.c.

References ao2_lock, ao2_unlock, stasis_unsubscribe(), and stasis_message_router::subscription.

Referenced by app_shutdown(), ast_endpoint_shutdown(), AST_TEST_DEFINE(), manager_confbridge_shutdown(), meetme_stasis_cleanup(), remove_stasis_subscriptions(), and setup_stasis_subs().

{
   if (!router) {
      return;
   }

   ao2_lock(router);
   router->subscription = stasis_unsubscribe(router->subscription);
   ao2_unlock(router);
}

stasis_message_router_unsubscribe_and_join()

void stasis_message_router_unsubscribe_and_join

(

struct stasis_message_router *

router

)

Unsubscribe the router from the upstream topic, blocking until the final message has been processed.

See stasis_unsubscribe_and_join() for info on when to use this vs. stasis_message_router_unsubscribe().

Parameters

router

Router to unsubscribe.

Since

12

Definition at line 269 of file stasis_message_router.c.

References stasis_unsubscribe_and_join(), and stasis_message_router::subscription.

Referenced by AST_TEST_DEFINE(), cdr_engine_shutdown(), cleanup_module(), destroy_routes(), manager_endpoints_shutdown(), manager_shutdown(), pjsip_outbound_registration_metrics_unload_cb(), and unload_module().

{
   if (!router) {
      return;
   }
   stasis_unsubscribe_and_join(router->subscription);
}