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/taskpool.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_subscription
struct	stasis_taskpool_conf
	Taskpool 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_HEADERS   "%-64s %-64s\n"
#define	INITIAL_SUBSCRIBERS_MAX   4
#define	TOPIC_ALL_BUCKETS   997
#define	topic_lock_both(topic1, topic2)
	Lock two topics. More...
#define	TOPIC_POOL_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 (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)
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...
	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_taskpool, 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...
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...
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 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	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	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 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)
struct aco_file	stasis_conf
static struct ast_taskpool *	taskpool
static struct aco_type	taskpool_option
static struct aco_type *	taskpool_options [] = ACO_TYPES(&threadpool_option, &taskpool_option)
static struct aco_type	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

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

FMT_HEADERS

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

INITIAL_SUBSCRIBERS_MAX

#define INITIAL_SUBSCRIBERS_MAX   4

Initial size of the subscribers list.

Definition at line 368 of file stasis.c.

TOPIC_ALL_BUCKETS

#define TOPIC_ALL_BUCKETS   997

Definition at line 384 of file stasis.c.

topic_lock_both

#define topic_lock_both	(		topic1,
			topic2
	)

Lock two topics.

Definition at line 492 of file stasis.c.

TOPIC_POOL_BUCKETS

#define TOPIC_POOL_BUCKETS   57

The number of buckets to use for topic pools

Definition at line 371 of file stasis.c.

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.
file,lineno,func

Returns

New stasis_subscription object.

Return values

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 1009 of file stasis.c.

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

References callback(), stasis_subscription::data, make_ari_stubs::file, internal_stasis_subscribe(), and stasis_subscription::topic.

Referenced by stasis_message_router_create_internal().

__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.
file,lineno,func

Returns

New stasis_subscription object.

Return values

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 1020 of file stasis.c.

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

References callback(), stasis_subscription::data, make_ari_stubs::file, internal_stasis_subscribe(), and stasis_subscription::topic.

Referenced by stasis_message_router_create_internal().

AO2_GLOBAL_OBJ_STATIC()

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
	)

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 2381 of file stasis.c.

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

References ast_str_container_add(), ast_strlen_zero(), stasis_declined_config::declined, and var.

Referenced by stasis_init().

dispatch_exec_async()

static int dispatch_exec_async ( struct ast_taskprocessor_local *  local )

static

Definition at line 1326 of file stasis.c.

{
    struct stasis_subscription *sub = local->local_data;
    struct stasis_message *message = local->data;
 
    subscription_invoke(sub, message);
    ao2_cleanup(message);
 
    return 0;
}

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

Referenced by dispatch_message().

dispatch_exec_sync()

static int dispatch_exec_sync ( struct ast_taskprocessor_local *  local )

static

Definition at line 1353 of file stasis.c.

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

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, sub, subscription_invoke(), and sync_task_data::task_data.

Referenced by dispatch_message().

dispatch_message()

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

static

Definition at line 1379 of file stasis.c.

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

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, 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, sub, subscription_invoke(), and sync_task_data::task_data.

Referenced by publish_msg(), and send_subscription_unsubscribe().

forward_dtor()

static void forward_dtor ( void *  obj )

static

Definition at line 1605 of file stasis.c.

{
    struct stasis_forward *forward = obj;
 
    ao2_cleanup(forward->from_topic);
    forward->from_topic = NULL;
    ao2_cleanup(forward->to_topic);
    forward->to_topic = NULL;
}

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

Referenced by stasis_forward_all().

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.
file,lineno,func

Returns

New stasis_subscription object.

Return values

NULL

on error.

Since

12

Definition at line 923 of file stasis.c.

{
    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_taskpool, 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_taskpool ? '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_taskpool) {
            sub->mailbox = ast_taskpool_serializer(tps_name, taskpool);
        } 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;
}

