stasis.c File Reference

Stasis Message Bus API. More...

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/stasis_internal.h"
#include "asterisk/stasis.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/threadpool.h"
#include "asterisk/utils.h"
#include "asterisk/uuid.h"
#include "asterisk/vector.h"
#include "asterisk/stasis_channels.h"
#include "asterisk/stasis_bridges.h"
#include "asterisk/stasis_endpoints.h"
#include "asterisk/config_options.h"
#include "asterisk/cli.h"

Include dependency graph for stasis.c:

Go to the source code of this file.

Data Structures
struct	ast_multi_object_blob
	A multi object blob data structure to carry user event stasis messages. More...
struct	stasis_config
struct	stasis_declined_config
	A structure to hold global configuration-related options. More...
struct	stasis_forward
	Forwarding information. More...
struct	stasis_message_type_statistics
struct	stasis_subscription
struct	stasis_subscription_statistics
struct	stasis_threadpool_conf
	Threadpool configuration options. More...
struct	stasis_topic
struct	stasis_topic_pool
struct	sync_task_data
struct	topic_pool_entry
struct	topic_proxy

Macros
#define	FMT_FIELDS   "%-64s %-64s\n"
#define	FMT_FIELDS   "%-64s %10d %10d %16ld %16ld\n"
#define	FMT_FIELDS   "%-64s %10d %10d %10d %16ld %16ld\n"
#define	FMT_FIELDS   "%-64s %10d %10d\n"
#define	FMT_FIELDS2   "%-64s %10d %10d\n"
#define	FMT_FIELDS2   "%-64s %10s %10d %10d\n"
#define	FMT_HEADERS   "%-64s %-64s\n"
#define	FMT_HEADERS   "%-64s %10s %10s %16s %16s\n"
#define	FMT_HEADERS   "%-64s %10s %10s %10s %16s %16s\n"
#define	FMT_HEADERS   "%-64s %10s %10s\n"
#define	INITIAL_SUBSCRIBERS_MAX   4
#define	SUBSCRIPTION_STATISTICS_BUCKETS   57
#define	TOPIC_ALL_BUCKETS   997
#define	topic_lock_both(topic1, topic2)
	Lock two topics. More...
#define	TOPIC_POOL_BUCKETS   57
#define	TOPIC_STATISTICS_BUCKETS   57

Functions
struct stasis_subscription *	__stasis_subscribe (struct stasis_topic *topic, stasis_subscription_cb callback, void *data, const char *file, int lineno, const char *func)
	Create a subscription. More...
struct stasis_subscription *	__stasis_subscribe_pool (struct stasis_topic *topic, stasis_subscription_cb callback, void *data, const char *file, int lineno, const char *func)
	Create a subscription whose callbacks occur on a thread pool. More...
static	AO2_GLOBAL_OBJ_STATIC (topic_statistics)
static	AO2_GLOBAL_OBJ_STATIC (subscription_statistics)
static	AO2_GLOBAL_OBJ_STATIC (globals)
	A global object container that will contain the stasis_config that gets swapped out on reloads. More...
	AO2_STRING_FIELD_CASE_SORT_FN (topic_proxy, name)
	AO2_STRING_FIELD_CMP_FN (topic_proxy, name)
	AO2_STRING_FIELD_HASH_FN (topic_proxy, name)
	AO2_STRING_FIELD_SORT_FN (stasis_subscription_statistics, uniqueid)
	AO2_STRING_FIELD_SORT_FN (stasis_topic_statistics, name)
void	ast_multi_object_blob_add (struct ast_multi_object_blob *multi, enum stasis_user_multi_object_snapshot_type type, void *object)
	Add an object (snapshot) to the blob. More...
struct ast_multi_object_blob *	ast_multi_object_blob_create (struct ast_json *blob)
	Create a stasis user event multi object blob. More...
void	ast_multi_object_blob_single_channel_publish (struct ast_channel *chan, struct stasis_message_type *type, struct ast_json *blob)
	Publish single channel user event (for app_userevent compatibility) More...
static	AST_VECTOR (struct stasis_message_type_statistics)
	CONFIG_INFO_CORE ("stasis", cfg_info, globals, stasis_config_alloc,.files=ACO_FILES(&stasis_conf),)
	Register information about the configs being processed by this module. More...
static int	declined_handler (const struct aco_option *opt, struct ast_variable *var, void *obj)
static int	dispatch_exec_async (struct ast_taskprocessor_local *local)
static int	dispatch_exec_sync (struct ast_taskprocessor_local *local)
static unsigned int	dispatch_message (struct stasis_subscription *sub, struct stasis_message *message, int synchronous)
static void	forward_dtor (void *obj)
struct stasis_subscription *	internal_stasis_subscribe (struct stasis_topic *topic, stasis_subscription_cb callback, void *data, int needs_mailbox, int use_thread_pool, const char *file, int lineno, const char *func)
	Create a subscription. More...
static int	link_topic_proxy (struct stasis_topic *topic, const char *name, const char *detail)
static void	multi_object_blob_dtor (void *obj)
static struct ast_str *	multi_object_blob_to_ami (void *obj)
static struct ast_manager_event_blob *	multi_user_event_to_ami (struct stasis_message *message)
static struct ast_json *	multi_user_event_to_json (struct stasis_message *message, const struct stasis_message_sanitizer *sanitize)
static void	proxy_dtor (void *weakproxy, void *container)
static void	publish_msg (struct stasis_topic *topic, struct stasis_message *message, struct stasis_subscription *sync_sub)
static void	send_subscription_subscribe (struct stasis_topic *topic, struct stasis_subscription *sub)
static void	send_subscription_unsubscribe (struct stasis_topic *topic, struct stasis_subscription *sub)
static void	stasis_cleanup (void)
	Cleanup function for graceful shutdowns. More...
static void *	stasis_config_alloc (void)
static void	stasis_config_destructor (void *obj)
static void	stasis_declined_config_destructor (void *obj)
struct stasis_forward *	stasis_forward_all (struct stasis_topic *from_topic, struct stasis_topic *to_topic)
	Create a subscription which forwards all messages from one topic to another. More...
struct stasis_forward *	stasis_forward_cancel (struct stasis_forward *forward)
int	stasis_init (void)
	Initialize the Stasis subsystem. More...
void	stasis_log_bad_type_access (const char *name)
int	stasis_message_type_declined (const char *name)
	Check whether a message type is declined. More...
	STASIS_MESSAGE_TYPE_DEFN (stasis_subscription_change_type)
void	stasis_publish (struct stasis_topic *topic, struct stasis_message *message)
	Publish a message to a topic's subscribers. More...
void	stasis_publish_sync (struct stasis_subscription *sub, struct stasis_message *message)
	Publish a message to a topic's subscribers, synchronizing on the specified subscriber. More...
static char *	stasis_show_topic (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	stasis_show_topics (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
void	stasis_subscription_accept_formatters (struct stasis_subscription *subscription, enum stasis_subscription_message_formatters formatters)
	Indicate to a subscription that we are interested in messages with one or more formatters. More...
int	stasis_subscription_accept_message_type (struct stasis_subscription *subscription, const struct stasis_message_type *type)
	Indicate to a subscription that we are interested in a message type. More...
void	stasis_subscription_cb_noop (void *data, struct stasis_subscription *sub, struct stasis_message *message)
	Stasis subscription callback function that does nothing. More...
int	stasis_subscription_decline_message_type (struct stasis_subscription *subscription, const struct stasis_message_type *type)
	Indicate to a subscription that we are not interested in a message type. More...
int	stasis_subscription_final_message (struct stasis_subscription *sub, struct stasis_message *msg)
	Determine whether a message is the final message to be received on a subscription. More...
int	stasis_subscription_is_done (struct stasis_subscription *subscription)
	Returns whether subscription has received its final message. More...
int	stasis_subscription_is_subscribed (const struct stasis_subscription *sub)
	Returns whether a subscription is currently subscribed. More...
void	stasis_subscription_join (struct stasis_subscription *subscription)
	Block until the last message is processed on a subscription. More...
int	stasis_subscription_set_congestion_limits (struct stasis_subscription *subscription, long low_water, long high_water)
	Set the high and low alert water marks of the stasis subscription. More...
int	stasis_subscription_set_filter (struct stasis_subscription *subscription, enum stasis_subscription_message_filter filter)
	Set the message type filtering level on a subscription. More...
static struct stasis_subscription_statistics *	stasis_subscription_statistics_create (struct stasis_subscription *sub, int needs_mailbox, int use_thread_pool, const char *file, int lineno, const char *func)
const char *	stasis_subscription_uniqueid (const struct stasis_subscription *sub)
	Get the unique ID for the subscription. More...
struct stasis_topic *	stasis_topic_create (const char *name)
	Create a new topic. More...
struct stasis_topic *	stasis_topic_create_with_detail (const char *name, const char *detail)
	Create a new topic with given detail. More...
const char *	stasis_topic_detail (const struct stasis_topic *topic)
	Return the detail of a topic. More...
struct stasis_topic *	stasis_topic_get (const char *name)
	Get a topic of the given name. More...
const char *	stasis_topic_name (const struct stasis_topic *topic)
	Return the name of a topic. More...
struct stasis_topic_pool *	stasis_topic_pool_create (struct stasis_topic *pooled_topic)
	Create a topic pool that routes messages from dynamically generated topics to the given topic. More...
void	stasis_topic_pool_delete_topic (struct stasis_topic_pool *pool, const char *topic_name)
	Delete a topic from the topic pool. More...
struct stasis_topic *	stasis_topic_pool_get_topic (struct stasis_topic_pool *pool, const char *topic_name)
	Find or create a topic in the pool. More...
int	stasis_topic_pool_topic_exists (const struct stasis_topic_pool *pool, const char *topic_name)
	Check if a topic exists in a pool. More...
static struct stasis_topic_statistics *	stasis_topic_statistics_create (struct stasis_topic *topic)
size_t	stasis_topic_subscribers (const struct stasis_topic *topic)
	Return the number of subscribers of a topic. More...
struct stasis_subscription *	stasis_unsubscribe (struct stasis_subscription *sub)
	Cancel a subscription. More...
struct stasis_subscription *	stasis_unsubscribe_and_join (struct stasis_subscription *subscription)
	Cancel a subscription, blocking until the last message is processed. More...
static char *	statistics_show_messages (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	statistics_show_subscription (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	statistics_show_subscriptions (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	statistics_show_topic (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	statistics_show_topics (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static int	sub_cleanup (void *data)
static struct stasis_subscription_change *	subscription_change_alloc (struct stasis_topic *topic, const char *uniqueid, const char *description)
static void	subscription_change_dtor (void *obj)
static void	subscription_dtor (void *obj)
static void	subscription_invoke (struct stasis_subscription *sub, struct stasis_message *message)
	Invoke the subscription's callback. More...
static int	subscription_statistics_cmp (void *obj, void *arg, int flags)
static char *	subscription_statistics_complete_name (const char *word, int state)
static void	subscription_statistics_destroy (void *obj)
static int	subscription_statistics_hash (const void *obj, const int flags)
static int	topic_add_subscription (struct stasis_topic *topic, struct stasis_subscription *sub)
	Add a subscriber to a topic. More...
static char *	topic_complete_name (const char *word)
static void	topic_dtor (void *obj)
static void	topic_pool_dtor (void *obj)
static struct topic_pool_entry *	topic_pool_entry_alloc (const char *topic_name)
static int	topic_pool_entry_cmp (void *obj, void *arg, int flags)
static void	topic_pool_entry_dtor (void *obj)
static int	topic_pool_entry_hash (const void *obj, const int flags)
static int	topic_remove_subscription (struct stasis_topic *topic, struct stasis_subscription *sub)
static int	topic_statistics_cmp (void *obj, void *arg, int flags)
static char *	topic_statistics_complete_name (const char *word, int state)
static void	topic_statistics_destroy (void *obj)
static int	topic_statistics_hash (const void *obj, const int flags)
static int	userevent_exclusion_cb (const char *key)
	STASIS_MESSAGE_TYPE_DEFN (ast_multi_user_event_type,.to_json=multi_user_event_to_json,.to_ami=multi_user_event_to_ami,)
	Define multi user event message type(s). More...

Variables
static struct ast_cli_entry	cli_stasis []
static struct ast_cli_entry	cli_stasis_statistics []
static struct aco_type	declined_option
	An aco_type structure to link the "declined_message_types" category to the stasis_declined_config type. More...
struct aco_type *	declined_options [] = ACO_TYPES(&declined_option)
static ast_mutex_t	message_type_statistics_lock = { PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP , NULL, {1, 0} }
struct aco_file	stasis_conf
static struct ast_threadpool *	threadpool
static struct aco_type	threadpool_option
static struct aco_type *	threadpool_options [] = ACO_TYPES(&threadpool_option)
struct ao2_container *	topic_all

Detailed Description

Stasis Message Bus API.

Author

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

Definition in file stasis.c.

Macro Definition Documentation

FMT_FIELDS [1/4]

#define FMT_FIELDS   "%-64s %-64s\n"

Referenced by stasis_show_topics(), statistics_show_messages(), statistics_show_subscriptions(), and statistics_show_topics().

FMT_FIELDS [2/4]

#define FMT_FIELDS   "%-64s %10d %10d %16ld %16ld\n"

FMT_FIELDS [3/4]

#define FMT_FIELDS   "%-64s %10d %10d %10d %16ld %16ld\n"

FMT_FIELDS [4/4]

#define FMT_FIELDS   "%-64s %10d %10d\n"

FMT_FIELDS2 [1/2]

#define FMT_FIELDS2   "%-64s %10d %10d\n"

Referenced by statistics_show_subscriptions(), and statistics_show_topics().

FMT_FIELDS2 [2/2]

#define FMT_FIELDS2   "%-64s %10s %10d %10d\n"

FMT_HEADERS [1/4]

#define FMT_HEADERS   "%-64s %-64s\n"

Referenced by stasis_show_topics(), statistics_show_messages(), statistics_show_subscriptions(), and statistics_show_topics().

FMT_HEADERS [2/4]

#define FMT_HEADERS   "%-64s %10s %10s %16s %16s\n"

FMT_HEADERS [3/4]

#define FMT_HEADERS   "%-64s %10s %10s %10s %16s %16s\n"

FMT_HEADERS [4/4]

#define FMT_HEADERS   "%-64s %10s %10s\n"

INITIAL_SUBSCRIBERS_MAX

#define INITIAL_SUBSCRIBERS_MAX   4

Initial size of the subscribers list.

Definition at line 302 of file stasis.c.

Referenced by stasis_topic_create_with_detail().

SUBSCRIPTION_STATISTICS_BUCKETS

#define SUBSCRIPTION_STATISTICS_BUCKETS   57

The number of buckets to use for subscription statistics

Definition at line 328 of file stasis.c.

Referenced by stasis_init().

TOPIC_ALL_BUCKETS

#define TOPIC_ALL_BUCKETS   997

Definition at line 318 of file stasis.c.

Referenced by stasis_init().

topic_lock_both

#define topic_lock_both	(		topic1,
			topic2
	)

Lock two topics.

Definition at line 426 of file stasis.c.

Referenced by stasis_forward_all(), and stasis_forward_cancel().

TOPIC_POOL_BUCKETS

#define TOPIC_POOL_BUCKETS   57

The number of buckets to use for topic pools

Definition at line 305 of file stasis.c.

Referenced by stasis_topic_pool_create().

TOPIC_STATISTICS_BUCKETS

#define TOPIC_STATISTICS_BUCKETS   57

The number of buckets to use for topic statistics

Definition at line 325 of file stasis.c.

Referenced by stasis_init().

Function Documentation

__stasis_subscribe()

struct stasis_subscription* __stasis_subscribe	(	struct stasis_topic *	topic,
		stasis_subscription_cb	callback,
		void *	data,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a subscription.

In addition to being AO2 managed memory (requiring an ao2_cleanup() to free up this reference), the subscription must be explicitly unsubscribed from its topic using stasis_unsubscribe().

The invocations of the callback are serialized, but may not always occur on the same thread. The invocation order of different subscriptions is unspecified.

Parameters

topic	Topic to subscribe to.
callback	Callback function for subscription messages.
data	Data to be passed to the callback, in addition to the message.

Returns

New stasis_subscription object.

NULL on error.

Since

12

Note

This callback will receive a callback with a message indicating it has been subscribed. This occurs immediately before accepted message types can be set and the callback must expect to receive it.

Definition at line 944 of file stasis.c.

References internal_stasis_subscribe().

Referenced by stasis_message_router_create_internal().

{
   return internal_stasis_subscribe(topic, callback, data, 1, 0, file, lineno, func);
}

__stasis_subscribe_pool()

struct stasis_subscription* __stasis_subscribe_pool	(	struct stasis_topic *	topic,
		stasis_subscription_cb	callback,
		void *	data,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a subscription whose callbacks occur on a thread pool.

In addition to being AO2 managed memory (requiring an ao2_cleanup() to free up this reference), the subscription must be explicitly unsubscribed from its topic using stasis_unsubscribe().

The invocations of the callback are serialized, but will almost certainly not always happen on the same thread. The invocation order of different subscriptions is unspecified.

Unlike stasis_subscribe, this function will explicitly use a threadpool to dispatch items to its callback. This form of subscription should be used when many subscriptions may be made to the specified topic.

Parameters

topic	Topic to subscribe to.
callback	Callback function for subscription messages.
data	Data to be passed to the callback, in addition to the message.

Returns

New stasis_subscription object.

NULL on error.

Since

12.8.0

Note

This callback will receive a callback with a message indicating it has been subscribed. This occurs immediately before accepted message types can be set and the callback must expect to receive it.

Definition at line 955 of file stasis.c.

References internal_stasis_subscribe().

Referenced by stasis_message_router_create_internal().

{
   return internal_stasis_subscribe(topic, callback, data, 1, 1, file, lineno, func);
}

AO2_GLOBAL_OBJ_STATIC() [1/3]

static AO2_GLOBAL_OBJ_STATIC ( topic_statistics  )

static

Global container which stores statistics for topics

AO2_GLOBAL_OBJ_STATIC() [2/3]

static AO2_GLOBAL_OBJ_STATIC ( subscription_statistics  )

static

Global container which stores statistics for subscriptions

AO2_GLOBAL_OBJ_STATIC() [3/3]

static AO2_GLOBAL_OBJ_STATIC ( globals  )

static

A global object container that will contain the stasis_config that gets swapped out on reloads.

AO2_STRING_FIELD_CASE_SORT_FN()

AO2_STRING_FIELD_CASE_SORT_FN	(	topic_proxy	,
		name
	)

AO2_STRING_FIELD_CMP_FN()

AO2_STRING_FIELD_CMP_FN	(	topic_proxy	,
		name
	)

AO2_STRING_FIELD_HASH_FN()

AO2_STRING_FIELD_HASH_FN	(	topic_proxy	,
		name
	)

AO2_STRING_FIELD_SORT_FN() [1/2]

AO2_STRING_FIELD_SORT_FN	(	stasis_subscription_statistics	,
		uniqueid
	)

Referenced by statistics_show_subscription().

AO2_STRING_FIELD_SORT_FN() [2/2]

AO2_STRING_FIELD_SORT_FN	(	stasis_topic_statistics	,
		name
	)

AST_VECTOR()

static AST_VECTOR ( struct stasis_message_type_statistics  )

static

Vector containing message type information

Highest time spent dispatching messages to subscribers

Lowest time spent dispatching messages to subscribers

The number of messages that were not dispatched to any subscriber

The number of messages that were dispatched to at least 1 subscriber

The ids of the subscribers to this topic

Pointer to the topic (NOT refcounted, and must NOT be accessed)

Name of the topic

Definition at line 350 of file stasis.c.

References name.

                               {
   /*! \brief Highest time spent dispatching messages to subscribers */
   long highest_time_dispatched;
   /*! \brief Lowest time spent dispatching messages to subscribers */
   long lowest_time_dispatched;
   /*! \brief The number of messages that were not dispatched to any subscriber */
   int messages_not_dispatched;
   /*! \brief The number of messages that were dispatched to at least 1 subscriber */
   int messages_dispatched;
   /*! \brief The ids of the subscribers to this topic */
   struct ao2_container *subscribers;
   /*! \brief Pointer to the topic (NOT refcounted, and must NOT be accessed) */
   struct stasis_topic *topic;
   /*! \brief Name of the topic */
   char name[0];
};

CONFIG_INFO_CORE()

CONFIG_INFO_CORE	(	"stasis"	,
		cfg_info	,
		globals	,
		stasis_config_alloc	,
		.	files = ACO_FILES(&stasis_conf)
	)

Register information about the configs being processed by this module.

declined_handler()

static int declined_handler ( const struct aco_option *  opt, struct ast_variable *  var, void *  obj  )

static

Definition at line 2304 of file stasis.c.

References ast_multi_user_event_type(), ast_str_container_add(), ast_strlen_zero, stasis_declined_config::declined, multi_user_event_to_ami(), multi_user_event_to_json(), STASIS_MESSAGE_TYPE_DEFN(), to_ami(), and ast_variable::value.

Referenced by stasis_init().

{
   struct stasis_declined_config *declined = obj;

   if (ast_strlen_zero(var->value)) {
      return 0;
   }

   if (ast_str_container_add(declined->declined, var->value)) {
      return -1;
   }

   return 0;
}

dispatch_exec_async()

static int dispatch_exec_async ( struct ast_taskprocessor_local *  local )

static

Definition at line 1261 of file stasis.c.

References ao2_cleanup, ast_taskprocessor_local::data, ast_taskprocessor_local::local_data, and subscription_invoke().

Referenced by dispatch_message().

{
   struct stasis_subscription *sub = local->local_data;
   struct stasis_message *message = local->data;

   subscription_invoke(sub, message);
   ao2_cleanup(message);

   return 0;
}

dispatch_exec_sync()

static int dispatch_exec_sync ( struct ast_taskprocessor_local *  local )

static

Definition at line 1288 of file stasis.c.

References ao2_cleanup, ast_cond_signal, ast_mutex_lock, ast_mutex_unlock, sync_task_data::complete, sync_task_data::cond, ast_taskprocessor_local::data, ast_taskprocessor_local::local_data, sync_task_data::lock, subscription_invoke(), and sync_task_data::task_data.

Referenced by dispatch_message().

{
   struct stasis_subscription *sub = local->local_data;
   struct sync_task_data *std = local->data;
   struct stasis_message *message = std->task_data;

   subscription_invoke(sub, message);
   ao2_cleanup(message);

   ast_mutex_lock(&std->lock);
   std->complete = 1;
   ast_cond_signal(&std->cond);
   ast_mutex_unlock(&std->lock);

   return 0;
}

dispatch_message()

static unsigned int dispatch_message ( struct stasis_subscription *  sub, struct stasis_message *  message, int  synchronous  )

static

Definition at line 1314 of file stasis.c.

References ao2_bump, ao2_cleanup, ast_atomic_fetchadd_int(), ast_cond_destroy, ast_cond_init, ast_cond_wait, ast_log, ast_mutex_destroy, ast_mutex_init, ast_mutex_lock, ast_mutex_unlock, ast_taskprocessor_push_local(), AST_VECTOR_GET, AST_VECTOR_SIZE, sync_task_data::complete, sync_task_data::cond, dispatch_exec_async(), dispatch_exec_sync(), sync_task_data::lock, LOG_ERROR, stasis_subscription::mailbox, stasis_message_type_statistics::message_type, NULL, stasis_message_type(), stasis_message_type_available_formatters(), stasis_message_type_id(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_final_message(), STASIS_SUBSCRIPTION_FORMATTER_NONE, subscription_invoke(), and sync_task_data::task_data.

Referenced by publish_msg(), and send_subscription_unsubscribe().

{
   int is_final = stasis_subscription_final_message(sub, message);

   /*
    * The 'do while' gives us an easy way to skip remaining logic once
    * we determine the message should be accepted.
    * The code looks more verbose than it needs to be but it optimizes
    * down very nicely.  It's just easier to understand and debug this way.
    */
   do {
      struct stasis_message_type *message_type = stasis_message_type(message);
      int type_id = stasis_message_type_id(message_type);
      int type_filter_specified = 0;
      int formatter_filter_specified = 0;
      int type_filter_passed = 0;
      int formatter_filter_passed = 0;

      /* We always accept final messages so only run the filter logic if not final */
      if (is_final) {
         break;
      }

      type_filter_specified = sub->filter & STASIS_SUBSCRIPTION_FILTER_SELECTIVE;
      formatter_filter_specified = sub->accepted_formatters != STASIS_SUBSCRIPTION_FORMATTER_NONE;

      /* Accept if no filters of either type were specified */
      if (!type_filter_specified && !formatter_filter_specified) {
         break;
      }

      type_filter_passed = type_filter_specified
         && type_id < AST_VECTOR_SIZE(&sub->accepted_message_types)
         && AST_VECTOR_GET(&sub->accepted_message_types, type_id);

      /*
       * Since the type and formatter filters are OR'd, we can skip
       * the formatter check if the type check passes.
       */
      if (type_filter_passed) {
         break;
      }

      formatter_filter_passed = formatter_filter_specified
         && (sub->accepted_formatters & stasis_message_type_available_formatters(message_type));

      if (formatter_filter_passed) {
         break;
      }

#ifdef AST_DEVMODE
      ast_atomic_fetchadd_int(&sub->statistics->messages_dropped, +1);
#endif

      return 0;

   } while (0);

#ifdef AST_DEVMODE
   ast_atomic_fetchadd_int(&sub->statistics->messages_passed, +1);
#endif

   if (!sub->mailbox) {
      /* Dispatch directly */
      subscription_invoke(sub, message);
      return 1;
   }

   /* Bump the message for the taskprocessor push. This will get de-ref'd
    * by the task processor callback.
    */
   ao2_bump(message);
   if (!synchronous) {
      if (ast_taskprocessor_push_local(sub->mailbox, dispatch_exec_async, message)) {
         /* Push failed; ugh. */
         ast_log(LOG_ERROR, "Dropping async dispatch\n");
         ao2_cleanup(message);
         return 0;
      }
   } else {
      struct sync_task_data std;

      ast_mutex_init(&std.lock);
      ast_cond_init(&std.cond, NULL);
      std.complete = 0;
      std.task_data = message;

      if (ast_taskprocessor_push_local(sub->mailbox, dispatch_exec_sync, &std)) {
         /* Push failed; ugh. */
         ast_log(LOG_ERROR, "Dropping sync dispatch\n");
         ao2_cleanup(message);
         ast_mutex_destroy(&std.lock);
         ast_cond_destroy(&std.cond);
         return 0;
      }

      ast_mutex_lock(&std.lock);
      while (!std.complete) {
         ast_cond_wait(&std.cond, &std.lock);
      }
      ast_mutex_unlock(&std.lock);

      ast_mutex_destroy(&std.lock);
      ast_cond_destroy(&std.cond);
   }

   return 1;
}

forward_dtor()

static void forward_dtor ( void *  obj )

static

Definition at line 1538 of file stasis.c.

References ao2_cleanup, stasis_forward::from_topic, NULL, and stasis_forward::to_topic.

Referenced by stasis_forward_all().

{
   struct stasis_forward *forward = obj;

   ao2_cleanup(forward->from_topic);
   forward->from_topic = NULL;
   ao2_cleanup(forward->to_topic);
   forward->to_topic = NULL;
}

internal_stasis_subscribe()

struct stasis_subscription* internal_stasis_subscribe	(	struct stasis_topic *	topic,
		stasis_subscription_cb	callback,
		void *	data,
		int	needs_mailbox,
		int	use_thread_pool,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a subscription.

In addition to being AO2 managed memory (requiring an ao2_cleanup() to free up this reference), the subscription must be explicitly unsubscribed from its topic using stasis_unsubscribe().

The invocations of the callback are serialized, but may not always occur on the same thread. The invocation order of different subscriptions is unspecified.

Note: modules outside of Stasis should use stasis_subscribe.

Parameters

topic	Topic to subscribe to.
callback	Callback function for subscription messages.
data	Data to be passed to the callback, in addition to the message.
needs_mailbox	Determines whether or not the subscription requires a mailbox. Subscriptions with mailboxes will be delivered on some non-publisher thread; subscriptions without mailboxes will be delivered on the publisher thread.
use_thread_pool	Use the thread pool for the subscription. This is only relevant if needs_mailbox is non-zero.

Returns

New stasis_subscription object.

NULL on error.

Since

12

Definition at line 858 of file stasis.c.

References ao2_ref, ao2_t_alloc, ast_asprintf, ast_atomic_fetchadd_int(), ast_cond_init, ast_taskprocessor_build_name(), ast_taskprocessor_get(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_set_local(), ast_threadpool_serializer(), AST_VECTOR_INIT, stasis_subscription::callback, stasis_subscription::data, stasis_subscription::join_cond, stasis_subscription::mailbox, NULL, send_subscription_subscribe(), STASIS_SUBSCRIPTION_FILTER_NONE, STASIS_SUBSCRIPTION_FORMATTER_NONE, stasis_subscription_statistics_create(), stasis_topic_name(), statistics(), sub, subscription_dtor(), stasis_subscription::topic, topic_add_subscription(), TPS_REF_DEFAULT, and stasis_subscription::uniqueid.

Referenced by __stasis_subscribe(), __stasis_subscribe_pool(), and stasis_caching_topic_create().

{
   struct stasis_subscription *sub;
   int ret;

   if (!topic) {
      return NULL;
   }

   /* The ao2 lock is used for join_cond. */
   sub = ao2_t_alloc(sizeof(*sub), subscription_dtor, stasis_topic_name(topic));
   if (!sub) {
      return NULL;
   }

#ifdef AST_DEVMODE
   ret = ast_asprintf(&sub->uniqueid, "%s:%s-%d", file, stasis_topic_name(topic), ast_atomic_fetchadd_int(&topic->subscriber_id, +1));
   sub->statistics = stasis_subscription_statistics_create(sub, needs_mailbox, use_thread_pool, file, lineno, func);
   if (ret < 0 || !sub->statistics) {
      ao2_ref(sub, -1);
      return NULL;
   }
#else
   ret = ast_asprintf(&sub->uniqueid, "%s-%d", stasis_topic_name(topic), ast_atomic_fetchadd_int(&topic->subscriber_id, +1));
   if (ret < 0) {
      ao2_ref(sub, -1);
      return NULL;
   }
#endif

   if (needs_mailbox) {
      char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1];

      /* Create name with seq number appended. */
      ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "stasis/%c:%s",
         use_thread_pool ? 'p' : 'm',
         stasis_topic_name(topic));

      /*
       * With a small number of subscribers, a thread-per-sub is
       * acceptable. For a large number of subscribers, a thread
       * pool should be used.
       */
      if (use_thread_pool) {
         sub->mailbox = ast_threadpool_serializer(tps_name, threadpool);
      } else {
         sub->mailbox = ast_taskprocessor_get(tps_name, TPS_REF_DEFAULT);
      }
      if (!sub->mailbox) {
         ao2_ref(sub, -1);

         return NULL;
      }
      ast_taskprocessor_set_local(sub->mailbox, sub);
      /* Taskprocessor has a reference */
      ao2_ref(sub, +1);
   }

   ao2_ref(topic, +1);
   sub->topic = topic;
   sub->callback = callback;
   sub->data = data;
   ast_cond_init(&sub->join_cond, NULL);
   sub->filter = STASIS_SUBSCRIPTION_FILTER_NONE;
   AST_VECTOR_INIT(&sub->accepted_message_types, 0);
   sub->accepted_formatters = STASIS_SUBSCRIPTION_FORMATTER_NONE;

   if (topic_add_subscription(topic, sub) != 0) {
      ao2_ref(sub, -1);
      ao2_ref(topic, -1);

      return NULL;
   }
   send_subscription_subscribe(topic, sub);

   return sub;
}

link_topic_proxy()

static int link_topic_proxy ( struct stasis_topic *  topic, const char *  name, const char *  detail  )

static

Definition at line 501 of file stasis.c.

References ao2_bump, ao2_cleanup, ao2_link_flags, ao2_ref, ao2_t_weakproxy_alloc, ao2_t_weakproxy_set_object, ao2_unlock, ao2_weakproxy_subscribe(), ao2_wrlock, ast_copy_string(), ast_log, ast_tvnow(), topic_proxy::buf, topic_proxy::creationtime, topic_proxy::detail, LOG_ERROR, topic_proxy::name, NULL, OBJ_NOLOCK, proxy_dtor(), and stasis_topic_get().

Referenced by stasis_topic_create_with_detail().

{
   struct topic_proxy *proxy;
   struct stasis_topic* topic_tmp;
   size_t detail_len;

   if (!topic || !name || !strlen(name) || !detail) {
      return -1;
   }

   ao2_wrlock(topic_all);

   topic_tmp = stasis_topic_get(name);
   if (topic_tmp) {
      ast_log(LOG_ERROR, "The same topic is already exist. name: %s\n", name);
      ao2_ref(topic_tmp, -1);
      ao2_unlock(topic_all);

      return -1;
   }

   detail_len = strlen(detail) + 1;

   proxy = ao2_t_weakproxy_alloc(
         sizeof(*proxy) + strlen(name) + 1 + detail_len, NULL, name);
   if (!proxy) {
      ao2_unlock(topic_all);

      return -1;
   }

   /* set the proxy info */
   proxy->name = proxy->buf;
   proxy->detail = proxy->name + strlen(name) + 1;

   strcpy(proxy->name, name); /* SAFE */
   ast_copy_string(proxy->detail, detail, detail_len); /* SAFE */
   proxy->creationtime = ast_tvnow();

   /* We have exclusive access to proxy, no need for locking here. */
   if (ao2_t_weakproxy_set_object(proxy, topic, OBJ_NOLOCK, "weakproxy link")) {
      ao2_cleanup(proxy);
      ao2_unlock(topic_all);

      return -1;
   }

   if (ao2_weakproxy_subscribe(proxy, proxy_dtor, ao2_bump(topic_all), OBJ_NOLOCK)) {
      ao2_cleanup(proxy);
      ao2_unlock(topic_all);
      ao2_cleanup(topic_all);

      return -1;
   }

   /* setting the topic point to the proxy */
   topic->name = proxy->name;
   topic->detail = proxy->detail;
   topic->creationtime = &(proxy->creationtime);

   ao2_link_flags(topic_all, proxy, OBJ_NOLOCK);
   ao2_ref(proxy, -1);

   ao2_unlock(topic_all);

   return 0;
}

multi_object_blob_dtor()

static void multi_object_blob_dtor ( void *  obj )

static

Definition at line 1959 of file stasis.c.

References ao2_cleanup, ast_json_unref(), AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_SIZE, ast_multi_object_blob::blob, STASIS_UMOS_MAX, and type.

Referenced by ast_multi_object_blob_create().

{
   struct ast_multi_object_blob *multi = obj;
   int type;
   int i;

   for (type = 0; type < STASIS_UMOS_MAX; ++type) {
      for (i = 0; i < AST_VECTOR_SIZE(&multi->snapshots[type]); ++i) {
         ao2_cleanup(AST_VECTOR_GET(&multi->snapshots[type], i));
      }
      AST_VECTOR_FREE(&multi->snapshots[type]);
   }
   ast_json_unref(multi->blob);
}

multi_object_blob_to_ami()

static struct ast_str* multi_object_blob_to_ami ( void *  obj )

static

Definition at line 2096 of file stasis.c.

References ast_asprintf, ast_free, ast_manager_build_bridge_state_string_prefix(), ast_manager_build_channel_state_string_prefix(), ast_str_append(), ast_str_buffer(), ast_str_create, AST_VECTOR_GET, AST_VECTOR_SIZE, name, NULL, STASIS_UMOS_BRIDGE, STASIS_UMOS_CHANNEL, STASIS_UMOS_ENDPOINT, STASIS_UMOS_MAX, and type.

Referenced by multi_user_event_to_ami().

{
   struct ast_str *ami_str=ast_str_create(1024);
   struct ast_str *ami_snapshot;
   const struct ast_multi_object_blob *multi = obj;
   enum stasis_user_multi_object_snapshot_type type;
   int i;

   if (!ami_str) {
      return NULL;
   }
   if (!multi) {
      ast_free(ami_str);
      return NULL;
   }

   for (type = 0; type < STASIS_UMOS_MAX; ++type) {
      for (i = 0; i < AST_VECTOR_SIZE(&multi->snapshots[type]); ++i) {
         char *name = NULL;
         void *snapshot = AST_VECTOR_GET(&multi->snapshots[type], i);
         ami_snapshot = NULL;

         if (i > 0) {
            ast_asprintf(&name, "%d", i + 1);
         }

         switch (type) {
         case STASIS_UMOS_CHANNEL:
            ami_snapshot = ast_manager_build_channel_state_string_prefix(snapshot, name ?: "");
            break;

         case STASIS_UMOS_BRIDGE:
            ami_snapshot = ast_manager_build_bridge_state_string_prefix(snapshot, name ?: "");
            break;

         case STASIS_UMOS_ENDPOINT:
            /* currently not sending endpoint snapshots to AMI */
            break;
         }
         if (ami_snapshot) {
            ast_str_append(&ami_str, 0, "%s", ast_str_buffer(ami_snapshot));
            ast_free(ami_snapshot);
         }
         ast_free(name);
      }
   }

   return ami_str;
}

multi_user_event_to_ami()

static struct ast_manager_event_blob* multi_user_event_to_ami ( struct stasis_message *  message )

static

Definition at line 2155 of file stasis.c.

References ast_free, ast_json_object_get(), ast_json_string_get(), ast_manager_event_blob_create(), ast_manager_str_from_json_object(), ast_str_buffer(), ast_multi_object_blob::blob, EVENT_FLAG_USER, multi_object_blob_to_ami(), NULL, RAII_VAR, stasis_message_data(), and userevent_exclusion_cb().

Referenced by declined_handler().

{
   RAII_VAR(struct ast_str *, object_string, NULL, ast_free);
   RAII_VAR(struct ast_str *, body, NULL, ast_free);
   struct ast_multi_object_blob *multi = stasis_message_data(message);
   const char *eventname;

   eventname = ast_json_string_get(ast_json_object_get(multi->blob, "eventname"));
   body = ast_manager_str_from_json_object(multi->blob, userevent_exclusion_cb);
   object_string = multi_object_blob_to_ami(multi);
   if (!object_string || !body) {
      return NULL;
   }

   return ast_manager_event_blob_create(EVENT_FLAG_USER, "UserEvent",
      "%s"
      "UserEvent: %s\r\n"
      "%s",
      ast_str_buffer(object_string),
      eventname,
      ast_str_buffer(body));
}

multi_user_event_to_json()

static struct ast_json* multi_user_event_to_json ( struct stasis_message *  message, const struct stasis_message_sanitizer *  sanitize  )

static

Definition at line 2045 of file stasis.c.

References ast_bridge_snapshot_to_json(), ast_channel_snapshot_to_json(), ast_endpoint_snapshot_to_json(), ast_json_object_create(), ast_json_object_get(), ast_json_object_set(), ast_json_ref(), ast_json_string_create(), ast_json_timeval(), AST_VECTOR_GET, AST_VECTOR_SIZE, ast_multi_object_blob::blob, name, NULL, out, stasis_message_data(), stasis_message_timestamp(), STASIS_UMOS_BRIDGE, STASIS_UMOS_CHANNEL, STASIS_UMOS_ENDPOINT, STASIS_UMOS_MAX, and type.

Referenced by declined_handler().

{
   struct ast_json *out;
   struct ast_multi_object_blob *multi = stasis_message_data(message);
   struct ast_json *blob = multi->blob;
   const struct timeval *tv = stasis_message_timestamp(message);
   enum stasis_user_multi_object_snapshot_type type;
   int i;

   out = ast_json_object_create();
   if (!out) {
      return NULL;
   }

   ast_json_object_set(out, "type", ast_json_string_create("ChannelUserevent"));
   ast_json_object_set(out, "timestamp", ast_json_timeval(*tv, NULL));
   ast_json_object_set(out, "eventname", ast_json_ref(ast_json_object_get(blob, "eventname")));
   ast_json_object_set(out, "userevent", ast_json_ref(blob));

   for (type = 0; type < STASIS_UMOS_MAX; ++type) {
      for (i = 0; i < AST_VECTOR_SIZE(&multi->snapshots[type]); ++i) {
         struct ast_json *json_object = NULL;
         char *name = NULL;
         void *snapshot = AST_VECTOR_GET(&multi->snapshots[type], i);

         switch (type) {
         case STASIS_UMOS_CHANNEL:
            json_object = ast_channel_snapshot_to_json(snapshot, sanitize);
            name = "channel";
            break;
         case STASIS_UMOS_BRIDGE:
            json_object = ast_bridge_snapshot_to_json(snapshot, sanitize);
            name = "bridge";
            break;
         case STASIS_UMOS_ENDPOINT:
            json_object = ast_endpoint_snapshot_to_json(snapshot, sanitize);
            name = "endpoint";
            break;
         }
         if (json_object) {
            ast_json_object_set(out, name, json_object);
         }
      }
   }

   return out;
}

proxy_dtor()

static void proxy_dtor ( void *  weakproxy, void *  container  )

static

Definition at line 413 of file stasis.c.

References ao2_cleanup, ao2_unlink, sub, topic_add_subscription(), and topic_remove_subscription().

Referenced by link_topic_proxy().

{
   ao2_unlink(container, weakproxy);
   ao2_cleanup(container);
}

publish_msg()

static void publish_msg ( struct stasis_topic *  topic, struct stasis_message *  message, struct stasis_subscription *  sync_sub  )

static

Definition at line 1432 of file stasis.c.

References ao2_lock, ao2_ref, ao2_unlock, ast_assert, ast_atomic_fetchadd_int(), ast_mutex_lock, ast_mutex_unlock, ast_tvdiff_ms(), ast_tvnow(), AST_VECTOR_GET, AST_VECTOR_GET_ADDR, AST_VECTOR_REPLACE, AST_VECTOR_SIZE, dispatch_message(), stasis_message_type_statistics::message_type, message_type_id, message_type_statistics_lock, NULL, stasis_message_type_statistics::published, stasis_message_type(), stasis_message_type_id(), stasis_topic_subscribers(), statistics(), and stasis_message_type_statistics::unused.

Referenced by stasis_publish(), and stasis_publish_sync().

{
   size_t i;
   unsigned int dispatched = 0;
#ifdef AST_DEVMODE
   int message_type_id = stasis_message_type_id(stasis_message_type(message));
   struct stasis_message_type_statistics *statistics;
   struct timeval start;
   long elapsed;
#endif

   ast_assert(topic != NULL);
   ast_assert(message != NULL);

#ifdef AST_DEVMODE
   ast_mutex_lock(&message_type_statistics_lock);
   if (message_type_id >= AST_VECTOR_SIZE(&message_type_statistics)) {
      struct stasis_message_type_statistics new_statistics = {
         .published = 0,
      };
      if (AST_VECTOR_REPLACE(&message_type_statistics, message_type_id, new_statistics)) {
         ast_mutex_unlock(&message_type_statistics_lock);
         return;
      }
   }
   statistics = AST_VECTOR_GET_ADDR(&message_type_statistics, message_type_id);
   statistics->message_type = stasis_message_type(message);
   ast_mutex_unlock(&message_type_statistics_lock);

   ast_atomic_fetchadd_int(&statistics->published, +1);
#endif

   /* If there are no subscribers don't bother */
   if (!stasis_topic_subscribers(topic)) {
#ifdef AST_DEVMODE
      ast_atomic_fetchadd_int(&statistics->unused, +1);
      ast_atomic_fetchadd_int(&topic->statistics->messages_not_dispatched, +1);
#endif
      return;
   }

   /*
    * The topic may be unref'ed by the subscription invocation.
    * Make sure we hold onto a reference while dispatching.
    */
   ao2_ref(topic, +1);
#ifdef AST_DEVMODE
   start = ast_tvnow();
#endif
   ao2_lock(topic);
   for (i = 0; i < AST_VECTOR_SIZE(&topic->subscribers); ++i) {
      struct stasis_subscription *sub = AST_VECTOR_GET(&topic->subscribers, i);

      ast_assert(sub != NULL);

      dispatched += dispatch_message(sub, message, (sub == sync_sub));
   }
   ao2_unlock(topic);

#ifdef AST_DEVMODE
   elapsed = ast_tvdiff_ms(ast_tvnow(), start);
   if (elapsed > topic->statistics->highest_time_dispatched) {
      topic->statistics->highest_time_dispatched = elapsed;
   }
   if (elapsed < topic->statistics->lowest_time_dispatched) {
      topic->statistics->lowest_time_dispatched = elapsed;
   }
   if (dispatched) {
      ast_atomic_fetchadd_int(&topic->statistics->messages_dispatched, +1);
   } else {
      ast_atomic_fetchadd_int(&statistics->unused, +1);
      ast_atomic_fetchadd_int(&topic->statistics->messages_not_dispatched, +1);
   }
#endif

   ao2_ref(topic, -1);
}

send_subscription_subscribe()

static void send_subscription_subscribe ( struct stasis_topic *  topic, struct stasis_subscription *  sub  )

static

Definition at line 1648 of file stasis.c.

References ao2_cleanup, ast_assert, stasis_message_create(), stasis_publish(), stasis_subscription_change_type(), stasis_subscription_is_subscribed(), subscription_change_alloc(), and stasis_subscription::uniqueid.

Referenced by internal_stasis_subscribe(), and subscription_invoke().

{
   struct stasis_subscription_change *change;
   struct stasis_message *msg;

   /* This assumes that we have already unsubscribed */
   ast_assert(stasis_subscription_is_subscribed(sub));

   if (!stasis_subscription_change_type()) {
      return;
   }

   change = subscription_change_alloc(topic, sub->uniqueid, "Subscribe");
   if (!change) {
      return;
   }

   msg = stasis_message_create(stasis_subscription_change_type(), change);
   if (!msg) {
      ao2_cleanup(change);
      return;
   }

   stasis_publish(topic, msg);
   ao2_cleanup(msg);
   ao2_cleanup(change);
}

send_subscription_unsubscribe()

static void send_subscription_unsubscribe ( struct stasis_topic *  topic, struct stasis_subscription *  sub  )

static

Definition at line 1676 of file stasis.c.

References ao2_cleanup, ast_assert, dispatch_message(), stasis_message_create(), stasis_publish(), stasis_subscription_change_type(), stasis_subscription_is_subscribed(), subscription_change_alloc(), and stasis_subscription::uniqueid.

Referenced by stasis_unsubscribe(), and subscription_invoke().

{
   struct stasis_subscription_change *change;
   struct stasis_message *msg;

   /* This assumes that we have already unsubscribed */
   ast_assert(!stasis_subscription_is_subscribed(sub));

   if (!stasis_subscription_change_type()) {
      return;
   }

   change = subscription_change_alloc(topic, sub->uniqueid, "Unsubscribe");
   if (!change) {
      return;
   }

   msg = stasis_message_create(stasis_subscription_change_type(), change);
   if (!msg) {
      ao2_cleanup(change);
      return;
   }

   stasis_publish(topic, msg);

   /* Now we have to dispatch to the subscription itself */
   dispatch_message(sub, msg, 0);

   ao2_cleanup(msg);
   ao2_cleanup(change);
}

stasis_cleanup()

static void stasis_cleanup ( void  )

static

Cleanup function for graceful shutdowns.

Definition at line 3042 of file stasis.c.

References aco_info_destroy(), ao2_cleanup, ao2_global_obj_release, ARRAY_LEN, ast_cli_unregister_multiple(), ast_multi_user_event_type(), ast_threadpool_shutdown(), AST_VECTOR_FREE, globals, NULL, STASIS_MESSAGE_TYPE_CLEANUP, and stasis_subscription_change_type().

Referenced by stasis_init().

{
#ifdef AST_DEVMODE
   ast_cli_unregister_multiple(cli_stasis_statistics, ARRAY_LEN(cli_stasis_statistics));
   AST_VECTOR_FREE(&message_type_statistics);
   ao2_global_obj_release(subscription_statistics);
   ao2_global_obj_release(topic_statistics);
#endif
   ast_cli_unregister_multiple(cli_stasis, ARRAY_LEN(cli_stasis));
   ao2_cleanup(topic_all);
   topic_all = NULL;
   ast_threadpool_shutdown(threadpool);
   threadpool = NULL;
   STASIS_MESSAGE_TYPE_CLEANUP(stasis_subscription_change_type);
   STASIS_MESSAGE_TYPE_CLEANUP(ast_multi_user_event_type);
   aco_info_destroy(&cfg_info);
   ao2_global_obj_release(globals);
}

stasis_config_alloc()

static void * stasis_config_alloc ( void  )

static

Definition at line 2253 of file stasis.c.

References ao2_alloc, ao2_ref, ast_calloc, ast_str_container_alloc, stasis_declined_config::declined, stasis_config::declined_message_types, NULL, stasis_config_destructor(), stasis_declined_config_destructor(), and stasis_config::threadpool_options.

Referenced by stasis_init().

{
   struct stasis_config *cfg;

   if (!(cfg = ao2_alloc(sizeof(*cfg), stasis_config_destructor))) {
      return NULL;
   }

   cfg->threadpool_options = ast_calloc(1, sizeof(*cfg->threadpool_options));
   if (!cfg->threadpool_options) {
      ao2_ref(cfg, -1);
      return NULL;
   }

   cfg->declined_message_types = ao2_alloc(sizeof(*cfg->declined_message_types),
      stasis_declined_config_destructor);
   if (!cfg->declined_message_types) {
      ao2_ref(cfg, -1);
      return NULL;
   }

   cfg->declined_message_types->declined = ast_str_container_alloc(13);
   if (!cfg->declined_message_types->declined) {
      ao2_ref(cfg, -1);
      return NULL;
   }

   return cfg;
}

stasis_config_destructor()

static void stasis_config_destructor ( void *  obj )

static

Definition at line 2245 of file stasis.c.

References ao2_cleanup, ast_free, stasis_config::declined_message_types, and stasis_config::threadpool_options.

Referenced by stasis_config_alloc().

{
   struct stasis_config *cfg = obj;

   ao2_cleanup(cfg->declined_message_types);
   ast_free(cfg->threadpool_options);
}

stasis_declined_config_destructor()

static void stasis_declined_config_destructor ( void *  obj )

static

Definition at line 2238 of file stasis.c.

References ao2_cleanup, and stasis_declined_config::declined.

Referenced by stasis_config_alloc().

{
   struct stasis_declined_config *declined = obj;

   ao2_cleanup(declined->declined);
}

stasis_forward_all()

struct stasis_forward* stasis_forward_all	(	struct stasis_topic *	from_topic,
		struct stasis_topic *	to_topic
	)

Create a subscription which forwards all messages from one topic to another.

Note that the topic parameter of the invoked callback will the be the topic the message was sent to, not the topic the subscriber subscribed to.

Parameters

from_topic	Topic to forward.
to_topic	Destination topic of forwarded messages.

Returns

New forwarding subscription.

NULL on error.

Since

12

Definition at line 1578 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_bump, ao2_ref, ao2_unlock, AST_VECTOR_APPEND, AST_VECTOR_GET, AST_VECTOR_SIZE, forward_dtor(), stasis_forward::from_topic, NULL, stasis_forward::to_topic, topic_add_subscription(), and topic_lock_both.

Referenced by __init_manager(), ari_bridges_play_new(), ast_ari_bridges_record(), ast_channel_forward_endpoint(), ast_channel_internal_setup_topics(), AST_TEST_DEFINE(), create_subscriptions(), endpoint_internal_create(), forwards_create_bridge(), forwards_create_channel(), forwards_create_endpoint(), load_general_config(), load_module(), manager_bridging_init(), manager_channels_init(), manager_mwi_init(), manager_subscriptions_init(), manager_system_init(), stasis_cp_all_create(), stasis_cp_single_create(), stasis_topic_pool_get_topic(), and state_alloc().

{
   int res;
   size_t idx;
   struct stasis_forward *forward;

   if (!from_topic || !to_topic) {
      return NULL;
   }

   forward = ao2_alloc_options(sizeof(*forward), forward_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!forward) {
      return NULL;
   }

   /* Forwards to ourselves are implicit. */
   if (to_topic == from_topic) {
      return forward;
   }

   forward->from_topic = ao2_bump(from_topic);
   forward->to_topic = ao2_bump(to_topic);

   topic_lock_both(to_topic, from_topic);
   res = AST_VECTOR_APPEND(&to_topic->upstream_topics, from_topic);
   if (res != 0) {
      ao2_unlock(from_topic);
      ao2_unlock(to_topic);
      ao2_ref(forward, -1);
      return NULL;
   }

   for (idx = 0; idx < AST_VECTOR_SIZE(&to_topic->subscribers); ++idx) {
      topic_add_subscription(from_topic, AST_VECTOR_GET(&to_topic->subscribers, idx));
   }
   ao2_unlock(from_topic);
   ao2_unlock(to_topic);

   return forward;
}

stasis_forward_cancel()

struct stasis_forward* stasis_forward_cancel

(

struct stasis_forward *

forward

)

Definition at line 1548 of file stasis.c.

References ao2_cleanup, ao2_unlock, AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_GET, AST_VECTOR_REMOVE_ELEM_UNORDERED, AST_VECTOR_SIZE, stasis_forward::from_topic, NULL, stasis_forward::to_topic, topic_lock_both, and topic_remove_subscription().

Referenced by all_dtor(), ari_bridges_play_new(), ast_ari_bridges_record(), ast_channel_internal_cleanup(), ast_endpoint_shutdown(), AST_TEST_DEFINE(), bridge_channel_control_thread(), cleanup_module(), destroy_subscriptions(), forwards_unsubscribe(), load_general_config(), load_module(), manager_bridging_cleanup(), manager_channels_shutdown(), manager_mwi_shutdown(), manager_shutdown(), manager_system_shutdown(), stasis_cp_single_unsubscribe(), state_dtor(), topic_pool_entry_dtor(), and unload_module().

{
   int idx;
   struct stasis_topic *from;
   struct stasis_topic *to;

   if (!forward) {
      return NULL;
   }

   from = forward->from_topic;
   to = forward->to_topic;

   if (from && to) {
      topic_lock_both(to, from);
      AST_VECTOR_REMOVE_ELEM_UNORDERED(&to->upstream_topics, from,
         AST_VECTOR_ELEM_CLEANUP_NOOP);

      for (idx = 0; idx < AST_VECTOR_SIZE(&to->subscribers); ++idx) {
         topic_remove_subscription(from, AST_VECTOR_GET(&to->subscribers, idx));
      }
      ao2_unlock(from);
      ao2_unlock(to);
   }

   ao2_cleanup(forward);

   return NULL;
}

stasis_init()

int stasis_init

(

void

)

Initialize the Stasis subsystem.

Returns

0 on success.

Non-zero on error.

Since

12

Definition at line 3061 of file stasis.c.

References ACO_EXACT, aco_info_init(), aco_option_register, aco_option_register_custom, aco_process_config(), ACO_PROCESS_ERROR, aco_set_defaults(), AO2_ALLOC_OPT_LOCK_MUTEX, ao2_cleanup, ao2_container_alloc_hash, ao2_global_obj_ref, ao2_global_obj_replace_unref, ao2_ref, ARRAY_LEN, ast_cli_register_multiple, ast_log, ast_multi_user_event_type(), ast_register_cleanup(), ast_threadpool_create(), AST_THREADPOOL_OPTIONS_VERSION, AST_VECTOR_INIT, ast_threadpool_options::auto_increment, declined_handler(), stasis_config::declined_message_types, FLDSET, globals, ast_threadpool_options::idle_timeout, stasis_threadpool_conf::idle_timeout_sec, ast_threadpool_options::initial_size, stasis_threadpool_conf::initial_size, LOG_ERROR, LOG_NOTICE, ast_threadpool_options::max_size, stasis_threadpool_conf::max_size, NULL, OPT_INT_T, PARSE_IN_RANGE, stasis_cache_init(), stasis_cleanup(), stasis_config_alloc(), STASIS_MESSAGE_TYPE_INIT, stasis_subscription_change_type(), SUBSCRIPTION_STATISTICS_BUCKETS, subscription_statistics_cmp(), subscription_statistics_hash(), stasis_config::threadpool_options, TOPIC_ALL_BUCKETS, TOPIC_STATISTICS_BUCKETS, topic_statistics_cmp(), topic_statistics_hash(), and ast_threadpool_options::version.

Referenced by asterisk_daemon().

{
   struct stasis_config *cfg;
   int cache_init;
   struct ast_threadpool_options threadpool_opts = { 0, };
#ifdef AST_DEVMODE
   struct ao2_container *subscription_stats;
   struct ao2_container *topic_stats;
#endif

   /* Be sure the types are cleaned up after the message bus */
   ast_register_cleanup(stasis_cleanup);

   if (aco_info_init(&cfg_info)) {
      return -1;
   }

   aco_option_register_custom(&cfg_info, "decline", ACO_EXACT,
      declined_options, "", declined_handler, 0);
   aco_option_register(&cfg_info, "initial_size", ACO_EXACT,
      threadpool_options, "5", OPT_INT_T, PARSE_IN_RANGE,
      FLDSET(struct stasis_threadpool_conf, initial_size), 0,
      INT_MAX);
   aco_option_register(&cfg_info, "idle_timeout_sec", ACO_EXACT,
      threadpool_options, "20", OPT_INT_T, PARSE_IN_RANGE,
      FLDSET(struct stasis_threadpool_conf, idle_timeout_sec), 0,
      INT_MAX);
   aco_option_register(&cfg_info, "max_size", ACO_EXACT,
      threadpool_options, "50", OPT_INT_T, PARSE_IN_RANGE,
      FLDSET(struct stasis_threadpool_conf, max_size), 0,
      INT_MAX);

   if (aco_process_config(&cfg_info, 0) == ACO_PROCESS_ERROR) {
      struct stasis_config *default_cfg = stasis_config_alloc();

      if (!default_cfg) {
         return -1;
      }

      if (aco_set_defaults(&threadpool_option, "threadpool", default_cfg->threadpool_options)) {
         ast_log(LOG_ERROR, "Failed to initialize defaults on Stasis configuration object\n");
         ao2_ref(default_cfg, -1);

         return -1;
      }

      if (aco_set_defaults(&declined_option, "declined_message_types", default_cfg->declined_message_types)) {
         ast_log(LOG_ERROR, "Failed to load stasis.conf and failed to initialize defaults.\n");
         ao2_ref(default_cfg, -1);

         return -1;
      }

      ast_log(LOG_NOTICE, "Could not load Stasis configuration; using defaults\n");
      ao2_global_obj_replace_unref(globals, default_cfg);
      cfg = default_cfg;
   } else {
      cfg = ao2_global_obj_ref(globals);
      if (!cfg) {
         ast_log(LOG_ERROR, "Failed to obtain Stasis configuration object\n");

         return -1;
      }
   }

   threadpool_opts.version = AST_THREADPOOL_OPTIONS_VERSION;
   threadpool_opts.initial_size = cfg->threadpool_options->initial_size;
   threadpool_opts.auto_increment = 1;
   threadpool_opts.max_size = cfg->threadpool_options->max_size;
   threadpool_opts.idle_timeout = cfg->threadpool_options->idle_timeout_sec;
   threadpool = ast_threadpool_create("stasis", NULL, &threadpool_opts);
   ao2_ref(cfg, -1);
   if (!threadpool) {
      ast_log(LOG_ERROR, "Failed to create 'stasis-core' threadpool\n");

      return -1;
   }

   cache_init = stasis_cache_init();
   if (cache_init != 0) {
      return -1;
   }

   if (STASIS_MESSAGE_TYPE_INIT(stasis_subscription_change_type) != 0) {
      return -1;
   }
   if (STASIS_MESSAGE_TYPE_INIT(ast_multi_user_event_type) != 0) {
      return -1;
   }

   topic_all = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, TOPIC_ALL_BUCKETS,
         topic_proxy_hash_fn, 0, topic_proxy_cmp_fn);
   if (!topic_all) {
      return -1;
   }

   if (ast_cli_register_multiple(cli_stasis, ARRAY_LEN(cli_stasis))) {
      return -1;
   }

#ifdef AST_DEVMODE
   /* Statistics information is stored separately so that we don't alter or interrupt the lifetime of the underlying
    * topic or subscripton.
    */
   subscription_stats = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, SUBSCRIPTION_STATISTICS_BUCKETS,
      subscription_statistics_hash, 0, subscription_statistics_cmp);
   if (!subscription_stats) {
      return -1;
   }
   ao2_global_obj_replace_unref(subscription_statistics, subscription_stats);
   ao2_cleanup(subscription_stats);

   topic_stats = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, TOPIC_STATISTICS_BUCKETS,
      topic_statistics_hash, 0, topic_statistics_cmp);
   if (!topic_stats) {
      return -1;
   }
   ao2_global_obj_replace_unref(topic_statistics, topic_stats);
   ao2_cleanup(topic_stats);
   if (!topic_stats) {
      return -1;
   }

   AST_VECTOR_INIT(&message_type_statistics, 0);

   if (ast_cli_register_multiple(cli_stasis_statistics, ARRAY_LEN(cli_stasis_statistics))) {
      return -1;
   }
#endif

   return 0;
}

stasis_log_bad_type_access()

void stasis_log_bad_type_access

(

const char *

name

)

Definition at line 1940 of file stasis.c.

References ast_log, LOG_ERROR, and stasis_message_type_declined().

{
#ifdef AST_DEVMODE
   if (!stasis_message_type_declined(name)) {
      ast_log(LOG_ERROR, "Use of %s() before init/after destruction\n", name);
   }
#endif
}

stasis_message_type_declined()

int stasis_message_type_declined

(

const char *

name

)

Check whether a message type is declined.

Parameters

name	The name of the message type to check

Return values

zero	The message type is not declined
non-zero	The message type is declined

Definition at line 2283 of file stasis.c.

References ao2_cleanup, ao2_find, ao2_global_obj_ref, ao2_ref, ast_log, stasis_declined_config::declined, stasis_config::declined_message_types, globals, LOG_NOTICE, and OBJ_SEARCH_KEY.

Referenced by stasis_log_bad_type_access(), and stasis_message_type_create().

{
   struct stasis_config *cfg = ao2_global_obj_ref(globals);
   char *name_in_declined;
   int res;

   if (!cfg || !cfg->declined_message_types) {
      ao2_cleanup(cfg);
      return 0;
   }

   name_in_declined = ao2_find(cfg->declined_message_types->declined, name, OBJ_SEARCH_KEY);
   res = name_in_declined ? 1 : 0;
   ao2_cleanup(name_in_declined);
   ao2_ref(cfg, -1);
   if (res) {
      ast_log(LOG_NOTICE, "Declining to allocate Stasis message type '%s' due to configuration\n", name);
   }
   return res;
}

STASIS_MESSAGE_TYPE_DEFN() [1/2]

STASIS_MESSAGE_TYPE_DEFN

(

stasis_subscription_change_type

)

Referenced by declined_handler().

STASIS_MESSAGE_TYPE_DEFN() [2/2]

STASIS_MESSAGE_TYPE_DEFN	(	ast_multi_user_event_type	,
		.	to_json = multi_user_event_to_json,
		.	to_ami = multi_user_event_to_ami
	)

Define multi user event message type(s).

stasis_publish()

void stasis_publish	(	struct stasis_topic *	topic,
		struct stasis_message *	message
	)

Publish a message to a topic's subscribers.

Parameters

topic	Topic.
message	Message to publish.

This call is asynchronous and will return immediately upon queueing the message for delivery to the topic's subscribers.

Since

12

Definition at line 1511 of file stasis.c.

References NULL, and publish_msg().

Referenced by __ast_test_suite_event_notify(), aoc_publish_blob(), app_send_end_msg(), ast_bridge_publish_attended_transfer(), ast_bridge_publish_blind_transfer(), ast_bridge_publish_enter(), ast_bridge_publish_leave(), ast_bridge_publish_merge(), ast_bridge_publish_state(), ast_cel_publish_event(), ast_channel_publish_blob(), ast_channel_publish_cached_blob(), ast_channel_publish_final_snapshot(), ast_channel_publish_snapshot(), ast_device_state_clear_cache(), ast_endpoint_blob_publish(), ast_endpoint_shutdown(), ast_manager_publish_event(), ast_monitor_start(), ast_monitor_stop(), ast_multi_object_blob_single_channel_publish(), ast_publish_device_state_full(), ast_rtp_publish_rtcp_message(), ast_system_publish_registry(), AST_TEST_DEFINE(), bridge_attended_transfer_handler(), bridge_blind_transfer_handler(), bridge_merge_handler(), bridge_publish_state_from_blob(), bridge_topics_destroy(), cache_test_aggregate_publish_fn(), caching_topic_exec(), cc_publish(), clear_node_cache(), device_state_aggregate_publish(), endpoint_publish_snapshot(), endpoint_state_cb(), handle_security_event(), local_optimization_finished_cb(), local_optimization_started_cb(), manager_mute_mixmonitor(), meetme_stasis_generate_msg(), mixmonitor_exec(), moh_post_start(), moh_post_stop(), notify_new_message(), phase_e_handler(), presence_state_event(), publish_acl_change(), publish_chanspy_message(), publish_cluster_discovery_to_stasis_full(), publish_corosync_ping_to_stasis(), publish_format_update(), publish_hint_change(), publish_hint_remove(), publish_load_message_type(), publish_local_bridge_message(), publish_message_for_channel_topics(), publish_parked_call(), publish_parked_call_failure(), queue_publish_member_blob(), queue_publish_multi_channel_snapshot_blob(), remove_device_states_cb(), report_fax_status(), report_receive_fax_status(), report_send_fax_status(), send_call_pickup_stasis_message(), send_conf_stasis(), send_conf_stasis_snapshots(), send_msg(), send_start_msg_snapshots(), send_subscription_subscribe(), send_subscription_unsubscribe(), stasis_app_control_publish(), stasis_app_user_event(), stasis_state_publish(), stasis_state_publish_by_id(), stasis_state_remove_publish_by_id(), stop_mixmonitor_full(), stun_monitor_request(), and talk_detect_audiohook_cb().

{
   publish_msg(topic, message, NULL);
}

stasis_publish_sync()

void stasis_publish_sync	(	struct stasis_subscription *	sub,
		struct stasis_message *	message
	)

Publish a message to a topic's subscribers, synchronizing on the specified subscriber.

Parameters

sub	Subscription to synchronize on.
message	Message to publish.

The caller of stasis_publish_sync will block until the specified subscriber completes handling of the message.

All other subscribers to the topic the stasis_subpscription is subscribed to are also delivered the message; this delivery however happens asynchronously.

Since

12.1.0

Definition at line 1516 of file stasis.c.

References ast_assert, NULL, publish_msg(), and stasis_subscription::topic.

Referenced by AST_TEST_DEFINE(), and stasis_message_router_publish_sync().

{
   ast_assert(sub != NULL);

   publish_msg(sub->topic, message, sub);
}

stasis_show_topic()

static char* stasis_show_topic ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2417 of file stasis.c.

References ao2_lock, ao2_ref, ao2_unlock, ast_cli_args::argc, ast_cli_args::argv, ast_cli(), ast_format_duration_hh_mm_ss(), ast_tvnow(), AST_VECTOR_GET, AST_VECTOR_SIZE, CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, NULL, ast_cli_args::pos, stasis_topic_get(), stasis_subscription::topic, topic_complete_name(), stasis_subscription::uniqueid, ast_cli_entry::usage, and ast_cli_args::word.

{
   struct stasis_topic *topic;
   char print_time[32];
   int i;

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis show topic";
      e->usage =
          "Usage: stasis show topic <name>\n"
          "       Show stasis topic detail info.\n";
      return NULL;
   case CLI_GENERATE:
      if (a->pos == 3) {
         return topic_complete_name(a->word);
      } else {
         return NULL;
      }
   }

   if (a->argc != 4) {
      return CLI_SHOWUSAGE;
   }

   topic = stasis_topic_get(a->argv[3]);
   if (!topic) {
      ast_cli(a->fd, "Specified topic '%s' does not exist\n", a->argv[3]);
      return CLI_FAILURE;
   }

   ast_cli(a->fd, "Name: %s\n", topic->name);
   ast_cli(a->fd, "Detail: %s\n", topic->detail);
   ast_cli(a->fd, "Subscribers count: %zu\n", AST_VECTOR_SIZE(&topic->subscribers));
   ast_cli(a->fd, "Forwarding topic count: %zu\n", AST_VECTOR_SIZE(&topic->upstream_topics));
   ast_format_duration_hh_mm_ss(ast_tvnow().tv_sec - topic->creationtime->tv_sec, print_time, sizeof(print_time));
   ast_cli(a->fd, "Duration time: %s\n", print_time);

   ao2_lock(topic);
   ast_cli(a->fd, "\nSubscribers:\n");
   for (i = 0; i < AST_VECTOR_SIZE(&topic->subscribers); i++) {
      struct stasis_subscription *subscription_tmp = AST_VECTOR_GET(&topic->subscribers, i);
      ast_cli(a->fd, "  UniqueID: %s, Topic: %s, Detail: %s\n",
            subscription_tmp->uniqueid, subscription_tmp->topic->name, subscription_tmp->topic->detail);
   }

   ast_cli(a->fd, "\nForwarded topics:\n");
   for (i = 0; i < AST_VECTOR_SIZE(&topic->upstream_topics); i++) {
      struct stasis_topic *topic_tmp = AST_VECTOR_GET(&topic->upstream_topics, i);
      ast_cli(a->fd, "  Topic: %s, Detail: %s\n", topic_tmp->name, topic_tmp->detail);
   }
   ao2_unlock(topic);

   ao2_ref(topic, -1);

   return CLI_SUCCESS;
}

stasis_show_topics()

static char* stasis_show_topics ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2334 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_list, ao2_container_dup(), ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, AO2_ITERATOR_UNLINK, ao2_ref, ast_cli_args::argc, ast_cli_entry::args, ast_cli(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, topic_proxy::detail, ast_cli_args::fd, FMT_FIELDS, FMT_HEADERS, topic_proxy::name, NULL, OBJ_SEARCH_OBJECT, and ast_cli_entry::usage.

{
   struct ao2_iterator iter;
   struct topic_proxy *topic;
   struct ao2_container *tmp_container;
   int count = 0;
#define FMT_HEADERS     "%-64s %-64s\n"
#define FMT_FIELDS      "%-64s %-64s\n"

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis show topics";
      e->usage =
         "Usage: stasis show topics\n"
         "  Shows a list of topics\n";
      return NULL;
   case CLI_GENERATE:
      return NULL;
   }

   if (a->argc != e->args) {
      return CLI_SHOWUSAGE;
   }

   ast_cli(a->fd, "\n" FMT_HEADERS, "Name", "Detail");

   tmp_container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0,
            topic_proxy_sort_fn, NULL);

   if (!tmp_container || ao2_container_dup(tmp_container, topic_all, OBJ_SEARCH_OBJECT)) {
      ao2_cleanup(tmp_container);

      return NULL;
   }

   /* getting all topic in order */
   iter = ao2_iterator_init(tmp_container, AO2_ITERATOR_UNLINK);
   while ((topic = ao2_iterator_next(&iter))) {
      ast_cli(a->fd, FMT_FIELDS, topic->name, topic->detail);
      ao2_ref(topic, -1);
      ++count;
   }
   ao2_iterator_destroy(&iter);
   ao2_cleanup(tmp_container);

   ast_cli(a->fd, "\n%d Total topics\n\n", count);

#undef FMT_HEADERS
#undef FMT_FIELDS

   return CLI_SUCCESS;
}

stasis_subscription_accept_formatters()

void stasis_subscription_accept_formatters	(	struct stasis_subscription *	subscription,
		enum stasis_subscription_message_formatters	formatters
	)

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

Parameters

subscription	Subscription to alter.
formatters	A bitmap of stasis_subscription_message_formatters we wish to receive.

Since

13.25.0

16.2.0

Definition at line 1095 of file stasis.c.

References ao2_lock, ao2_unlock, ast_assert, NULL, and stasis_subscription::topic.

Referenced by AST_TEST_DEFINE(), stasis_message_router_accept_formatters(), and stasis_message_router_set_formatters_default().

{
   ast_assert(subscription != NULL);

   ao2_lock(subscription->topic);
   subscription->accepted_formatters = formatters;
   ao2_unlock(subscription->topic);

   return;
}

stasis_subscription_accept_message_type()

int stasis_subscription_accept_message_type	(	struct stasis_subscription *	subscription,
		const struct stasis_message_type *	type
	)

Indicate to a subscription that we are interested in a message type.

This will cause the subscription to allow the given message type to be raised to our subscription callback. This enables internal filtering in the stasis message bus to reduce messages.

Parameters

subscription	Subscription to add message type to.
type	The message type we wish to receive.

Return values

0	on success
-1	failure

Since

17.0.0

Note

If you are wanting to use stasis_final_message you will need to accept stasis_subscription_change_type as a message type.

Until the subscription is set to selective filtering it is possible for it to receive messages of message types that would not normally be accepted.

Definition at line 1025 of file stasis.c.

References ao2_lock, ao2_unlock, ast_assert, AST_VECTOR_REPLACE, NULL, stasis_message_type_id(), stasis_message_type_name(), STASIS_SUBSCRIPTION_FILTER_FORCED_NONE, and stasis_subscription::topic.

Referenced by acl_change_stasis_subscribe(), add_peer_mwi_subs(), ast_mwi_subscribe_pool(), ast_res_pjsip_initialize_configuration(), AST_TEST_DEFINE(), asterisk_start_devicestate_publishing(), asterisk_start_mwi_publishing(), cc_generic_agent_start_monitoring(), create_new_generic_list(), create_parked_subscription_full(), devstate_init(), load_module(), load_pbx(), mwi_stasis_subscription_alloc(), network_change_stasis_subscribe(), park_and_announce_app_exec(), parking_manager_enable_stasis(), refer_blind_callback(), rtp_reload(), stasis_caching_accept_message_type(), stasis_message_router_add(), stasis_message_router_add_cache_update(), stasis_message_router_create_internal(), subscribe_device_state(), and xmpp_init_event_distribution().

{
   if (!subscription) {
      return -1;
   }

   ast_assert(type != NULL);
   ast_assert(stasis_message_type_name(type) != NULL);

   if (!type || !stasis_message_type_name(type)) {
      /* Filtering is unreliable as this message type is not yet initialized
       * so force all messages through.
       */
      subscription->filter = STASIS_SUBSCRIPTION_FILTER_FORCED_NONE;
      return 0;
   }

   ao2_lock(subscription->topic);
   if (AST_VECTOR_REPLACE(&subscription->accepted_message_types, stasis_message_type_id(type), 1)) {
      /* We do this for the same reason as above. The subscription can still operate, so allow
       * it to do so by forcing all messages through.
       */
      subscription->filter = STASIS_SUBSCRIPTION_FILTER_FORCED_NONE;
   }
   ao2_unlock(subscription->topic);

   return 0;
}

stasis_subscription_cb_noop()

void stasis_subscription_cb_noop	(	void *	data,
		struct stasis_subscription *	sub,
		struct stasis_message *	message
	)

Stasis subscription callback function that does nothing.

Note

This callback should be used for events are not directly processed, but need to be generated so data can be retrieved from cache later. Subscriptions with this callback can be released with stasis_unsubscribe, even during module unload.

Since

13.5

Definition at line 811 of file stasis.c.

Referenced by build_gateway(), build_peer(), and mkintf().

{
}

stasis_subscription_decline_message_type()

int stasis_subscription_decline_message_type	(	struct stasis_subscription *	subscription,
		const struct stasis_message_type *	type
	)

Indicate to a subscription that we are not interested in a message type.

Parameters

subscription	Subscription to remove message type from.
type	The message type we don't wish to receive.

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 1055 of file stasis.c.

References ao2_lock, ao2_unlock, ast_assert, AST_VECTOR_REPLACE, AST_VECTOR_SIZE, NULL, stasis_message_type_id(), stasis_message_type_name(), and stasis_subscription::topic.

Referenced by AST_TEST_DEFINE().

{
   if (!subscription) {
      return -1;
   }

   ast_assert(type != NULL);
   ast_assert(stasis_message_type_name(type) != NULL);

   if (!type || !stasis_message_type_name(type)) {
      return 0;
   }

   ao2_lock(subscription->topic);
   if (stasis_message_type_id(type) < AST_VECTOR_SIZE(&subscription->accepted_message_types)) {
      /* The memory is already allocated so this can't fail */
      AST_VECTOR_REPLACE(&subscription->accepted_message_types, stasis_message_type_id(type), 0);
   }
   ao2_unlock(subscription->topic);

   return 0;
}

stasis_subscription_final_message()

int stasis_subscription_final_message	(	struct stasis_subscription *	sub,
		struct stasis_message *	msg
	)

Determine whether a message is the final message to be received on a subscription.

Parameters

sub	Subscription on which the message was received.
msg	Message to check.

Returns

zero if the provided message is not the final message.

non-zero if the provided message is the final message.

Since

12

Definition at line 1176 of file stasis.c.

References stasis_subscription_change::description, stasis_message_data(), stasis_message_type(), stasis_subscription_change_type(), stasis_subscription_uniqueid(), and stasis_subscription_change::uniqueid.

Referenced by bridge_subscription_change_handler(), caching_topic_exec(), consumer_exec(), consumer_exec_sync(), consumer_finalize(), default_route(), device_state_cb(), dispatch_message(), endpoint_subscription_change(), generic_agent_devstate_cb(), message_sink_cb(), mwi_event_cb(), mwi_stasis_cb(), park_announce_update_cb(), parker_update_cb(), queue_bridge_cb(), queue_channel_cb(), refer_progress_bridge(), router_dispatch(), statsmaker(), sub_subscription_change_handler(), and subscription_invoke().

{
   struct stasis_subscription_change *change;

   if (stasis_message_type(msg) != stasis_subscription_change_type()) {
      return 0;
   }

   change = stasis_message_data(msg);
   if (strcmp("Unsubscribe", change->description)) {
      return 0;
   }

   if (strcmp(stasis_subscription_uniqueid(sub), change->uniqueid)) {
      return 0;
   }

   return 1;
}

stasis_subscription_is_done()

int stasis_subscription_is_done

(

struct stasis_subscription *

subscription

)

Returns whether subscription has received its final message.

Note that a subscription is considered done even while the stasis_subscription_final_message() is being processed. This allows cleanup routines to check the status of the subscription.

Parameters

subscription

Subscription.

Returns

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

False (zero) if waiting for the end.

Definition at line 1120 of file stasis.c.

References ao2_lock, ao2_unlock, and stasis_subscription::final_message_rxed.

Referenced by router_dtor(), stasis_caching_topic_dtor(), stasis_message_router_is_done(), and subscription_dtor().

{
   if (subscription) {
      int ret;

      ao2_lock(subscription);
      ret = subscription->final_message_rxed;
      ao2_unlock(subscription);

      return ret;
   }

   /* Null subscription is about as done as you can get */
   return 1;
}

stasis_subscription_is_subscribed()

int stasis_subscription_is_subscribed

(

const struct stasis_subscription *

sub

)

Returns whether a subscription is currently subscribed.

Note that there may still be messages queued up to be dispatched to this subscription, but the stasis_subscription_final_message() has been enqueued.

Parameters

sub	Subscription to check

Returns

False (zero) if subscription is not subscribed.

True (non-zero) if still subscribed.

Definition at line 1152 of file stasis.c.

References ao2_lock, ao2_unlock, AST_VECTOR_GET, AST_VECTOR_SIZE, sub, and stasis_subscription::topic.

Referenced by asterisk_publisher_devstate_cb(), asterisk_publisher_mwistate_cb(), router_dtor(), send_subscription_subscribe(), send_subscription_unsubscribe(), stasis_caching_topic_dtor(), stasis_caching_unsubscribe(), subscription_dtor(), xmpp_pubsub_devstate_cb(), and xmpp_pubsub_mwi_cb().

{
   if (sub) {
      size_t i;
      struct stasis_topic *topic = sub->topic;

      ao2_lock(topic);
      for (i = 0; i < AST_VECTOR_SIZE(&topic->subscribers); ++i) {
         if (AST_VECTOR_GET(&topic->subscribers, i) == sub) {
            ao2_unlock(topic);
            return 1;
         }
      }
      ao2_unlock(topic);
   }

   return 0;
}

stasis_subscription_join()

void stasis_subscription_join

(

struct stasis_subscription *

subscription

)

Block until the last message is processed on a subscription.

This function will not return until the subscription's callback for the stasis_subscription_final_message() completes. This allows cleanup routines to run before unblocking the joining thread.

Parameters

subscription

Subscription to block on.

Since

12

Definition at line 1107 of file stasis.c.

References ao2_lock, ao2_object_get_lockaddr(), ao2_unlock, ast_cond_wait, stasis_subscription::final_message_processed, and stasis_subscription::join_cond.

Referenced by stasis_caching_unsubscribe_and_join(), and stasis_unsubscribe_and_join().

{
   if (subscription) {
      ao2_lock(subscription);
      /* Wait until the processed flag has been set */
      while (!subscription->final_message_processed) {
         ast_cond_wait(&subscription->join_cond,
            ao2_object_get_lockaddr(subscription));
      }
      ao2_unlock(subscription);
   }
}

stasis_subscription_set_congestion_limits()

int stasis_subscription_set_congestion_limits	(	struct stasis_subscription *	subscription,
		long	low_water,
		long	high_water
	)

Set the high and low alert water marks of the stasis subscription.

Since

13.10.0

Parameters

subscription	Pointer to a stasis subscription
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 1013 of file stasis.c.

References ast_taskprocessor_alert_set_levels(), and stasis_subscription::mailbox.

Referenced by stasis_message_router_set_congestion_limits().

{
   int res = -1;

   if (subscription) {
      res = ast_taskprocessor_alert_set_levels(subscription->mailbox,
         low_water, high_water);
   }
   return res;
}

stasis_subscription_set_filter()

int stasis_subscription_set_filter	(	struct stasis_subscription *	subscription,
		enum stasis_subscription_message_filter	filter
	)

Set the message type filtering level on a subscription.

This will cause the subscription to filter messages according to the provided filter level. For example if selective is used then only messages matching those provided to stasis_subscription_accept_message_type will be raised to the subscription callback.

Parameters

subscription	Subscription that should receive all messages.
filter	What filter to use

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 1079 of file stasis.c.

References ao2_lock, ao2_unlock, filter(), STASIS_SUBSCRIPTION_FILTER_FORCED_NONE, and stasis_subscription::topic.

Referenced by acl_change_stasis_subscribe(), ast_mwi_subscribe_pool(), ast_res_pjsip_initialize_configuration(), AST_TEST_DEFINE(), asterisk_start_devicestate_publishing(), asterisk_start_mwi_publishing(), cc_generic_agent_start_monitoring(), create_new_generic_list(), create_parked_subscription_full(), devstate_init(), load_module(), load_pbx(), network_change_stasis_subscribe(), park_and_announce_app_exec(), parking_manager_enable_stasis(), refer_blind_callback(), rtp_reload(), stasis_caching_set_filter(), stasis_message_router_add(), stasis_message_router_add_cache_update(), stasis_message_router_set_formatters_default(), subscribe_device_state(), and xmpp_init_event_distribution().

{
   if (!subscription) {
      return -1;
   }

   ao2_lock(subscription->topic);
   if (subscription->filter != STASIS_SUBSCRIPTION_FILTER_FORCED_NONE) {
      subscription->filter = filter;
   }
   ao2_unlock(subscription->topic);

   return 0;
}

stasis_subscription_statistics_create()

static struct stasis_subscription_statistics* stasis_subscription_statistics_create ( struct stasis_subscription *  sub, int  needs_mailbox, int  use_thread_pool, const char *  file, int  lineno, const char *  func  )

static

Definition at line 823 of file stasis.c.

References ao2_alloc, ao2_cleanup, ao2_global_obj_ref, ao2_link, ao2_ref, ast_str_container_alloc, make_ari_stubs::file, stasis_subscription_statistics::file, stasis_subscription_statistics::func, stasis_subscription_statistics::lineno, NULL, RAII_VAR, statistics(), sub, stasis_subscription_statistics::sub, subscription_statistics_destroy(), stasis_subscription_statistics::topics, stasis_subscription_statistics::uniqueid, stasis_subscription::uniqueid, stasis_subscription_statistics::uses_mailbox, and stasis_subscription_statistics::uses_threadpool.

Referenced by internal_stasis_subscribe().

{
   struct stasis_subscription_statistics *statistics;
   RAII_VAR(struct ao2_container *, subscription_stats, ao2_global_obj_ref(subscription_statistics), ao2_cleanup);

   if (!subscription_stats) {
      return NULL;
   }

   statistics = ao2_alloc(sizeof(*statistics) + strlen(sub->uniqueid) + 1, subscription_statistics_destroy);
   if (!statistics) {
      return NULL;
   }

   statistics->topics = ast_str_container_alloc(1);
   if (!statistics->topics) {
      ao2_ref(statistics, -1);
      return NULL;
   }

   statistics->file = file;
   statistics->lineno = lineno;
   statistics->func = func;
   statistics->uses_mailbox = needs_mailbox;
   statistics->uses_threadpool = use_thread_pool;
   strcpy(statistics->uniqueid, sub->uniqueid); /* SAFE */
   statistics->sub = sub;
   ao2_link(subscription_stats, statistics);

   return statistics;
}

stasis_subscription_uniqueid()

const char* stasis_subscription_uniqueid

(

const struct stasis_subscription *

sub

)

Get the unique ID for the subscription.

Parameters

sub	Subscription for which to get the unique ID.

Returns

Unique ID for the subscription.

Since

12

Definition at line 1171 of file stasis.c.

References stasis_subscription::uniqueid.

Referenced by AST_TEST_DEFINE(), stasis_subscription_final_message(), topic_add_subscription(), and topic_remove_subscription().

{
   return sub->uniqueid;
}

stasis_topic_create()

struct stasis_topic* stasis_topic_create

(

const char *

name

)

Create a new topic.

Parameters

name	Name of the new topic.

Returns

New topic instance.

NULL on error.

Since

12

Note

There is no explicit ability to unsubscribe all subscribers from a topic and destroy it. As a result the topic can persist until the last subscriber unsubscribes itself even if there is no publisher.

Topic names should be in the form of <subsystem>:<functionality>[/<object>]

Definition at line 618 of file stasis.c.

References stasis_topic_create_with_detail().

Referenced by __init_manager(), app_create(), app_init(), ast_channel_internal_setup_topics(), ast_parking_stasis_init(), ast_presence_state_engine_init(), ast_rtp_engine_init(), ast_security_stasis_init(), ast_stasis_bridging_init(), ast_stasis_channels_init(), ast_stasis_system_init(), AST_TEST_DEFINE(), ast_test_init(), create_cts(), create_subscriptions(), devstate_init(), load_module(), stasis_caching_topic_create(), stasis_cp_all_create(), stasis_cp_sink_create(), stasis_state_manager_create(), stasis_topic_pool_get_topic(), and state_alloc().

{
   return stasis_topic_create_with_detail(name, "");
}

stasis_topic_create_with_detail()

struct stasis_topic* stasis_topic_create_with_detail	(	const char *	name,
		const char *	detail
	)

Create a new topic with given detail.

Parameters

name	Name of the new topic.
detail	Detail description of the new topic. i.e. "Queue main topic for subscribing every queue event"

Returns

New topic instance.

NULL on error.

Note

There is no explicit ability to unsubscribe all subscribers from a topic and destroy it. As a result the topic can persist until the last subscriber unsubscribes itself even if there is no publisher.

Definition at line 569 of file stasis.c.

References ao2_ref, ao2_t_alloc, ast_debug, AST_VECTOR_INIT, INITIAL_SUBSCRIBERS_MAX, link_topic_proxy(), NULL, stasis_topic_get(), stasis_topic_statistics_create(), and topic_dtor().

Referenced by stasis_topic_create().

{
   struct stasis_topic *topic;
   int res = 0;

   if (!name|| !strlen(name) || !detail) {
      return NULL;
   }
   ast_debug(2, "Creating topic. name: %s, detail: %s\n", name, detail);

   topic = stasis_topic_get(name);
   if (topic) {
      ast_debug(2, "Topic is already exist. name: %s, detail: %s\n",
            name, detail);
      return topic;
   }

   topic = ao2_t_alloc(sizeof(*topic), topic_dtor, name);
   if (!topic) {
      return NULL;
   }

   res |= AST_VECTOR_INIT(&topic->subscribers, INITIAL_SUBSCRIBERS_MAX);
   res |= AST_VECTOR_INIT(&topic->upstream_topics, 0);
   if (res) {
      ao2_ref(topic, -1);
      return NULL;
   }

   /* link to the proxy */
   if (link_topic_proxy(topic, name, detail)) {
      ao2_ref(topic, -1);
      return NULL;
   }

#ifdef AST_DEVMODE
   topic->statistics = stasis_topic_statistics_create(topic);
   if (!topic->statistics) {
      ao2_ref(topic, -1);
      return NULL;
   }
#endif
   ast_debug(1, "Topic '%s': %p created\n", topic->name, topic);

   return topic;
}

stasis_topic_detail()

const char* stasis_topic_detail

(

const struct stasis_topic *

topic

)

Return the detail of a topic.

Parameters

topic

Topic.

Returns

Detail of the topic.

NULL if topic is NULL.

Since

12

Definition at line 636 of file stasis.c.

References NULL.

{
   if (!topic) {
      return NULL;
   }
   return topic->detail;
}

stasis_topic_get()

struct stasis_topic* stasis_topic_get

(

const char *

name

)

Get a topic of the given name.

Parameters

name	Topic's name.

Returns

Name of the topic.

NULL on error or not exist.

Note

This SHOULD NOT be used in normal operation for publishing messages.

Definition at line 623 of file stasis.c.

References ao2_weakproxy_find, and OBJ_SEARCH_KEY.

Referenced by link_topic_proxy(), stasis_show_topic(), and stasis_topic_create_with_detail().

{
   return ao2_weakproxy_find(topic_all, name, OBJ_SEARCH_KEY, "");
}

stasis_topic_name()

const char* stasis_topic_name

(

const struct stasis_topic *

topic

)

Return the name of a topic.

Parameters

topic

Topic.

Returns

Name of the topic.

NULL if topic is NULL.

Definition at line 628 of file stasis.c.

References NULL.

Referenced by bridge_topics_destroy(), caching_topic_exec(), internal_stasis_subscribe(), stasis_caching_topic_create(), stasis_caching_topic_dtor(), stasis_message_router_create_internal(), stasis_state_add_publisher(), stasis_state_add_subscriber(), stasis_state_manager_create(), stasis_state_subscribe_pool(), stasis_topic_pool_create(), stasis_topic_pool_delete_topic(), stasis_topic_pool_get_topic(), state_alloc(), state_id_by_topic(), state_manager_dtor(), topic_add_subscription(), topic_pool_dtor(), topic_pool_entry_cmp(), and topic_remove_subscription().

{
   if (!topic) {
      return NULL;
   }
   return topic->name;
}

stasis_topic_pool_create()

struct stasis_topic_pool* stasis_topic_pool_create

(

struct stasis_topic *

pooled_topic

)

Create a topic pool that routes messages from dynamically generated topics to the given topic.

Parameters

pooled_topic

Topic to which messages will be routed

Returns

the new stasis_topic_pool

NULL on failure

Definition at line 1833 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_MUTEX, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_cleanup, ao2_container_alloc_hash, ao2_container_register(), ao2_ref, ast_alloca, NULL, pool, stasis_topic_pool::pool_container, stasis_topic_pool::pool_topic, stasis_topic_name(), TOPIC_POOL_BUCKETS, topic_pool_dtor(), topic_pool_entry_cmp(), and topic_pool_entry_hash().

Referenced by app_init(), ast_stasis_bridging_init(), and devstate_init().

{
   struct stasis_topic_pool *pool;

   pool = ao2_alloc_options(sizeof(*pool), topic_pool_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!pool) {
      return NULL;
   }

   pool->pool_container = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      TOPIC_POOL_BUCKETS, topic_pool_entry_hash, NULL, topic_pool_entry_cmp);
   if (!pool->pool_container) {
      ao2_cleanup(pool);
      return NULL;
   }

#ifdef AO2_DEBUG
   {
      char *container_name =
         ast_alloca(strlen(stasis_topic_name(pooled_topic)) + strlen("-pool") + 1);
      sprintf(container_name, "%s-pool", stasis_topic_name(pooled_topic));
      ao2_container_register(container_name, pool->pool_container, topic_pool_prnt_obj);
   }
#endif

   ao2_ref(pooled_topic, +1);
   pool->pool_topic = pooled_topic;

   return pool;
}

stasis_topic_pool_delete_topic()

void stasis_topic_pool_delete_topic	(	struct stasis_topic_pool *	pool,
		const char *	topic_name
	)

Delete a topic from the topic pool.

Parameters

pool	Pool from which to delete the topic
topic_name	Name of the topic to delete in the form of <pool_topic_name>/<topic_name> or just <topic_name>

Since

13.24

15.6

16.1

Definition at line 1864 of file stasis.c.

References ao2_find, OBJ_NODATA, OBJ_SEARCH_KEY, OBJ_UNLINK, stasis_topic_pool::pool_container, stasis_topic_pool::pool_topic, and stasis_topic_name().

Referenced by bridge_topics_destroy().

{
   /*
    * The topic_name passed in could be a fully-qualified name like <pool_topic_name>/<topic_name>
    * or just <topic_name>.  If it's fully qualified, we need to skip past <pool_topic_name>
    * name and search only on <topic_name>.
    */
   const char *pool_topic_name = stasis_topic_name(pool->pool_topic);
   int pool_topic_name_len = strlen(pool_topic_name);
   const char *search_topic_name;

   if (strncmp(pool_topic_name, topic_name, pool_topic_name_len) == 0) {
      search_topic_name = topic_name + pool_topic_name_len + 1;
   } else {
      search_topic_name = topic_name;
   }

   ao2_find(pool->pool_container, search_topic_name, OBJ_SEARCH_KEY | OBJ_NODATA | OBJ_UNLINK);
}

stasis_topic_pool_get_topic()

struct stasis_topic* stasis_topic_pool_get_topic	(	struct stasis_topic_pool *	pool,
		const char *	topic_name
	)

Find or create a topic in the pool.

Parameters

pool	Pool for which to get the topic
topic_name	Name of the topic to get

Returns

The already stored or newly allocated topic

NULL if the topic was not found and could not be allocated

Definition at line 1884 of file stasis.c.

References ao2_cleanup, ao2_find, ao2_link_flags, ast_asprintf, ast_free, topic_pool_entry::forward, NULL, OBJ_NOLOCK, OBJ_SEARCH_KEY, stasis_topic_pool::pool_container, stasis_topic_pool::pool_topic, RAII_VAR, SCOPED_AO2LOCK, stasis_forward_all(), stasis_topic_create(), stasis_topic_name(), topic_pool_entry::topic, and topic_pool_entry_alloc().

Referenced by ast_device_state_topic(), ast_queue_topic(), and bridge_topics_init().

{
   RAII_VAR(struct topic_pool_entry *, topic_pool_entry, NULL, ao2_cleanup);
   SCOPED_AO2LOCK(topic_container_lock, pool->pool_container);
   char *new_topic_name;
   int ret;

   topic_pool_entry = ao2_find(pool->pool_container, topic_name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
   if (topic_pool_entry) {
      return topic_pool_entry->topic;
   }

   topic_pool_entry = topic_pool_entry_alloc(topic_name);
   if (!topic_pool_entry) {
      return NULL;
   }

   /* To provide further detail and to ensure that the topic is unique within the scope of the
    * system we prefix it with the pooling topic name, which should itself already be unique.
    */
   ret = ast_asprintf(&new_topic_name, "%s/%s", stasis_topic_name(pool->pool_topic), topic_name);
   if (ret < 0) {
      return NULL;
   }

   topic_pool_entry->topic = stasis_topic_create(new_topic_name);
   ast_free(new_topic_name);
   if (!topic_pool_entry->topic) {
      return NULL;
   }

   topic_pool_entry->forward = stasis_forward_all(topic_pool_entry->topic, pool->pool_topic);
   if (!topic_pool_entry->forward) {
      return NULL;
   }

   if (!ao2_link_flags(pool->pool_container, topic_pool_entry, OBJ_NOLOCK)) {
      return NULL;
   }

   return topic_pool_entry->topic;
}

stasis_topic_pool_topic_exists()

int stasis_topic_pool_topic_exists	(	const struct stasis_topic_pool *	pool,
		const char *	topic_name
	)

Check if a topic exists in a pool.

Parameters

pool	Pool to check
topic_name	Name of the topic to check

Return values

1	exists
0	does not exist

Since

13.23.0

Definition at line 1927 of file stasis.c.

References ao2_find, ao2_ref, OBJ_SEARCH_KEY, and stasis_topic_pool::pool_container.

Referenced by ast_publish_device_state_full().

{
   struct topic_pool_entry *topic_pool_entry;

   topic_pool_entry = ao2_find(pool->pool_container, topic_name, OBJ_SEARCH_KEY);
   if (!topic_pool_entry) {
      return 0;
   }

   ao2_ref(topic_pool_entry, -1);
   return 1;
}

stasis_topic_statistics_create()

static struct stasis_topic_statistics* stasis_topic_statistics_create ( struct stasis_topic *  topic )

static

Definition at line 472 of file stasis.c.

References ao2_alloc, ao2_cleanup, ao2_global_obj_ref, ao2_link, ao2_ref, ast_str_container_alloc, NULL, RAII_VAR, statistics(), and topic_statistics_destroy().

Referenced by stasis_topic_create_with_detail().

{
   struct stasis_topic_statistics *statistics;
   RAII_VAR(struct ao2_container *, topic_stats, ao2_global_obj_ref(topic_statistics), ao2_cleanup);

   if (!topic_stats) {
      return NULL;
   }

   statistics = ao2_alloc(sizeof(*statistics) + strlen(topic->name) + 1, topic_statistics_destroy);
   if (!statistics) {
      return NULL;
   }

   statistics->subscribers = ast_str_container_alloc(1);
   if (!statistics->subscribers) {
      ao2_ref(statistics, -1);
      return NULL;
   }

   /* This is strictly used for the pointer address when showing the topic */
   statistics->topic = topic;
   strcpy(statistics->name, topic->name); /* SAFE */
   ao2_link(topic_stats, statistics);

   return statistics;
}

stasis_topic_subscribers()

size_t stasis_topic_subscribers

(

const struct stasis_topic *

topic

)

Return the number of subscribers of a topic.

Parameters

topic

Topic.

Returns

Number of subscribers of the topic.

Since

17.0.0

Definition at line 644 of file stasis.c.

References AST_VECTOR_SIZE.

Referenced by caching_topic_exec(), and publish_msg().

{
   return AST_VECTOR_SIZE(&topic->subscribers);
}

stasis_unsubscribe()

struct stasis_subscription* stasis_unsubscribe

(

struct stasis_subscription *

subscription

)

Cancel a subscription.

Note that in an asynchronous system, there may still be messages queued or in transit to the subscription's callback. These will still be delivered. There will be a final 'SubscriptionCancelled' message, indicating the delivery of the final message.

Parameters

subscription

Subscription to cancel.

Returns

NULL for convenience

Since

12

Definition at line 973 of file stasis.c.

References ao2_bump, ao2_cleanup, ast_log, ast_taskprocessor_push(), LOG_ERROR, stasis_subscription::mailbox, NULL, send_subscription_unsubscribe(), sub_cleanup(), stasis_subscription::topic, and topic_remove_subscription().

Referenced by AST_TEST_DEFINE(), cc_generic_agent_destructor(), destroy_cts(), generic_agent_devstate_cb(), generic_monitor_instance_list_destructor(), mwi_startup_event_cb(), park_and_announce_app_exec(), parked_subscription_datastore_destroy(), refer_progress_bridge(), refer_progress_destroy(), refer_progress_framehook_destroy(), startup_event_cb(), stasis_caching_unsubscribe(), stasis_message_router_unsubscribe(), stasis_state_unsubscribe(), stasis_unsubscribe_and_join(), subscription_persistence_event_cb(), unload_module(), and xmpp_init_event_distribution().

{
   /* The subscription may be the last ref to this topic. Hold
    * the topic ref open until after the unlock. */
   struct stasis_topic *topic;

   if (!sub) {
      return NULL;
   }

   topic = ao2_bump(sub->topic);

   /* We have to remove the subscription first, to ensure the unsubscribe
    * is the final message */
   if (topic_remove_subscription(sub->topic, sub) != 0) {
      ast_log(LOG_ERROR,
         "Internal error: subscription has invalid topic\n");
      ao2_cleanup(topic);

      return NULL;
   }

   /* Now let everyone know about the unsubscribe */
   send_subscription_unsubscribe(topic, sub);

   /* When all that's done, remove the ref the mailbox has on the sub */
   if (sub->mailbox) {
      if (ast_taskprocessor_push(sub->mailbox, sub_cleanup, sub)) {
         /* Nothing we can do here, the conditional is just to keep
          * the compiler happy that we're not ignoring the result. */
      }
   }

   /* Unsubscribing unrefs the subscription */
   ao2_cleanup(sub);
   ao2_cleanup(topic);

   return NULL;
}

stasis_unsubscribe_and_join()

struct stasis_subscription* stasis_unsubscribe_and_join

(

struct stasis_subscription *

subscription

)

Cancel a subscription, blocking until the last message is processed.

While normally it's recommended to stasis_unsubscribe() and wait for stasis_subscription_final_message(), there are times (like during a module unload) where you have to wait for the final message (otherwise you'll call a function in a shared module that no longer exists).

Parameters

subscription

Subscription to cancel.

Returns

NULL for convenience

Since

12

Definition at line 1136 of file stasis.c.

References ao2_cleanup, ao2_ref, NULL, stasis_subscription_join(), and stasis_unsubscribe().

Referenced by acl_change_event_stasis_unsubscribe(), acl_change_stasis_unsubscribe(), ast_res_pjsip_destroy_configuration(), AST_TEST_DEFINE(), ast_xmpp_client_disconnect(), asterisk_stop_devicestate_publishing(), asterisk_stop_mwi_publishing(), devstate_cleanup(), network_change_stasis_unsubscribe(), parking_manager_disable_stasis(), presence_change_common(), remove_device_state_subscription(), rtp_reload(), stasis_message_router_unsubscribe_and_join(), stasis_state_unsubscribe_and_join(), unload_module(), and unload_pbx().

{
   if (!subscription) {
      return NULL;
   }

   /* Bump refcount to hold it past the unsubscribe */
   ao2_ref(subscription, +1);
   stasis_unsubscribe(subscription);
   stasis_subscription_join(subscription);
   /* Now decrement the refcount back */
   ao2_cleanup(subscription);
   return NULL;
}

statistics_show_messages()

static char* statistics_show_messages ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2865 of file stasis.c.

References ast_cli_args::argc, ast_cli_entry::args, ast_cli(), ast_mutex_lock, ast_mutex_unlock, AST_VECTOR_GET_ADDR, AST_VECTOR_SIZE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, FMT_FIELDS, FMT_HEADERS, stasis_message_type_statistics::message_type, message_type_statistics_lock, NULL, stasis_message_type_statistics::published, stasis_message_type_name(), statistics(), stasis_message_type_statistics::unused, and ast_cli_entry::usage.

{
   int i;
   int count = 0;
   int published = 0;
   int unused = 0;
#define FMT_HEADERS     "%-64s %10s %10s\n"
#define FMT_FIELDS      "%-64s %10d %10d\n"

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis statistics show messages";
      e->usage =
         "Usage: stasis statistics show messages\n"
         "  Shows a list of message types and their general statistics\n";
      return NULL;
   case CLI_GENERATE:
      return NULL;
   }

   if (a->argc != e->args) {
      return CLI_SHOWUSAGE;
   }

   ast_cli(a->fd, "\n" FMT_HEADERS, "Message Type", "Published", "Unused");

   ast_mutex_lock(&message_type_statistics_lock);
   for (i = 0; i < AST_VECTOR_SIZE(&message_type_statistics); ++i) {
      struct stasis_message_type_statistics *statistics = AST_VECTOR_GET_ADDR(&message_type_statistics, i);

      if (!statistics->message_type) {
         continue;
      }

      ast_cli(a->fd, FMT_FIELDS, stasis_message_type_name(statistics->message_type), statistics->published,
         statistics->unused);
      published += statistics->published;
      unused += statistics->unused;
      ++count;
   }
   ast_mutex_unlock(&message_type_statistics_lock);

   ast_cli(a->fd, FMT_FIELDS, "Total", published, unused);
   ast_cli(a->fd, "\n%d seen message types\n\n", count);

#undef FMT_HEADERS
#undef FMT_FIELDS

   return CLI_SUCCESS;
}

statistics_show_subscription()

static char* statistics_show_subscription ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2604 of file stasis.c.

References ao2_container_count(), ao2_find, ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_lock, ao2_ref, AO2_STRING_FIELD_SORT_FN(), ao2_unlock, ast_cli_args::argc, ast_cli_args::argv, ast_cli(), CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, stasis_subscription_statistics::file, stasis_subscription_statistics::func, stasis_subscription_statistics::highest_time_invoked, stasis_subscription_statistics::highest_time_message_type, stasis_subscription_statistics::lineno, stasis_subscription_statistics::lowest_time_invoked, stasis_subscription_statistics::messages_dropped, stasis_subscription_statistics::messages_passed, ast_cli_args::n, name, NULL, OBJ_SEARCH_KEY, ast_cli_args::pos, S_OR, stasis_message_type_name(), statistics(), stasis_subscription_statistics::sub, subscription_statistics_complete_name(), stasis_subscription_statistics::topics, stasis_subscription_statistics::uniqueid, ast_cli_entry::usage, stasis_subscription_statistics::uses_mailbox, stasis_subscription_statistics::uses_threadpool, and ast_cli_args::word.

{
   struct stasis_subscription_statistics *statistics;
   struct ao2_container *subscription_stats;
   struct ao2_iterator i;
   char *name;

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis statistics show subscription";
      e->usage =
          "Usage: stasis statistics show subscription <uniqueid>\n"
          "       Show stasis subscription statistics.\n";
      return NULL;
   case CLI_GENERATE:
      if (a->pos == 4) {
         return subscription_statistics_complete_name(a->word, a->n);
      } else {
         return NULL;
      }
   }

   if (a->argc != 5) {
      return CLI_SHOWUSAGE;
   }

   subscription_stats = ao2_global_obj_ref(subscription_statistics);
   if (!subscription_stats) {
      ast_cli(a->fd, "Could not fetch subcription_statistics container\n");
      return CLI_FAILURE;
   }

   statistics = ao2_find(subscription_stats, a->argv[4], OBJ_SEARCH_KEY);
   if (!statistics) {
      ao2_ref(subscription_stats, -1);
      ast_cli(a->fd, "Specified subscription '%s' does not exist\n", a->argv[4]);
      return CLI_FAILURE;
   }

   ao2_ref(subscription_stats, -1);

   ast_cli(a->fd, "Subscription: %s\n", statistics->uniqueid);
   ast_cli(a->fd, "Pointer Address: %p\n", statistics->sub);
   ast_cli(a->fd, "Source filename: %s\n", S_OR(statistics->file, "<unavailable>"));
   ast_cli(a->fd, "Source line number: %d\n", statistics->lineno);
   ast_cli(a->fd, "Source function: %s\n", S_OR(statistics->func, "<unavailable>"));
   ast_cli(a->fd, "Number of messages dropped due to filtering: %d\n", statistics->messages_dropped);
   ast_cli(a->fd, "Number of messages passed to subscriber callback: %d\n", statistics->messages_passed);
   ast_cli(a->fd, "Using mailbox to queue messages: %s\n", statistics->uses_mailbox ? "Yes" : "No");
   ast_cli(a->fd, "Using stasis threadpool for handling messages: %s\n", statistics->uses_threadpool ? "Yes" : "No");
   ast_cli(a->fd, "Lowest amount of time (in milliseconds) spent invoking message: %ld\n", statistics->lowest_time_invoked);
   ast_cli(a->fd, "Highest amount of time (in milliseconds) spent invoking message: %ld\n", statistics->highest_time_invoked);

   ao2_lock(statistics);
   if (statistics->highest_time_message_type) {
      ast_cli(a->fd, "Offender message type for highest invoking time: %s\n", stasis_message_type_name(statistics->highest_time_message_type));
   }
   ao2_unlock(statistics);

   ast_cli(a->fd, "Number of topics: %d\n", ao2_container_count(statistics->topics));

   ast_cli(a->fd, "Subscribed topics:\n");
   i = ao2_iterator_init(statistics->topics, 0);
   while ((name = ao2_iterator_next(&i))) {
      ast_cli(a->fd, "\t%s\n", name);
      ao2_ref(name, -1);
   }
   ao2_iterator_destroy(&i);

   ao2_ref(statistics, -1);

   return CLI_SUCCESS;
}

statistics_show_subscriptions()

static char* statistics_show_subscriptions ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2490 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_container_alloc_rbtree, ao2_container_dup(), ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_cli_args::argc, ast_cli_entry::args, ast_cli(), CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, FMT_FIELDS, FMT_FIELDS2, FMT_HEADERS, stasis_subscription_statistics::highest_time_invoked, stasis_subscription_statistics::lowest_time_invoked, stasis_subscription_statistics::messages_dropped, stasis_subscription_statistics::messages_passed, NULL, statistics(), stasis_subscription_statistics::uniqueid, and ast_cli_entry::usage.

{
   struct ao2_container *sorted_subscriptions;
   struct ao2_container *subscription_stats;
   struct ao2_iterator iter;
   struct stasis_subscription_statistics *statistics;
   int count = 0;
   int dropped = 0;
   int passed = 0;
#define FMT_HEADERS     "%-64s %10s %10s %16s %16s\n"
#define FMT_FIELDS      "%-64s %10d %10d %16ld %16ld\n"
#define FMT_FIELDS2     "%-64s %10d %10d\n"

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis statistics show subscriptions";
      e->usage =
         "Usage: stasis statistics show subscriptions\n"
         "  Shows a list of subscriptions and their general statistics\n";
      return NULL;
   case CLI_GENERATE:
      return NULL;
   }

   if (a->argc != e->args) {
      return CLI_SHOWUSAGE;
   }

   subscription_stats = ao2_global_obj_ref(subscription_statistics);
   if (!subscription_stats) {
      ast_cli(a->fd, "Could not fetch subscription_statistics container\n");
      return CLI_FAILURE;
   }

   sorted_subscriptions = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0,
      stasis_subscription_statistics_sort_fn, NULL);
   if (!sorted_subscriptions) {
      ao2_ref(subscription_stats, -1);
      ast_cli(a->fd, "Could not create container for sorting subscription statistics\n");
      return CLI_SUCCESS;
   }

   if (ao2_container_dup(sorted_subscriptions, subscription_stats, 0)) {
      ao2_ref(sorted_subscriptions, -1);
      ao2_ref(subscription_stats, -1);
      ast_cli(a->fd, "Could not sort subscription statistics\n");
      return CLI_SUCCESS;
   }

   ao2_ref(subscription_stats, -1);

   ast_cli(a->fd, "\n" FMT_HEADERS, "Subscription", "Dropped", "Passed", "Lowest Invoke", "Highest Invoke");

   iter = ao2_iterator_init(sorted_subscriptions, 0);
   while ((statistics = ao2_iterator_next(&iter))) {
      ast_cli(a->fd, FMT_FIELDS, statistics->uniqueid, statistics->messages_dropped, statistics->messages_passed,
         statistics->lowest_time_invoked, statistics->highest_time_invoked);
      dropped += statistics->messages_dropped;
      passed += statistics->messages_passed;
      ao2_ref(statistics, -1);
      ++count;
   }
   ao2_iterator_destroy(&iter);

   ao2_ref(sorted_subscriptions, -1);

   ast_cli(a->fd, FMT_FIELDS2, "Total", dropped, passed);
   ast_cli(a->fd, "\n%d subscriptions\n\n", count);