References ao2_ref, ao2_t_alloc, ast_asprintf, ast_atomic_fetchadd_int(), ast_cond_init, ast_taskpool_serializer(), ast_taskprocessor_build_name(), ast_taskprocessor_get(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_set_local(), AST_VECTOR_INIT, callback(), stasis_subscription::data, make_ari_stubs::file, NULL, send_subscription_subscribe(), STASIS_SUBSCRIPTION_FILTER_NONE, STASIS_SUBSCRIPTION_FORMATTER_NONE, stasis_topic_name(), statistics(), sub, stasis_topic::subscriber_id, subscription_dtor(), taskpool, stasis_subscription::topic, topic_add_subscription(), and TPS_REF_DEFAULT.

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

link_topic_proxy()

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

static

Definition at line 567 of file stasis.c.

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

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, stasis_topic::creationtime, topic_proxy::creationtime, stasis_topic::detail, topic_proxy::detail, LOG_ERROR, name, stasis_topic::name, topic_proxy::name, NULL, OBJ_NOLOCK, proxy_dtor(), stasis_topic_get(), and topic_all.

Referenced by stasis_topic_create_with_detail().

multi_object_blob_dtor()

static void multi_object_blob_dtor ( void *  obj )

static

Definition at line 2026 of file stasis.c.

{
    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);
}

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

Referenced by ast_multi_object_blob_create().

multi_object_blob_to_ami()

static struct ast_str * multi_object_blob_to_ami ( void *  obj )

static

Definition at line 2163 of file stasis.c.

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

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, ast_multi_object_blob::snapshots, STASIS_UMOS_BRIDGE, STASIS_UMOS_CHANNEL, STASIS_UMOS_ENDPOINT, STASIS_UMOS_MAX, and type.

Referenced by multi_user_event_to_ami().

multi_user_event_to_ami()

static struct ast_manager_event_blob * multi_user_event_to_ami ( struct stasis_message *  message )

static

Definition at line 2222 of file stasis.c.

{
    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));
}

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().

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 2112 of file stasis.c.

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

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, ast_multi_object_blob::snapshots, stasis_message_data(), stasis_message_timestamp(), STASIS_UMOS_BRIDGE, STASIS_UMOS_CHANNEL, STASIS_UMOS_ENDPOINT, STASIS_UMOS_MAX, and type.

proxy_dtor()

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

static

Definition at line 479 of file stasis.c.

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

References ao2_cleanup, ao2_unlink, and container.

Referenced by link_topic_proxy().

publish_msg()

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

static

Definition at line 1497 of file stasis.c.

{
    size_t i;
#ifdef AST_DEVMODE
    unsigned int dispatched = 0;
    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);
#ifdef AST_DEVMODE
        dispatched +=
#endif
            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);
}

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(), message_type_id, NULL, stasis_message_type(), stasis_message_type_id(), stasis_topic_subscribers(), statistics(), sub, stasis_topic::subscribers, and stasis_subscription::topic.

Referenced by stasis_publish(), and stasis_publish_sync().

send_subscription_subscribe()

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

static

Definition at line 1715 of file stasis.c.

{
    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);
}

References ao2_cleanup, ast_assert, stasis_message_create(), stasis_publish(), stasis_subscription_change_type(), stasis_subscription_is_subscribed(), sub, and subscription_change_alloc().

Referenced by internal_stasis_subscribe().

send_subscription_unsubscribe()

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

static

Definition at line 1743 of file stasis.c.

{
    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);
}

References ao2_cleanup, ast_assert, dispatch_message(), stasis_message_create(), stasis_publish(), stasis_subscription_change_type(), stasis_subscription_is_subscribed(), sub, and subscription_change_alloc().

Referenced by stasis_unsubscribe().

stasis_cleanup()

static void stasis_cleanup ( void  )

static

Cleanup function for graceful shutdowns.

Definition at line 3119 of file stasis.c.

{
#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_taskpool_shutdown(taskpool);
    taskpool = 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);
}

References aco_info_destroy(), ao2_cleanup, ao2_global_obj_release, ARRAY_LEN, ast_cli_unregister_multiple(), ast_multi_user_event_type(), ast_taskpool_shutdown(), AST_VECTOR_FREE, cli_stasis, globals, NULL, STASIS_MESSAGE_TYPE_CLEANUP, stasis_subscription_change_type(), taskpool, and topic_all.

Referenced by stasis_init().

stasis_config_alloc()

static void * stasis_config_alloc ( void  )

static