#undef FMT_HEADERS
#undef FMT_FIELDS
#undef FMT_FIELDS2

   return CLI_SUCCESS;
}

statistics_show_topic()

static char* statistics_show_topic ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2799 of file stasis.c.

References ao2_container_count(), ao2_find, ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_cli_args::argc, ast_cli_args::argv, ast_cli(), CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, ast_cli_args::n, NULL, OBJ_SEARCH_KEY, ast_cli_args::pos, statistics(), topic_statistics_complete_name(), ast_cli_entry::usage, and ast_cli_args::word.

{
   struct stasis_topic_statistics *statistics;
   struct ao2_container *topic_stats;
   struct ao2_iterator i;
   char *uniqueid;

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis statistics show topic";
      e->usage =
          "Usage: stasis statistics show topic <name>\n"
          "       Show stasis topic statistics.\n";
      return NULL;
   case CLI_GENERATE:
      if (a->pos == 4) {
         return topic_statistics_complete_name(a->word, a->n);
      } else {
         return NULL;
      }
   }

   if (a->argc != 5) {
      return CLI_SHOWUSAGE;
   }

   topic_stats = ao2_global_obj_ref(topic_statistics);
   if (!topic_stats) {
      ast_cli(a->fd, "Could not fetch topic_statistics container\n");
      return CLI_FAILURE;
   }

   statistics = ao2_find(topic_stats, a->argv[4], OBJ_SEARCH_KEY);
   if (!statistics) {
      ao2_ref(topic_stats, -1);
      ast_cli(a->fd, "Specified topic '%s' does not exist\n", a->argv[4]);
      return CLI_FAILURE;
   }

   ao2_ref(topic_stats, -1);

   ast_cli(a->fd, "Topic: %s\n", statistics->name);
   ast_cli(a->fd, "Pointer Address: %p\n", statistics->topic);
   ast_cli(a->fd, "Number of messages published that went to no subscriber: %d\n", statistics->messages_not_dispatched);
   ast_cli(a->fd, "Number of messages that went to at least one subscriber: %d\n", statistics->messages_dispatched);
   ast_cli(a->fd, "Lowest amount of time (in milliseconds) spent dispatching message: %ld\n", statistics->lowest_time_dispatched);
   ast_cli(a->fd, "Highest amount of time (in milliseconds) spent dispatching messages: %ld\n", statistics->highest_time_dispatched);
   ast_cli(a->fd, "Number of subscribers: %d\n", ao2_container_count(statistics->subscribers));

   ast_cli(a->fd, "Subscribers:\n");
   i = ao2_iterator_init(statistics->subscribers, 0);
   while ((uniqueid = ao2_iterator_next(&i))) {
      ast_cli(a->fd, "\t%s\n", uniqueid);
      ao2_ref(uniqueid, -1);
   }
   ao2_iterator_destroy(&i);

   ao2_ref(statistics, -1);

   return CLI_SUCCESS;
}

statistics_show_topics()

static char* statistics_show_topics ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 2684 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_container_alloc_rbtree, ao2_container_count(), ao2_container_dup(), ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_cli_args::argc, ast_cli_entry::args, ast_cli(), CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, FMT_FIELDS, FMT_FIELDS2, FMT_HEADERS, NULL, statistics(), and ast_cli_entry::usage.

{
   struct ao2_container *sorted_topics;
   struct ao2_container *topic_stats;
   struct ao2_iterator iter;
   struct stasis_topic_statistics *statistics;
   int count = 0;
   int not_dispatched = 0;
   int dispatched = 0;
#define FMT_HEADERS     "%-64s %10s %10s %10s %16s %16s\n"
#define FMT_FIELDS      "%-64s %10d %10d %10d %16ld %16ld\n"
#define FMT_FIELDS2     "%-64s %10s %10d %10d\n"

   switch (cmd) {
   case CLI_INIT:
      e->command = "stasis statistics show topics";
      e->usage =
         "Usage: stasis statistics show topics\n"
         "  Shows a list of topics and their general statistics\n";
      return NULL;
   case CLI_GENERATE:
      return NULL;
   }

   if (a->argc != e->args) {
      return CLI_SHOWUSAGE;
   }

   topic_stats = ao2_global_obj_ref(topic_statistics);
   if (!topic_stats) {
      ast_cli(a->fd, "Could not fetch topic_statistics container\n");
      return CLI_FAILURE;
   }

   sorted_topics = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0,
      stasis_topic_statistics_sort_fn, NULL);
   if (!sorted_topics) {
      ao2_ref(topic_stats, -1);
      ast_cli(a->fd, "Could not create container for sorting topic statistics\n");
      return CLI_SUCCESS;
   }

   if (ao2_container_dup(sorted_topics, topic_stats, 0)) {
      ao2_ref(sorted_topics, -1);
      ao2_ref(topic_stats, -1);
      ast_cli(a->fd, "Could not sort topic statistics\n");
      return CLI_SUCCESS;
   }

   ao2_ref(topic_stats, -1);

   ast_cli(a->fd, "\n" FMT_HEADERS, "Topic", "Subscribers", "Dropped", "Dispatched", "Lowest Dispatch", "Highest Dispatch");

   iter = ao2_iterator_init(sorted_topics, 0);
   while ((statistics = ao2_iterator_next(&iter))) {
      ast_cli(a->fd, FMT_FIELDS, statistics->name, ao2_container_count(statistics->subscribers),
         statistics->messages_not_dispatched, statistics->messages_dispatched,
         statistics->lowest_time_dispatched, statistics->highest_time_dispatched);
      not_dispatched += statistics->messages_not_dispatched;
      dispatched += statistics->messages_dispatched;
      ao2_ref(statistics, -1);
      ++count;
   }
   ao2_iterator_destroy(&iter);

   ao2_ref(sorted_topics, -1);

   ast_cli(a->fd, FMT_FIELDS2, "Total", "", not_dispatched, dispatched);
   ast_cli(a->fd, "\n%d topics\n\n", count);