Definition at line 2330 of file stasis.c.

{
    struct stasis_config *cfg;
 
    if (!(cfg = ao2_alloc(sizeof(*cfg), stasis_config_destructor))) {
        return NULL;
    }
 
    cfg->taskpool_options = ast_calloc(1, sizeof(*cfg->taskpool_options));
    if (!cfg->taskpool_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;
}

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::taskpool_options.

Referenced by stasis_init().

stasis_config_destructor()

static void stasis_config_destructor ( void *  obj )

static

Definition at line 2322 of file stasis.c.

{
    struct stasis_config *cfg = obj;
 
    ao2_cleanup(cfg->declined_message_types);
    ast_free(cfg->taskpool_options);
}

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

Referenced by stasis_config_alloc().

stasis_declined_config_destructor()

static void stasis_declined_config_destructor ( void *  obj )

static

Definition at line 2315 of file stasis.c.

{
    struct stasis_declined_config *declined = obj;
 
    ao2_cleanup(declined->declined);
}

References ao2_cleanup, and stasis_declined_config::declined.

Referenced by stasis_config_alloc().

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.

Return values

NULL

on error.

Since

12

Definition at line 1645 of file stasis.c.

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

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_topic::subscribers, stasis_forward::to_topic, topic_add_subscription(), topic_lock_both, and stasis_topic::upstream_topics.

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().

stasis_forward_cancel()

struct stasis_forward * stasis_forward_cancel

(

struct stasis_forward *

forward

)

Definition at line 1615 of file stasis.c.

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

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_topic::subscribers, stasis_forward::to_topic, topic_lock_both, topic_remove_subscription(), and stasis_topic::upstream_topics.

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().

stasis_init()

int stasis_init

(

void

)

Initialize the Stasis subsystem.

Returns

0 on success.

Non-zero on error.

Since

12

Definition at line 3138 of file stasis.c.

{
    struct stasis_config *cfg;
    int cache_init;
    struct ast_taskpool_options taskpool_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, "minimum_size", ACO_EXACT,
        taskpool_options, "5", OPT_INT_T, PARSE_IN_RANGE,
        FLDSET(struct stasis_taskpool_conf, minimum_size), 0,
        INT_MAX);
    aco_option_register(&cfg_info, "initial_size", ACO_EXACT,
        taskpool_options, "5", OPT_INT_T, PARSE_IN_RANGE,
        FLDSET(struct stasis_taskpool_conf, initial_size), 0,
        INT_MAX);
    aco_option_register(&cfg_info, "idle_timeout_sec", ACO_EXACT,
        taskpool_options, "20", OPT_INT_T, PARSE_IN_RANGE,
        FLDSET(struct stasis_taskpool_conf, idle_timeout_sec), 0,
        INT_MAX);
    aco_option_register(&cfg_info, "max_size", ACO_EXACT,
        taskpool_options, "50", OPT_INT_T, PARSE_IN_RANGE,
        FLDSET(struct stasis_taskpool_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(&taskpool_option, "taskpool", default_cfg->taskpool_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;
        }
    }
 
    taskpool_opts.version = AST_TASKPOOL_OPTIONS_VERSION;
    taskpool_opts.minimum_size = cfg->taskpool_options->minimum_size;
    taskpool_opts.initial_size = cfg->taskpool_options->initial_size;
    taskpool_opts.auto_increment = 1;
    taskpool_opts.max_size = cfg->taskpool_options->max_size;
    taskpool_opts.idle_timeout = cfg->taskpool_options->idle_timeout_sec;
    taskpool = ast_taskpool_create("stasis", &taskpool_opts);
    ao2_ref(cfg, -1);
    if (!taskpool) {
        ast_log(LOG_ERROR, "Failed to create 'stasis-core' taskpool\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;
}

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_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, AST_VECTOR_INIT, ast_taskpool_options::auto_increment, cli_stasis, declined_handler(), stasis_config::declined_message_types, declined_option, declined_options, FLDSET, globals, ast_taskpool_options::idle_timeout, stasis_taskpool_conf::idle_timeout_sec, ast_taskpool_options::initial_size, stasis_taskpool_conf::initial_size, LOG_ERROR, LOG_NOTICE, ast_taskpool_options::max_size, stasis_taskpool_conf::max_size, ast_taskpool_options::minimum_size, stasis_taskpool_conf::minimum_size, OPT_INT_T, PARSE_IN_RANGE, stasis_cache_init(), stasis_cleanup(), stasis_config_alloc(), STASIS_MESSAGE_TYPE_INIT, stasis_subscription_change_type(), taskpool, taskpool_option, stasis_config::taskpool_options, taskpool_options, topic_all, TOPIC_ALL_BUCKETS, and ast_taskpool_options::version.

Referenced by asterisk_daemon().

stasis_log_bad_type_access()

void stasis_log_bad_type_access

(

const char *

name

)

Definition at line 2007 of file stasis.c.

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

References ast_log, LOG_ERROR, name, and stasis_message_type_declined().

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 2360 of file stasis.c.

{
    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_debug(4, "Declining to allocate Stasis message type '%s' due to configuration\n", name);
    }
    return res;
}

References ao2_cleanup, ao2_find, ao2_global_obj_ref, ao2_ref, ast_debug, stasis_declined_config::declined, stasis_config::declined_message_types, globals, name, and OBJ_SEARCH_KEY.

Referenced by stasis_log_bad_type_access(), and stasis_message_type_create().

STASIS_MESSAGE_TYPE_DEFN() [1/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_MESSAGE_TYPE_DEFN() [2/2]

STASIS_MESSAGE_TYPE_DEFN

(

stasis_subscription_change_type

)

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 1578 of file stasis.c.

{
    publish_msg(topic, message, NULL);
}

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

Referenced by 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_multi_object_blob_single_channel_publish(), ast_publish_device_state_full(), ast_refer_notify_transfer_request(), 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(), detect_callback(), 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(), 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().

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_subscription is subscribed to are also delivered the message; this delivery however happens asynchronously.

Since

12.1.0

Definition at line 1583 of file stasis.c.

{
    ast_assert(sub != NULL);
 
    publish_msg(sub->topic, message, sub);
}

References ast_assert, NULL, publish_msg(), and sub.

Referenced by AST_TEST_DEFINE(), and stasis_message_router_publish_sync().

stasis_show_topic()

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

static

Definition at line 2494 of file stasis.c.

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

References a, ao2_lock, ao2_ref, ao2_unlock, 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, stasis_topic::creationtime, stasis_topic::detail, stasis_topic::name, NULL, stasis_topic_get(), stasis_topic::subscribers, stasis_subscription::topic, topic_complete_name(), stasis_subscription::uniqueid, stasis_topic::upstream_topics, and ast_cli_entry::usage.

stasis_show_topics()

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

static

Definition at line 2411 of file stasis.c.

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

References a, 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_entry::args, ast_cli(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, topic_proxy::detail, FMT_FIELDS, FMT_HEADERS, topic_proxy::name, NULL, OBJ_SEARCH_OBJECT, topic_all, and ast_cli_entry::usage.

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 1160 of file stasis.c.

{
    ast_assert(subscription != NULL);
 
    ao2_lock(subscription->topic);
    subscription->accepted_formatters = formatters;
    ao2_unlock(subscription->topic);
 
    return;
}

References stasis_subscription::accepted_formatters, 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().

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 1090 of file stasis.c.

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

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

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(), common_config_load(), 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().

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 876 of file stasis.c.

{
}

Referenced by 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 1120 of file stasis.c.

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

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

Referenced by AST_TEST_DEFINE().

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 1241 of file stasis.c.

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

References stasis_subscription_change::description, stasis_message_data(), stasis_message_type(), stasis_subscription_change_type(), stasis_subscription_uniqueid(), sub, 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_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().

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 1185 of file stasis.c.

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

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().

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 1217 of file stasis.c.

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

References ao2_lock, ao2_unlock, AST_VECTOR_GET, AST_VECTOR_SIZE, sub, and stasis_topic::subscribers.

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().

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 1172 of file stasis.c.

{
    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);
    }
}

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().

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 1078 of file stasis.c.

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

References ast_taskprocessor_alert_set_levels(), and stasis_subscription::mailbox.

Referenced by stasis_message_router_set_congestion_limits().

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 1144 of file stasis.c.

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

References ao2_lock, ao2_unlock, filter(), stasis_subscription::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().

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 1236 of file stasis.c.

{
    return sub->uniqueid;
}

References sub.

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

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.

Return values

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 684 of file stasis.c.

{
    return stasis_topic_create_with_detail(name, "");
}

References name, and 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().

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.

Return values

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 635 of file stasis.c.

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

References ao2_ref, ao2_t_alloc, ast_debug, AST_VECTOR_INIT, stasis_topic::detail, INITIAL_SUBSCRIBERS_MAX, link_topic_proxy(), name, stasis_topic::name, NULL, stasis_topic_get(), stasis_topic::subscribers, topic_dtor(), and stasis_topic::upstream_topics.

Referenced by stasis_topic_create().

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.

Return values

NULL	if topic is NULL.

Since

12

Definition at line 702 of file stasis.c.

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

References stasis_topic::detail, and NULL.

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.

Return values

NULL	on error or not exist.

Note

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

Definition at line 689 of file stasis.c.

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

References ao2_weakproxy_find, name, OBJ_SEARCH_KEY, and topic_all.

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

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.

Return values

NULL	if topic is NULL.

Definition at line 694 of file stasis.c.

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

References stasis_topic::name, and 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(), and topic_remove_subscription().

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

Return values

NULL	on failure

Definition at line 1900 of file stasis.c.

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

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, 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().

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>

Since

13.24

15.6

16.1

Definition at line 1931 of file stasis.c.

{
    /*
     * 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);
}

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().

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

Return values

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

Definition at line 1951 of file stasis.c.

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

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().

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 1994 of file stasis.c.

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

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

Referenced by ast_bridge_topic_exists(), and ast_publish_device_state_full().

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 710 of file stasis.c.

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

References AST_VECTOR_SIZE, and stasis_topic::subscribers.

Referenced by caching_topic_exec(), and publish_msg().

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.

Return values

NULL	for convenience

Since

12

Definition at line 1038 of file stasis.c.

{
    /* 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;
}

References ao2_bump, ao2_cleanup, ast_log, ast_taskprocessor_push(), LOG_ERROR, NULL, send_subscription_unsubscribe(), sub, sub_cleanup(), and topic_remove_subscription().

Referenced by AST_TEST_DEFINE(), cc_generic_agent_destructor(), common_config_unload(), 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(), 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().

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.

Return values

NULL	for convenience

Since

12

Definition at line 1201 of file stasis.c.

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

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

Referenced by 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(), remove_device_state_subscription(), rtp_reload(), stasis_message_router_unsubscribe_and_join(), stasis_state_unsubscribe_and_join(), unload_module(), and unload_pbx().

sub_cleanup()

static int sub_cleanup ( void *  data )

static

Definition at line 1031 of file stasis.c.

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

References ao2_cleanup, stasis_subscription::data, and sub.

Referenced by stasis_unsubscribe().

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 1694 of file stasis.c.

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

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().

subscription_change_dtor()

static void subscription_change_dtor ( void *  obj )

static

Definition at line 1687 of file stasis.c.

{
    struct stasis_subscription_change *change = obj;
 
    ao2_cleanup(change->topic);
}

References ao2_cleanup, and stasis_subscription_change::topic.

Referenced by subscription_change_alloc().

subscription_dtor()

static void subscription_dtor ( void *  obj )

static

Definition at line 781 of file stasis.c.

{
    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
}

References ao2_cleanup, ao2_global_obj_ref, ao2_ref, ao2_unlink, ast_assert, ast_cond_destroy, ast_free, ast_taskprocessor_unreference(), AST_VECTOR_FREE, NULL, stasis_subscription_is_done(), stasis_subscription_is_subscribed(), and sub.

Referenced by internal_stasis_subscribe().

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.
message	Message to send.

Definition at line 821 of file stasis.c.

{
    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
}

References ao2_lock, ao2_unlock, ast_cond_signal, ast_tvdiff_ms(), ast_tvnow(), AST_VECTOR_GET, AST_VECTOR_SIZE, message_type_id, stasis_message_type(), stasis_message_type_id(), stasis_subscription_change_type(), STASIS_SUBSCRIPTION_FILTER_SELECTIVE, stasis_subscription_final_message(), statistics(), and sub.

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

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 1268 of file stasis.c.

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

References ao2_lock, ao2_unlock, ast_str_container_add(), AST_VECTOR_APPEND, AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_subscription_uniqueid(), stasis_topic_name(), sub, stasis_topic::subscribers, stasis_subscription_change::topic, topic_add_subscription(), and stasis_topic::upstream_topics.

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

topic_complete_name()

static char * topic_complete_name ( const char *  word )

static

Definition at line 2468 of file stasis.c.

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

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

Referenced by stasis_show_topic().

topic_dtor()

static void topic_dtor ( void *  obj )

static

Definition at line 500 of file stasis.c.

{
    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
}

References ao2_global_obj_ref, ao2_ref, ao2_unlink, ast_assert, ast_debug, AST_VECTOR_FREE, AST_VECTOR_SIZE, stasis_topic::detail, stasis_topic::name, stasis_topic::subscribers, and stasis_topic::upstream_topics.

Referenced by stasis_topic_create_with_detail().

topic_pool_dtor()

static void topic_pool_dtor ( void *  obj )

static

Definition at line 1811 of file stasis.c.

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

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

Referenced by stasis_topic_pool_create().

topic_pool_entry_alloc()

static struct topic_pool_entry * topic_pool_entry_alloc ( const char *  topic_name )

static

Definition at line 1791 of file stasis.c.

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

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().

topic_pool_entry_cmp()

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

static

Definition at line 1851 of file stasis.c.

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

References ast_assert, CMP_MATCH, topic_pool_entry::name, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, and OBJ_SEARCH_PARTIAL_KEY.

Referenced by stasis_topic_pool_create().

topic_pool_entry_dtor()

static void topic_pool_entry_dtor ( void *  obj )

static

Definition at line 1782 of file stasis.c.

{
    struct topic_pool_entry *entry = obj;
 
    entry->forward = stasis_forward_cancel(entry->forward);
    ao2_cleanup(entry->topic);
    entry->topic = NULL;
}

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

Referenced by topic_pool_entry_alloc().

topic_pool_entry_hash()

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

static

Definition at line 1830 of file stasis.c.

{
    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);
}

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

Referenced by stasis_topic_pool_create().

topic_remove_subscription()

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

static

Definition at line 1296 of file stasis.c.

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

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(), stasis_topic_name(), sub, stasis_topic::subscribers, stasis_subscription_change::topic, topic_remove_subscription(), and stasis_topic::upstream_topics.

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

userevent_exclusion_cb()

static int userevent_exclusion_cb ( const char *  key )

static

Definition at line 2214 of file stasis.c.

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

Referenced by multi_user_event_to_ami().

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 2553 of file stasis.c.

Referenced by stasis_cleanup(), and stasis_init().

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 2290 of file stasis.c.

Referenced by stasis_init().

declined_options

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

Definition at line 2298 of file stasis.c.

Referenced by stasis_init().

stasis_conf

struct aco_file stasis_conf

Initial value:

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

Definition at line 2300 of file stasis.c.

taskpool

struct ast_taskpool* taskpool

static

Taskpool for topics that don't want a dedicated taskprocessor

Definition at line 374 of file stasis.c.

Referenced by ast_serializer_taskpool_create(), internal_stasis_subscribe(), stasis_cleanup(), and stasis_init().

taskpool_option

struct aco_type taskpool_option

static

Definition at line 2279 of file stasis.c.

Referenced by stasis_init().

taskpool_options

struct aco_type* taskpool_options[] = ACO_TYPES(&threadpool_option, &taskpool_option)

static

Definition at line 2287 of file stasis.c.

Referenced by stasis_init().

threadpool_option

struct aco_type threadpool_option

static

Definition at line 2271 of file stasis.c.

topic_all

struct ao2_container* topic_all

Definition at line 462 of file stasis.c.

Referenced by link_topic_proxy(), stasis_cleanup(), stasis_init(), stasis_show_topics(), stasis_topic_get(), and topic_complete_name().