#undef FMT_HEADERS
#undef FMT_FIELDS
#undef FMT_FIELDS2

   return CLI_SUCCESS;
}

sub_cleanup()

static int sub_cleanup ( void *  data )

static

Definition at line 966 of file stasis.c.

References ao2_cleanup, and stasis_subscription::data.

Referenced by stasis_unsubscribe().

{
   struct stasis_subscription *sub = data;
   ao2_cleanup(sub);
   return 0;
}

subscription_change_alloc()

static struct stasis_subscription_change* subscription_change_alloc ( struct stasis_topic *  topic, const char *  uniqueid, const char *  description  )

static

Definition at line 1627 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_ref, ast_copy_string(), stasis_subscription_change::description, NULL, subscription_change_dtor(), stasis_subscription_change::topic, and stasis_subscription_change::uniqueid.

Referenced by send_subscription_subscribe(), and send_subscription_unsubscribe().

{
   size_t description_len = strlen(description) + 1;
   size_t uniqueid_len = strlen(uniqueid) + 1;
   struct stasis_subscription_change *change;

   change = ao2_alloc_options(sizeof(*change) + description_len + uniqueid_len,
      subscription_change_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!change) {
      return NULL;
   }

   strcpy(change->description, description); /* SAFE */
   change->uniqueid = change->description + description_len;
   ast_copy_string(change->uniqueid, uniqueid, uniqueid_len); /* SAFE */
   ao2_ref(topic, +1);
   change->topic = topic;

   return change;
}

subscription_change_dtor()

static void subscription_change_dtor ( void *  obj )

static

Definition at line 1620 of file stasis.c.

References ao2_cleanup, and stasis_subscription_change::topic.

Referenced by subscription_change_alloc().

{
   struct stasis_subscription_change *change = obj;

   ao2_cleanup(change->topic);
}

subscription_dtor()

static void subscription_dtor ( void *  obj )

static

Definition at line 715 of file stasis.c.

References ao2_cleanup, ao2_global_obj_ref, ao2_ref, ao2_unlink, ast_assert, ast_cond_destroy, ast_free, ast_taskprocessor_unreference(), AST_VECTOR_FREE, stasis_subscription::join_cond, stasis_subscription::mailbox, NULL, stasis_subscription_is_done(), stasis_subscription_is_subscribed(), stasis_subscription::topic, and stasis_subscription::uniqueid.

Referenced by internal_stasis_subscribe().

{
   struct stasis_subscription *sub = obj;
#ifdef AST_DEVMODE
   struct ao2_container *subscription_stats;
#endif

   /* Subscriptions need to be manually unsubscribed before destruction
    * b/c there's a cyclic reference between topics and subscriptions */
   ast_assert(!stasis_subscription_is_subscribed(sub));
   /* If there are any messages in flight to this subscription; that would
    * be bad. */
   ast_assert(stasis_subscription_is_done(sub));

   ast_free(sub->uniqueid);
   ao2_cleanup(sub->topic);
   sub->topic = NULL;
   ast_taskprocessor_unreference(sub->mailbox);
   sub->mailbox = NULL;
   ast_cond_destroy(&sub->join_cond);

   AST_VECTOR_FREE(&sub->accepted_message_types);

#ifdef AST_DEVMODE
   if (sub->statistics) {
      subscription_stats = ao2_global_obj_ref(subscription_statistics);
      if (subscription_stats) {
         ao2_unlink(subscription_stats, sub->statistics);
         ao2_ref(subscription_stats, -1);
      }
      ao2_ref(sub->statistics, -1);
   }
#endif
}

subscription_invoke()

static void subscription_invoke ( struct stasis_subscription *  sub, struct stasis_message *  message  )

static

Invoke the subscription's callback.

Parameters

sub	Subscription to invoke.
topic	Topic message was published to.
message	Message to send.

Definition at line 756 of file stasis.c.

References ao2_lock, ao2_unlock, ast_cond_signal, ast_tvdiff_ms(), ast_tvnow(), AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_subscription::callback, stasis_subscription::data, stasis_subscription::final_message_processed, stasis_subscription::final_message_rxed, stasis_subscription::join_cond, message_type_id, send_subscription_subscribe(), send_subscription_unsubscribe(), stasis_message_type(), stasis_message_type_id(), stasis_subscription_change_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_final_message(), and statistics().

Referenced by dispatch_exec_async(), dispatch_exec_sync(), and dispatch_message().

{
   unsigned int final = stasis_subscription_final_message(sub, message);
   int message_type_id = stasis_message_type_id(stasis_subscription_change_type());
#ifdef AST_DEVMODE
   struct timeval start;
   long elapsed;

   start = ast_tvnow();
#endif

   /* Notify that the final message has been received */
   if (final) {
      ao2_lock(sub);
      sub->final_message_rxed = 1;
      ast_cond_signal(&sub->join_cond);
      ao2_unlock(sub);
   }

   /*
    * If filtering is turned on and this is a 'final' message, we only invoke the callback
    * if the subscriber accepts subscription_change message types.
    */
   if (!final || sub->filter != STASIS_SUBSCRIPTION_FILTER_SELECTIVE ||
      (message_type_id < AST_VECTOR_SIZE(&sub->accepted_message_types) && AST_VECTOR_GET(&sub->accepted_message_types, message_type_id))) {
      /* Since sub is mostly immutable, no need to lock sub */
      sub->callback(sub->data, sub, message);
   }

   /* Notify that the final message has been processed */
   if (final) {
      ao2_lock(sub);
      sub->final_message_processed = 1;
      ast_cond_signal(&sub->join_cond);
      ao2_unlock(sub);
   }

#ifdef AST_DEVMODE
   elapsed = ast_tvdiff_ms(ast_tvnow(), start);
   if (elapsed > sub->statistics->highest_time_invoked) {
      sub->statistics->highest_time_invoked = elapsed;
      ao2_lock(sub->statistics);
      sub->statistics->highest_time_message_type = stasis_message_type(message);
      ao2_unlock(sub->statistics);
   }
   if (elapsed < sub->statistics->lowest_time_invoked) {
      sub->statistics->lowest_time_invoked = elapsed;
   }
#endif
}

subscription_statistics_cmp()

static int subscription_statistics_cmp ( void *  obj, void *  arg, int  flags  )

static

Definition at line 2945 of file stasis.c.

References ast_assert, CMP_MATCH, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, OBJ_SEARCH_PARTIAL_KEY, and stasis_subscription_statistics::uniqueid.

Referenced by stasis_init().

{
   const struct stasis_subscription_statistics *object_left = obj;
   const struct stasis_subscription_statistics *object_right = arg;
   const char *right_key = arg;
   int cmp;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_OBJECT:
      right_key = object_right->uniqueid;
      /* Fall through */
   case OBJ_SEARCH_KEY:
      cmp = strcasecmp(object_left->uniqueid, right_key);
      break;
   case OBJ_SEARCH_PARTIAL_KEY:
      /* Not supported by container */
      ast_assert(0);
      cmp = -1;
      break;
   default:
      /*
       * What arg points to is specific to this traversal callback
       * and has no special meaning to astobj2.
       */
      cmp = 0;
      break;
   }
   if (cmp) {
      return 0;
   }
   /*
    * At this point the traversal callback is identical to a sorted
    * container.
    */
   return CMP_MATCH;
}

subscription_statistics_complete_name()

static char* subscription_statistics_complete_name ( const char *  word, int  state  )

static

Definition at line 2570 of file stasis.c.

References ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_strdup, NULL, result, statistics(), and stasis_subscription_statistics::uniqueid.

Referenced by statistics_show_subscription().

{
   struct stasis_subscription_statistics *statistics;
   struct ao2_container *subscription_stats;
   struct ao2_iterator it_statistics;
   int wordlen = strlen(word);
   int which = 0;
   char *result = NULL;

   subscription_stats = ao2_global_obj_ref(subscription_statistics);
   if (!subscription_stats) {
      return result;
   }

   it_statistics = ao2_iterator_init(subscription_stats, 0);
   while ((statistics = ao2_iterator_next(&it_statistics))) {
      if (!strncasecmp(word, statistics->uniqueid, wordlen)
         && ++which > state) {
         result = ast_strdup(statistics->uniqueid);
      }
      ao2_ref(statistics, -1);
      if (result) {
         break;
      }
   }
   ao2_iterator_destroy(&it_statistics);
   ao2_ref(subscription_stats, -1);
   return result;
}

subscription_statistics_destroy()

static void subscription_statistics_destroy ( void *  obj )

static

Definition at line 816 of file stasis.c.

References ao2_cleanup, statistics(), and stasis_subscription_statistics::topics.

Referenced by stasis_subscription_statistics_create().

{
   struct stasis_subscription_statistics *statistics = obj;

   ao2_cleanup(statistics->topics);
}

subscription_statistics_hash()

static int subscription_statistics_hash ( const void *  obj, const int  flags  )

static

Definition at line 2924 of file stasis.c.

References ast_assert, ast_str_case_hash(), OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, and OBJ_SEARCH_OBJECT.

Referenced by stasis_init().

{
   const struct stasis_subscription_statistics *object;
   const char *key;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_KEY:
      key = obj;
      break;
   case OBJ_SEARCH_OBJECT:
      object = obj;
      key = object->uniqueid;
      break;
   default:
      /* Hash can only work on something with a full key. */
      ast_assert(0);
      return 0;
   }
   return ast_str_case_hash(key);
}

topic_add_subscription()

static int topic_add_subscription ( struct stasis_topic *  topic, struct stasis_subscription *  sub  )

static

Add a subscriber to a topic.

Parameters

topic	Topic
sub	Subscriber

Returns

0 on success

Non-zero on error

Definition at line 1203 of file stasis.c.

References ao2_lock, ao2_unlock, ast_str_container_add(), AST_VECTOR_APPEND, AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_subscription_uniqueid(), and stasis_topic_name().

Referenced by internal_stasis_subscribe(), proxy_dtor(), and stasis_forward_all().

{
   size_t idx;

   ao2_lock(topic);
   /* The reference from the topic to the subscription is shared with
    * the owner of the subscription, which will explicitly unsubscribe
    * to release it.
    *
    * If we bumped the refcount here, the owner would have to unsubscribe
    * and cleanup, which is a bit awkward. */
   AST_VECTOR_APPEND(&topic->subscribers, sub);

   for (idx = 0; idx < AST_VECTOR_SIZE(&topic->upstream_topics); ++idx) {
      topic_add_subscription(
         AST_VECTOR_GET(&topic->upstream_topics, idx), sub);
   }

#ifdef AST_DEVMODE
   ast_str_container_add(topic->statistics->subscribers, stasis_subscription_uniqueid(sub));
   ast_str_container_add(sub->statistics->topics, stasis_topic_name(topic));
#endif

   ao2_unlock(topic);

   return 0;
}

topic_complete_name()

static char* topic_complete_name ( const char *  word )

static

Definition at line 2391 of file stasis.c.

References ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_cli_completion_add(), ast_strdup, topic_proxy::name, and NULL.

Referenced by stasis_show_topic().

{
   struct topic_proxy *topic;
   struct ao2_iterator it;
   int wordlen = strlen(word);
   int ret;

   it = ao2_iterator_init(topic_all, 0);
   while ((topic = ao2_iterator_next(&it))) {
      if (!strncasecmp(word, topic->name, wordlen)) {
         ret = ast_cli_completion_add(ast_strdup(topic->name));
         if (ret) {
            ao2_ref(topic, -1);
            break;
         }
      }
      ao2_ref(topic, -1);
   }
   ao2_iterator_destroy(&it);
   return NULL;
}

topic_dtor()

static void topic_dtor ( void *  obj )

static

Definition at line 434 of file stasis.c.

References ao2_global_obj_ref, ao2_ref, ao2_unlink, ast_assert, ast_debug, AST_VECTOR_FREE, and AST_VECTOR_SIZE.

Referenced by stasis_topic_create_with_detail().

{
   struct stasis_topic *topic = obj;
#ifdef AST_DEVMODE
   struct ao2_container *topic_stats;
#endif

   ast_debug(2, "Destroying topic. name: %s, detail: %s\n",
         topic->name, topic->detail);

   /* Subscribers hold a reference to topics, so they should all be
    * unsubscribed before we get here. */
   ast_assert(AST_VECTOR_SIZE(&topic->subscribers) == 0);

   AST_VECTOR_FREE(&topic->subscribers);
   AST_VECTOR_FREE(&topic->upstream_topics);
   ast_debug(1, "Topic '%s': %p destroyed\n", topic->name, topic);

#ifdef AST_DEVMODE
   if (topic->statistics) {
      topic_stats = ao2_global_obj_ref(topic_statistics);
      if (topic_stats) {
         ao2_unlink(topic_stats, topic->statistics);
         ao2_ref(topic_stats, -1);
      }
      ao2_ref(topic->statistics, -1);
   }
#endif
}

topic_pool_dtor()

static void topic_pool_dtor ( void *  obj )

static

Definition at line 1744 of file stasis.c.

References ao2_cleanup, ao2_container_unregister(), ast_alloca, NULL, pool, stasis_topic_pool::pool_container, stasis_topic_pool::pool_topic, and stasis_topic_name().

Referenced by stasis_topic_pool_create().

{
   struct stasis_topic_pool *pool = obj;

#ifdef AO2_DEBUG
   {
      char *container_name =
         ast_alloca(strlen(stasis_topic_name(pool->pool_topic)) + strlen("-pool") + 1);
      sprintf(container_name, "%s-pool", stasis_topic_name(pool->pool_topic));
      ao2_container_unregister(container_name);
   }
#endif

   ao2_cleanup(pool->pool_container);
   pool->pool_container = NULL;
   ao2_cleanup(pool->pool_topic);
   pool->pool_topic = NULL;
}

topic_pool_entry_alloc()

static struct topic_pool_entry* topic_pool_entry_alloc ( const char *  topic_name )

static

Definition at line 1724 of file stasis.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, topic_pool_entry::name, NULL, and topic_pool_entry_dtor().

Referenced by stasis_topic_pool_get_topic().

{
   struct topic_pool_entry *topic_pool_entry;

   topic_pool_entry = ao2_alloc_options(sizeof(*topic_pool_entry) + strlen(topic_name) + 1,
      topic_pool_entry_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!topic_pool_entry) {
      return NULL;
   }

   strcpy(topic_pool_entry->name, topic_name); /* Safe */

   return topic_pool_entry;
}

topic_pool_entry_cmp()

static int topic_pool_entry_cmp ( void *  obj, void *  arg, int  flags  )

static

Definition at line 1784 of file stasis.c.

References ast_assert, CMP_MATCH, topic_pool_entry::name, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, OBJ_SEARCH_PARTIAL_KEY, stasis_topic_name(), and topic_pool_entry::topic.

Referenced by stasis_topic_pool_create().

{
   const struct topic_pool_entry *object_left = obj;
   const struct topic_pool_entry *object_right = arg;
   const char *right_key = arg;
   int cmp;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_OBJECT:
      right_key = object_right->name;
      /* Fall through */
   case OBJ_SEARCH_KEY:
      cmp = strcasecmp(object_left->name, right_key);
      break;
   case OBJ_SEARCH_PARTIAL_KEY:
      /* Not supported by container */
      ast_assert(0);
      cmp = -1;
      break;
   default:
      /*
       * What arg points to is specific to this traversal callback
       * and has no special meaning to astobj2.
       */
      cmp = 0;
      break;
   }
   if (cmp) {
      return 0;
   }
   /*
    * At this point the traversal callback is identical to a sorted
    * container.
    */
   return CMP_MATCH;
}

topic_pool_entry_dtor()

static void topic_pool_entry_dtor ( void *  obj )

static

Definition at line 1715 of file stasis.c.

References ao2_cleanup, topic_pool_entry::forward, NULL, stasis_forward_cancel(), and topic_pool_entry::topic.

Referenced by topic_pool_entry_alloc().

{
   struct topic_pool_entry *entry = obj;

   entry->forward = stasis_forward_cancel(entry->forward);
   ao2_cleanup(entry->topic);
   entry->topic = NULL;
}

topic_pool_entry_hash()

static int topic_pool_entry_hash ( const void *  obj, const int  flags  )

static

Definition at line 1763 of file stasis.c.

References ast_assert, ast_str_case_hash(), OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, and OBJ_SEARCH_OBJECT.

Referenced by stasis_topic_pool_create().

{
   const struct topic_pool_entry *object;
   const char *key;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_KEY:
      key = obj;
      break;
   case OBJ_SEARCH_OBJECT:
      object = obj;
      key = object->name;
      break;
   default:
      /* Hash can only work on something with a full key. */
      ast_assert(0);
      return 0;
   }
   return ast_str_case_hash(key);
}

topic_remove_subscription()

static int topic_remove_subscription ( struct stasis_topic *  topic, struct stasis_subscription *  sub  )

static

Definition at line 1231 of file stasis.c.

References ao2_lock, ao2_unlock, ast_str_container_remove(), AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_GET, AST_VECTOR_REMOVE_ELEM_UNORDERED, AST_VECTOR_SIZE, stasis_subscription_uniqueid(), and stasis_topic_name().

Referenced by proxy_dtor(), stasis_forward_cancel(), and stasis_unsubscribe().

{
   size_t idx;
   int res;

   ao2_lock(topic);
   for (idx = 0; idx < AST_VECTOR_SIZE(&topic->upstream_topics); ++idx) {
      topic_remove_subscription(
         AST_VECTOR_GET(&topic->upstream_topics, idx), sub);
   }
   res = AST_VECTOR_REMOVE_ELEM_UNORDERED(&topic->subscribers, sub,
      AST_VECTOR_ELEM_CLEANUP_NOOP);

#ifdef AST_DEVMODE
   if (!res) {
      ast_str_container_remove(topic->statistics->subscribers, stasis_subscription_uniqueid(sub));
      ast_str_container_remove(sub->statistics->topics, stasis_topic_name(topic));
   }
#endif

   ao2_unlock(topic);

   return res;
}

topic_statistics_cmp()

static int topic_statistics_cmp ( void *  obj, void *  arg, int  flags  )

static

Definition at line 3003 of file stasis.c.

References ast_assert, CMP_MATCH, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, and OBJ_SEARCH_PARTIAL_KEY.

Referenced by stasis_init().

{
   const struct stasis_topic_statistics *object_left = obj;
   const struct stasis_topic_statistics *object_right = arg;
   const char *right_key = arg;
   int cmp;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_OBJECT:
      right_key = object_right->name;
      /* Fall through */
   case OBJ_SEARCH_KEY:
      cmp = strcasecmp(object_left->name, right_key);
      break;
   case OBJ_SEARCH_PARTIAL_KEY:
      /* Not supported by container */
      ast_assert(0);
      cmp = -1;
      break;
   default:
      /*
       * What arg points to is specific to this traversal callback
       * and has no special meaning to astobj2.
       */
      cmp = 0;
      break;
   }
   if (cmp) {
      return 0;
   }
   /*
    * At this point the traversal callback is identical to a sorted
    * container.
    */
   return CMP_MATCH;
}

topic_statistics_complete_name()

static char* topic_statistics_complete_name ( const char *  word, int  state  )

static

Definition at line 2765 of file stasis.c.

References ao2_global_obj_ref, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, ast_strdup, NULL, result, and statistics().

Referenced by statistics_show_topic().

{
   struct stasis_topic_statistics *statistics;
   struct ao2_container *topic_stats;
   struct ao2_iterator it_statistics;
   int wordlen = strlen(word);
   int which = 0;
   char *result = NULL;

   topic_stats = ao2_global_obj_ref(topic_statistics);
   if (!topic_stats) {
      return result;
   }

   it_statistics = ao2_iterator_init(topic_stats, 0);
   while ((statistics = ao2_iterator_next(&it_statistics))) {
      if (!strncasecmp(word, statistics->name, wordlen)
         && ++which > state) {
         result = ast_strdup(statistics->name);
      }
      ao2_ref(statistics, -1);
      if (result) {
         break;
      }
   }
   ao2_iterator_destroy(&it_statistics);
   ao2_ref(topic_stats, -1);
   return result;
}

topic_statistics_destroy()

static void topic_statistics_destroy ( void *  obj )

static

Definition at line 465 of file stasis.c.

References ao2_cleanup, and statistics().

Referenced by stasis_topic_statistics_create().

{
   struct stasis_topic_statistics *statistics = obj;

   ao2_cleanup(statistics->subscribers);
}

topic_statistics_hash()

static int topic_statistics_hash ( const void *  obj, const int  flags  )

static

Definition at line 2982 of file stasis.c.

References ast_assert, ast_str_case_hash(), OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, and OBJ_SEARCH_OBJECT.

Referenced by stasis_init().

{
   const struct stasis_topic_statistics *object;
   const char *key;

   switch (flags & OBJ_SEARCH_MASK) {
   case OBJ_SEARCH_KEY:
      key = obj;
      break;
   case OBJ_SEARCH_OBJECT:
      object = obj;
      key = object->name;
      break;
   default:
      /* Hash can only work on something with a full key. */
      ast_assert(0);
      return 0;
   }
   return ast_str_case_hash(key);
}

userevent_exclusion_cb()

static int userevent_exclusion_cb ( const char *  key )

static

Definition at line 2147 of file stasis.c.

Referenced by multi_user_event_to_ami().

{
   if (!strcmp("eventname", key)) {
      return 1;
   }
   return 0;
}

Variable Documentation

cli_stasis

struct ast_cli_entry cli_stasis[]

static

Initial value:

= {
   { .handler =  stasis_show_topics , .summary =  "Show all topics" ,},
   { .handler =  stasis_show_topic , .summary =  "Show topic" ,},
}

Definition at line 2476 of file stasis.c.

cli_stasis_statistics

struct ast_cli_entry cli_stasis_statistics[]

static

Definition at line 2916 of file stasis.c.

declined_option

struct aco_type declined_option

static

An aco_type structure to link the "declined_message_types" category to the stasis_declined_config type.

Definition at line 2213 of file stasis.c.

declined_options

struct aco_type* declined_options[] = ACO_TYPES(&declined_option)

Definition at line 2221 of file stasis.c.

message_type_statistics_lock

ast_mutex_t message_type_statistics_lock = { PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP , NULL, {1, 0} }

static

Lock to protect the message types vector

Definition at line 347 of file stasis.c.

Referenced by publish_msg(), and statistics_show_messages().

stasis_conf

struct aco_file stasis_conf

Initial value:

= {
        .filename = "stasis.conf",
   .types = ACO_TYPES(&declined_option, &threadpool_option),
}

Definition at line 2223 of file stasis.c.

threadpool

struct ast_threadpool* threadpool

static

Thread pool for topics that don't want a dedicated taskprocessor

Definition at line 308 of file stasis.c.

threadpool_option

struct aco_type threadpool_option

static

Definition at line 2202 of file stasis.c.

threadpool_options

struct aco_type* threadpool_options[] = ACO_TYPES(&threadpool_option)

static

Definition at line 2210 of file stasis.c.

topic_all

struct ao2_container* topic_all

Definition at line 396 of file stasis.c.