stasis.h File Reference

Stasis Message Bus API. See Stasis Message Bus API for detailed documentation. More...

#include "asterisk/json.h"
#include "asterisk/manager.h"
#include "asterisk/utils.h"
#include "asterisk/event.h"

Include dependency graph for stasis.h:

Go to the source code of this file.

Data Structures
struct	stasis_cache_update
	Cache update message. More...
struct	stasis_message_sanitizer
	Structure containing callbacks for Stasis message sanitization. More...
struct	stasis_message_vtable
	Virtual table providing methods for messages. More...
struct	stasis_subscription_change
	Holds details about changes to subscriptions for the specified topic. More...

Macros
#define	STASIS_MESSAGE_TYPE_CLEANUP(name)
	Boiler-plate messaging macro for cleaning up message types.
#define	STASIS_MESSAGE_TYPE_DEFN(name, ...)
	Boiler-plate messaging macro for defining public message types.
#define	STASIS_MESSAGE_TYPE_DEFN_LOCAL(name, ...)
	Boiler-plate messaging macro for defining local message types.
#define	STASIS_MESSAGE_TYPE_INIT(name)
	Boiler-plate messaging macro for initializing message types.
#define	stasis_subscribe(topic, callback, data)   __stasis_subscribe(topic, callback, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	stasis_subscribe_pool(topic, callback, data)   __stasis_subscribe_pool(topic, callback, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	STASIS_UMOS_MAX   (STASIS_UMOS_ENDPOINT + 1)
	Number of snapshot types.

Typedefs
typedef struct stasis_message *(*	cache_aggregate_calc_fn) (struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
	Callback to calculate the aggregate cache entry.
typedef void(*	cache_aggregate_publish_fn) (struct stasis_topic *topic, struct stasis_message *aggregate)
	Callback to publish the aggregate cache entry message.
typedef const char *(*	snapshot_get_id) (struct stasis_message *message)
	Callback extract a unique identity from a snapshot message.
typedef void(*	stasis_subscription_cb) (void *data, struct stasis_subscription *sub, struct stasis_message *message)
	Callback function type for Stasis subscriptions.

Enumerations
enum	stasis_message_type_result { STASIS_MESSAGE_TYPE_ERROR = -1 , STASIS_MESSAGE_TYPE_SUCCESS , STASIS_MESSAGE_TYPE_DECLINED }
	Return code for Stasis message type creation attempts. More...
enum	stasis_subscription_message_filter { STASIS_SUBSCRIPTION_FILTER_NONE = 0 , STASIS_SUBSCRIPTION_FILTER_FORCED_NONE , STASIS_SUBSCRIPTION_FILTER_SELECTIVE }
	Stasis subscription message filters. More...
enum	stasis_subscription_message_formatters { STASIS_SUBSCRIPTION_FORMATTER_NONE = 0 , STASIS_SUBSCRIPTION_FORMATTER_JSON = 1 << 0 , STASIS_SUBSCRIPTION_FORMATTER_AMI = 1 << 1 , STASIS_SUBSCRIPTION_FORMATTER_EVENT = 1 << 2 }
	Stasis subscription formatter filters. More...
enum	stasis_user_multi_object_snapshot_type { STASIS_UMOS_CHANNEL = 0 , STASIS_UMOS_BRIDGE , STASIS_UMOS_ENDPOINT }
	Object type code for multi user object snapshots. More...

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.
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 task pool.
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 to a multi object blob previously created.
struct ast_multi_object_blob *	ast_multi_object_blob_create (struct ast_json *blob)
	Create a stasis multi object blob.
void	ast_multi_object_blob_single_channel_publish (struct ast_channel *chan, struct stasis_message_type *type, struct ast_json *blob)
	Create and publish a stasis message blob on a channel with it's snapshot.
struct stasis_message_type *	ast_multi_user_event_type (void)
	Message type for custom user defined events with multi object blobs.
struct stasis_message *	stasis_cache_clear_create (struct stasis_message *message)
	A message which instructs the caching topic to remove an entry from its cache.
struct stasis_message_type *	stasis_cache_clear_type (void)
	Message type for clearing a message from a stasis cache.
struct stasis_cache *	stasis_cache_create (snapshot_get_id id_fn)
	Create a cache.
struct stasis_cache *	stasis_cache_create_full (snapshot_get_id id_fn, cache_aggregate_calc_fn aggregate_calc_fn, cache_aggregate_publish_fn aggregate_publish_fn)
	Create a cache.
struct ao2_container *	stasis_cache_dump (struct stasis_cache *cache, struct stasis_message_type *type)
	Dump cached items to a subscription for the ast_eid_default entity.
struct ao2_container *	stasis_cache_dump_all (struct stasis_cache *cache, struct stasis_message_type *type)
	Dump all entity items from the cache to a subscription.
struct ao2_container *	stasis_cache_dump_by_eid (struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
	Dump cached items to a subscription for a specific entity.
struct stasis_message *	stasis_cache_entry_get_aggregate (struct stasis_cache_entry *entry)
	Get the aggregate cache entry snapshot.
struct stasis_message *	stasis_cache_entry_get_local (struct stasis_cache_entry *entry)
	Get the local entity's cache entry snapshot.
struct stasis_message *	stasis_cache_entry_get_remote (struct stasis_cache_entry *entry, int idx)
	Get a remote entity's cache entry snapshot by index.
struct stasis_message *	stasis_cache_get (struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
	Retrieve an item from the cache for the ast_eid_default entity.
struct ao2_container *	stasis_cache_get_all (struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
	Retrieve all matching entity items from the cache.
struct stasis_message *	stasis_cache_get_by_eid (struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid)
	Retrieve an item from the cache for a specific entity.
int	stasis_cache_init (void)
struct stasis_message_type *	stasis_cache_update_type (void)
	Message type for cache update messages.
int	stasis_caching_accept_message_type (struct stasis_caching_topic *caching_topic, struct stasis_message_type *type)
	Indicate to a caching topic that we are interested in a message type.
struct stasis_topic *	stasis_caching_get_topic (struct stasis_caching_topic *caching_topic)
	Returns the topic of cached events from a caching topics.
int	stasis_caching_set_filter (struct stasis_caching_topic *caching_topic, enum stasis_subscription_message_filter filter)
	Set the message type filtering level on a cache.
struct stasis_caching_topic *	stasis_caching_topic_create (struct stasis_topic *original_topic, struct stasis_cache *cache)
	Create a topic which monitors and caches messages from another topic.
struct stasis_caching_topic *	stasis_caching_unsubscribe (struct stasis_caching_topic *caching_topic)
	Unsubscribes a caching topic from its upstream topic.
struct stasis_caching_topic *	stasis_caching_unsubscribe_and_join (struct stasis_caching_topic *caching_topic)
	Unsubscribes a caching topic from its upstream topic, blocking until all messages have been forwarded.
int	stasis_config_init (void)
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.
struct stasis_forward *	stasis_forward_cancel (struct stasis_forward *forward)
int	stasis_init (void)
	Initialize the Stasis subsystem.
void	stasis_log_bad_type_access (const char *name)
int	stasis_message_can_be_ami (struct stasis_message *msg)
	Determine if the given message can be converted to AMI.
struct stasis_message *	stasis_message_create (struct stasis_message_type *type, void *data)
	Create a new message.
struct stasis_message *	stasis_message_create_full (struct stasis_message_type *type, void *data, const struct ast_eid *eid)
	Create a new message for an entity.
void *	stasis_message_data (const struct stasis_message *msg)
	Get the data contained in a message.
const struct ast_eid *	stasis_message_eid (const struct stasis_message *msg)
	Get the entity id for a stasis_message.
const struct timeval *	stasis_message_timestamp (const struct stasis_message *msg)
	Get the time when a message was created.
struct ast_manager_event_blob *	stasis_message_to_ami (struct stasis_message *msg)
	Build the AMI representation of the message.
struct ast_event *	stasis_message_to_event (struct stasis_message *msg)
	Build the Generic event system representation of the message.
struct ast_json *	stasis_message_to_json (struct stasis_message *msg, struct stasis_message_sanitizer *sanitize)
	Build the JSON representation of the message.
struct stasis_message_type *	stasis_message_type (const struct stasis_message *msg)
	Get the message type for a stasis_message.
enum stasis_subscription_message_formatters	stasis_message_type_available_formatters (const struct stasis_message_type *message_type)
	Get a bitmap of available formatters for a message type.
enum stasis_message_type_result	stasis_message_type_create (const char *name, struct stasis_message_vtable *vtable, struct stasis_message_type **result)
	Create a new message type.
int	stasis_message_type_declined (const char *name)
	Check whether a message type is declined.
unsigned int	stasis_message_type_hash (const struct stasis_message_type *type)
	Gets the hash of a given message type.
int	stasis_message_type_id (const struct stasis_message_type *type)
	Gets the id of a given message type.
const char *	stasis_message_type_name (const struct stasis_message_type *type)
	Gets the name of a given message type.
void	stasis_publish (struct stasis_topic *topic, struct stasis_message *message)
	Publish a message to a topic's subscribers.
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.
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.
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.
void	stasis_subscription_cb_noop (void *data, struct stasis_subscription *sub, struct stasis_message *message)
	Stasis subscription callback function that does nothing.
struct stasis_message_type *	stasis_subscription_change_type (void)
	Gets the message type for subscription change notices.
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.
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.
int	stasis_subscription_is_done (struct stasis_subscription *subscription)
	Returns whether subscription has received its final message.
int	stasis_subscription_is_subscribed (const struct stasis_subscription *sub)
	Returns whether a subscription is currently subscribed.
void	stasis_subscription_join (struct stasis_subscription *subscription)
	Block until the last message is processed on a subscription.
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.
int	stasis_subscription_set_filter (struct stasis_subscription *subscription, enum stasis_subscription_message_filter filter)
	Set the message type filtering level on a subscription.
const char *	stasis_subscription_uniqueid (const struct stasis_subscription *sub)
	Get the unique ID for the subscription.
struct stasis_topic *	stasis_topic_create (const char *name)
	Create a new topic.
struct stasis_topic *	stasis_topic_create_with_detail (const char *name, const char *detail)
	Create a new topic with given detail.
const char *	stasis_topic_detail (const struct stasis_topic *topic)
	Return the detail of a topic.
struct stasis_topic *	stasis_topic_get (const char *name)
	Get a topic of the given name.
const char *	stasis_topic_name (const struct stasis_topic *topic)
	Return the name of a topic.
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.
void	stasis_topic_pool_delete_topic (struct stasis_topic_pool *pool, const char *topic_name)
	Delete a topic from the topic pool.
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.
int	stasis_topic_pool_topic_exists (const struct stasis_topic_pool *pool, const char *topic_name)
	Check if a topic exists in a pool.
size_t	stasis_topic_subscribers (const struct stasis_topic *topic)
	Return the number of subscribers of a topic.
const char *	stasis_topic_uniqueid (const struct stasis_topic *topic)
	Return the uniqueid of a topic.
struct stasis_subscription *	stasis_unsubscribe (struct stasis_subscription *subscription)
	Cancel a subscription.
struct stasis_subscription *	stasis_unsubscribe_and_join (struct stasis_subscription *subscription)
	Cancel a subscription, blocking until the last message is processed.

Detailed Description

Stasis Message Bus API. See Stasis Message Bus API for detailed documentation.

Author

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

Since

12

Definition in file stasis.h.

Macro Definition Documentation

STASIS_MESSAGE_TYPE_CLEANUP

#define STASIS_MESSAGE_TYPE_CLEANUP

(

name

)

Value:

    ({                  \
        ao2_cleanup(_priv_ ## name);    \
        _priv_ ## name = NULL;      \
    })

Boiler-plate messaging macro for cleaning up message types.

Note that if your type is defined in core instead of a loadable module, you should call message type cleanup from an ast_register_cleanup() handler instead of an ast_register_atexit() handler.

The reason is that during an immediate shutdown, loadable modules (which may refer to core message types) are not unloaded. While the atexit handlers are run, there's a window of time where a module subscription might reference a core message type after it's been cleaned up. Which is bad.

Parameters

name	Name of message type.

Since

12

Definition at line 1515 of file stasis.h.

     {                  \
        ao2_cleanup(_priv_ ## name);    \
        _priv_ ## name = NULL;      \
    })

STASIS_MESSAGE_TYPE_DEFN

#define STASIS_MESSAGE_TYPE_DEFN	(		name,
			...
	)

Boiler-plate messaging macro for defining public message types.

STASIS_MESSAGE_TYPE_DEFN(ast_foo_type,
   .to_ami = foo_to_ami,
   .to_json = foo_to_json,
   .to_event = foo_to_event,
   );

Parameters

name	Name of message type.
...	Virtual table methods for messages of this type.

Since

12

Definition at line 1440 of file stasis.h.

                                                               {    \
        __VA_ARGS__                     \
    };                              \
    static struct stasis_message_type *_priv_ ## name;      \
    struct stasis_message_type *name(void) {            \
        if (_priv_ ## name == NULL) {               \
            stasis_log_bad_type_access(#name);      \
        }                           \
        return _priv_ ## name;                  \
    }

STASIS_MESSAGE_TYPE_DEFN_LOCAL

#define STASIS_MESSAGE_TYPE_DEFN_LOCAL	(		name,
			...
	)

Boiler-plate messaging macro for defining local message types.

STASIS_MESSAGE_TYPE_DEFN_LOCAL(ast_foo_type,
   .to_ami = foo_to_ami,
   .to_json = foo_to_json,
   .to_event = foo_to_event,
   );

Parameters

name	Name of message type.
...	Virtual table methods for messages of this type.

Since

12

Definition at line 1467 of file stasis.h.

                                                               {    \
        __VA_ARGS__                     \
    };                              \
    static struct stasis_message_type *_priv_ ## name;      \
    static struct stasis_message_type *name(void) {         \
        if (_priv_ ## name == NULL) {               \
            stasis_log_bad_type_access(#name);      \
        }                           \
        return _priv_ ## name;                  \
    }

STASIS_MESSAGE_TYPE_INIT

#define STASIS_MESSAGE_TYPE_INIT

(

name

)

Value:

    ({                              \
        ast_assert(_priv_ ## name == NULL);         \
        stasis_message_type_create(#name,   \
            &_priv_ ## name ## _v, &_priv_ ## name) == STASIS_MESSAGE_TYPE_ERROR ? 1 : 0;   \
    })

Boiler-plate messaging macro for initializing message types.

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

Parameters

name	Name of message type.

Returns

0 if initialization is successful.

Non-zero on failure.

Since

12

Definition at line 1493 of file stasis.h.

     {                              \
        ast_assert(_priv_ ## name == NULL);         \
        stasis_message_type_create(#name,   \
            &_priv_ ## name ## _v, &_priv_ ## name) == STASIS_MESSAGE_TYPE_ERROR ? 1 : 0;   \
    })

stasis_subscribe

#define stasis_subscribe	(		topic,
			callback,
			data
	)		__stasis_subscribe(topic, callback, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 649 of file stasis.h.

stasis_subscribe_pool

#define stasis_subscribe_pool	(		topic,
			callback,
			data
	)		__stasis_subscribe_pool(topic, callback, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 680 of file stasis.h.

Typedef Documentation

cache_aggregate_calc_fn

typedef struct stasis_message *(* cache_aggregate_calc_fn) (struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)

Callback to calculate the aggregate cache entry.

Since

12.2.0

Parameters

entry	Cache entry to calculate a new aggregate snapshot.
new_snapshot	The shapshot that is being updated.

Note

Return a ref bumped pointer from stasis_cache_entry_get_aggregate() if a new aggregate could not be calculated because of error.

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Returns

New aggregate-snapshot calculated on success. Caller has a reference on return.

Definition at line 1009 of file stasis.h.

cache_aggregate_publish_fn

typedef void(* cache_aggregate_publish_fn) (struct stasis_topic *topic, struct stasis_message *aggregate)

Callback to publish the aggregate cache entry message.

Since

12.2.0

Once an aggregate message is calculated. This callback publishes the message so subscribers will know the new value of an aggregated state.

Parameters

topic	The aggregate message may be published to this topic. It is the topic to which the cache itself is subscribed.
aggregate	The aggregate shapshot message to publish.

Note

It is up to the function to determine if there is a better topic the aggregate message should be published over.

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Definition at line 1055 of file stasis.h.

snapshot_get_id

typedef const char *(* snapshot_get_id) (struct stasis_message *message)

Callback extract a unique identity from a snapshot message.

This identity is unique to the underlying object of the snapshot, such as the UniqueId field of a channel.

Parameters

message

Message to extract id from.

Returns

String representing the snapshot's id.

Return values

NULL	if the message_type of the message isn't a handled snapshot.

Since

12

Definition at line 1009 of file stasis.h.

stasis_subscription_cb

typedef void(* stasis_subscription_cb) (void *data, struct stasis_subscription *sub, struct stasis_message *message)

Callback function type for Stasis subscriptions.

Parameters

data	Data field provided with subscription.
sub	Subscription published on.
message	Published message.

Since

12

Definition at line 611 of file stasis.h.

Enumeration Type Documentation

stasis_message_type_result

enum stasis_message_type_result

Return code for Stasis message type creation attempts.

Enumerator
STASIS_MESSAGE_TYPE_ERROR	Message type was not created due to allocation failure
STASIS_MESSAGE_TYPE_SUCCESS	Message type was created successfully
STASIS_MESSAGE_TYPE_DECLINED	Message type was not created due to configuration

Definition at line 285 of file stasis.h.

                                {
    STASIS_MESSAGE_TYPE_ERROR = -1, /*!< Message type was not created due to allocation failure */
    STASIS_MESSAGE_TYPE_SUCCESS,    /*!< Message type was created successfully */
    STASIS_MESSAGE_TYPE_DECLINED,   /*!< Message type was not created due to configuration */
};

stasis_subscription_message_filter

enum stasis_subscription_message_filter

Stasis subscription message filters.

Enumerator
STASIS_SUBSCRIPTION_FILTER_NONE	No filter is in place, all messages are raised
STASIS_SUBSCRIPTION_FILTER_FORCED_NONE	No filter is in place or can be set, all messages are raised
STASIS_SUBSCRIPTION_FILTER_SELECTIVE	Only messages of allowed message types are raised

Definition at line 294 of file stasis.h.

                                        {
    STASIS_SUBSCRIPTION_FILTER_NONE = 0,    /*!< No filter is in place, all messages are raised */
    STASIS_SUBSCRIPTION_FILTER_FORCED_NONE, /*!< No filter is in place or can be set, all messages are raised */
    STASIS_SUBSCRIPTION_FILTER_SELECTIVE,   /*!< Only messages of allowed message types are raised */
};

stasis_subscription_message_formatters

enum stasis_subscription_message_formatters

Stasis subscription formatter filters.

There should be an entry here for each member of stasis_message_vtable

Since

13.25.0

16.2.0

Enumerator
STASIS_SUBSCRIPTION_FORMATTER_NONE
STASIS_SUBSCRIPTION_FORMATTER_JSON	Allow messages with a to_json formatter
STASIS_SUBSCRIPTION_FORMATTER_AMI	Allow messages with a to_ami formatter
STASIS_SUBSCRIPTION_FORMATTER_EVENT	Allow messages with a to_event formatter

Definition at line 308 of file stasis.h.

                                            {
    STASIS_SUBSCRIPTION_FORMATTER_NONE =  0,
    STASIS_SUBSCRIPTION_FORMATTER_JSON =  1 << 0,  /*!< Allow messages with a to_json formatter */
    STASIS_SUBSCRIPTION_FORMATTER_AMI =   1 << 1,  /*!< Allow messages with a to_ami formatter */
    STASIS_SUBSCRIPTION_FORMATTER_EVENT = 1 << 2,  /*!< Allow messages with a to_event formatter */
};

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, 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 task 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 taskpool 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, internal_stasis_subscribe(), and stasis_subscription::topic.

Referenced by stasis_message_router_create_internal().

stasis_cache_clear_create()

struct stasis_message * stasis_cache_clear_create

(

struct stasis_message *

message

)

A message which instructs the caching topic to remove an entry from its cache.

Parameters

message

Message representative of the cache entry that should be cleared. This will become the data held in the stasis_cache_clear message.

Returns

Message which, when sent to a stasis_caching_topic, will clear the item from the cache.

Return values

NULL

on error.

Since

12

Definition at line 778 of file stasis_cache.c.

{
    return stasis_message_create(stasis_cache_clear_type(), id_message);
}

References stasis_cache_clear_type(), and stasis_message_create().

Referenced by ast_delete_mwi_state_full(), ast_device_state_clear_cache(), ast_endpoint_shutdown(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), clear_node_cache(), and remove_device_states_cb().

stasis_cache_create()

struct stasis_cache * stasis_cache_create

(

snapshot_get_id

id_fn

)

Create a cache.

This is the backend store for a stasis_caching_topic. The cache is thread safe, allowing concurrent reads and writes.

The returned object is AO2 managed, so ao2_cleanup() when you're done.

Parameters

id_fn

Callback to extract the id from a snapshot message.

Returns

New cache indexed by id_fn.

Return values

NULL

on error

Since

12

Definition at line 360 of file stasis_cache.c.

{
    return stasis_cache_create_full(id_fn, NULL, NULL);
}

References stasis_cache::id_fn, NULL, and stasis_cache_create_full().

Referenced by ast_presence_state_engine_init(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), mwi_init(), and stasis_cp_all_create().

stasis_cache_create_full()

struct stasis_cache * stasis_cache_create_full	(	snapshot_get_id	id_fn,
		cache_aggregate_calc_fn	aggregate_calc_fn,
		cache_aggregate_publish_fn	aggregate_publish_fn
	)

Create a cache.

This is the backend store for a stasis_caching_topic. The cache is thread safe, allowing concurrent reads and writes.

The returned object is AO2 managed, so ao2_cleanup() when you're done.

Parameters

id_fn	Callback to extract the id from a snapshot message.
aggregate_calc_fn	Callback to calculate the aggregate cache entry.
aggregate_publish_fn	Callback to publish the aggregate cache entry.

Note

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Returns

New cache indexed by id_fn.

Return values

NULL

on error

Since

12.2.0

Definition at line 334 of file stasis_cache.c.

{
    struct stasis_cache *cache;
 
    cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
        AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!cache) {
        return NULL;
    }
 
    cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
        NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp);
    if (!cache->entries) {
        ao2_cleanup(cache);
        return NULL;
    }
 
    cache->id_fn = id_fn;
    cache->aggregate_calc_fn = aggregate_calc_fn;
    cache->aggregate_publish_fn = aggregate_publish_fn;
 
    return cache;
}

References stasis_cache::aggregate_calc_fn, stasis_cache::aggregate_publish_fn, AO2_ALLOC_OPT_LOCK_NOLOCK, AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_alloc_options, ao2_cleanup, ao2_container_alloc_hash, cache, cache_dtor(), cache_entry_cmp(), cache_entry_hash(), stasis_cache::id_fn, NULL, and NUM_CACHE_BUCKETS.

Referenced by AST_TEST_DEFINE(), devstate_init(), and stasis_cache_create().

stasis_cache_dump()

struct ao2_container * stasis_cache_dump	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type
	)

Dump cached items to a subscription for the ast_eid_default entity.

Parameters

cache	The cache to query.
type	Type of message to dump (any type if NULL).

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Since

12

Definition at line 736 of file stasis_cache.c.

{
    return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
}

References ast_eid_default, cache, stasis_cache_dump_by_eid(), and type.

Referenced by action_presencestatelist(), ast_ari_endpoints_list(), ast_ari_endpoints_list_by_tech(), AST_TEST_DEFINE(), asterisk_publication_devicestate_refresh(), asterisk_publication_mwi_refresh(), asterisk_start_devicestate_publishing(), asterisk_start_mwi_publishing(), endpoints_scrape_cb(), load_module(), load_module(), unload_module(), and xmpp_init_event_distribution().

stasis_cache_dump_all()

struct ao2_container * stasis_cache_dump_all	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type
	)

Dump all entity items from the cache to a subscription.

Since

12.2.0

Parameters

cache	The cache to query.
type	Type of message to dump (any type if NULL).

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Definition at line 757 of file stasis_cache.c.

{
    struct cache_dump_data cache_dump;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
 
    cache_dump.eid = NULL;
    cache_dump.type = type;
    cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
    if (!cache_dump.container) {
        return NULL;
    }
 
    ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump);
    return cache_dump.container;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache, cache_dump_all_cb(), cache_dump_data::container, cache_dump_data::eid, NULL, OBJ_MULTIPLE, OBJ_NODATA, type, and cache_dump_data::type.

Referenced by AST_TEST_DEFINE(), cache_cleanup(), and cleanup_module().

stasis_cache_dump_by_eid()

struct ao2_container * stasis_cache_dump_by_eid	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const struct ast_eid *	eid
	)

Dump cached items to a subscription for a specific entity.

Since

12.2.0

Parameters

cache	The cache to query.
type	Type of message to dump (any type if NULL).
eid	Specific entity id to retrieve. NULL for aggregate.

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Definition at line 718 of file stasis_cache.c.

{
    struct cache_dump_data cache_dump;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
 
    cache_dump.eid = eid;
    cache_dump.type = type;
    cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
    if (!cache_dump.container) {
        return NULL;
    }
 
    ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump);
    return cache_dump.container;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache, cache_dump_by_eid_cb(), cache_dump_data::container, cache_dump_data::eid, NULL, OBJ_MULTIPLE, OBJ_NODATA, type, and cache_dump_data::type.

Referenced by action_devicestatelist(), AST_TEST_DEFINE(), cpg_confchg_cb(), and stasis_cache_dump().

stasis_cache_entry_get_aggregate()

struct stasis_message * stasis_cache_entry_get_aggregate

(

struct stasis_cache_entry *

entry

)

Get the aggregate cache entry snapshot.

Since

12.2.0

Parameters

entry

Cache entry to get the aggregate snapshot.

Note

A reference is not given to the returned pointer so don't unref it.

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Return values

Aggregate-snapshot	in cache.
NULL	if not present.

Definition at line 365 of file stasis_cache.c.

{
    return entry->aggregate;
}

References stasis_cache_entry::aggregate.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

stasis_cache_entry_get_local()

struct stasis_message * stasis_cache_entry_get_local

(

struct stasis_cache_entry *

entry

)

Get the local entity's cache entry snapshot.

Since

12.2.0

Parameters

entry

Cache entry to get the local entity's snapshot.

Note

A reference is not given to the returned pointer so don't unref it.

Return values

Internal-snapshot	in cache.
NULL	if not present.

Definition at line 370 of file stasis_cache.c.

{
    return entry->local;
}

References stasis_cache_entry::local.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

stasis_cache_entry_get_remote()

struct stasis_message * stasis_cache_entry_get_remote	(	struct stasis_cache_entry *	entry,
		int	idx
	)

Get a remote entity's cache entry snapshot by index.

Since

12.2.0

Parameters

entry	Cache entry to get a remote entity's snapshot.
idx	Which remote entity's snapshot to get.

Note

A reference is not given to the returned pointer so don't unref it.

Return values

Remote-entity-snapshot	in cache.
NULL	if not present.

Definition at line 375 of file stasis_cache.c.

{
    if (idx < AST_VECTOR_SIZE(&entry->remote)) {
        return AST_VECTOR_GET(&entry->remote, idx);
    }
    return NULL;
}

References AST_VECTOR_GET, AST_VECTOR_SIZE, NULL, and stasis_cache_entry::remote.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

stasis_cache_get()

struct stasis_message * stasis_cache_get	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const char *	id
	)

Retrieve an item from the cache for the ast_eid_default entity.

The returned item is AO2 managed, so ao2_cleanup() when you're done with it.

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.

Returns

Message from the cache.

Return values

NULL	if message is not found.

Since

12

Definition at line 686 of file stasis_cache.c.

{
    return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
}

References ast_eid_default, cache, stasis_cache_get_by_eid(), and type.

Referenced by ast_endpoint_latest_snapshot(), AST_TEST_DEFINE(), has_voicemail(), presence_state_cached(), unistim_send_mwi_to_peer(), and update_registry().

stasis_cache_get_all()

struct ao2_container * stasis_cache_get_all	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const char *	id
	)

Retrieve all matching entity items from the cache.

Since

12.2.0

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.

Returns

Container of matching items found.

Return values

NULL

if error.

Definition at line 587 of file stasis_cache.c.

{
    struct stasis_cache_entry *cached_entry;
    struct ao2_container *found;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
    ast_assert(id != NULL);
 
    if (!type) {
        return NULL;
    }
 
    found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
    if (!found) {
        return NULL;
    }
 
    ao2_rdlock(cache->entries);
 
    cached_entry = cache_find(cache->entries, type, id);
    if (cached_entry && cache_entry_dump(found, cached_entry)) {
        ao2_cleanup(found);
        found = NULL;
    }
 
    ao2_unlock(cache->entries);
 
    ao2_cleanup(cached_entry);
    return found;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_list, ao2_rdlock, ao2_unlock, ast_assert, cache, cache_entry_dump(), cache_find(), NULL, and type.

Referenced by AST_TEST_DEFINE().

stasis_cache_get_by_eid()

struct stasis_message * stasis_cache_get_by_eid	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const char *	id,
		const struct ast_eid *	eid
	)

Retrieve an item from the cache for a specific entity.

The returned item is AO2 managed, so ao2_cleanup() when you're done with it.

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.
eid	Specific entity id to retrieve. NULL for aggregate.

Note

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Returns

Message from the cache.

Return values

NULL	if message is not found.

Since

12.2.0

Definition at line 659 of file stasis_cache.c.

{
    struct stasis_cache_entry *cached_entry;
    struct stasis_message *snapshot = NULL;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
    ast_assert(id != NULL);
 
    if (!type) {
        return NULL;
    }
 
    ao2_rdlock(cache->entries);
 
    cached_entry = cache_find(cache->entries, type, id);
    if (cached_entry) {
        snapshot = cache_entry_by_eid(cached_entry, eid);
        ao2_bump(snapshot);
    }
 
    ao2_unlock(cache->entries);
 
    ao2_cleanup(cached_entry);
    return snapshot;
}

References ao2_bump, ao2_cleanup, ao2_rdlock, ao2_unlock, ast_assert, cache, cache_entry_by_eid(), cache_find(), stasis_message::eid, NULL, and type.

Referenced by ast_delete_mwi_state_full(), ast_device_state_clear_cache(), AST_TEST_DEFINE(), check_cache_aggregate(), devstate_cached(), and stasis_cache_get().

stasis_cache_init()

int stasis_cache_init

(

void

)

Definition at line 1008 of file stasis_cache.c.

{
    ast_register_cleanup(stasis_cache_cleanup);
 
    if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
        return -1;
    }
 
    if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
        return -1;
    }
 
    return 0;
}

References ast_register_cleanup(), stasis_cache_cleanup(), stasis_cache_clear_type(), stasis_cache_update_type(), and STASIS_MESSAGE_TYPE_INIT.

Referenced by stasis_init().

stasis_caching_accept_message_type()

int stasis_caching_accept_message_type	(	struct stasis_caching_topic *	caching_topic,
		struct stasis_message_type *	type
	)

Indicate to a caching topic that we are interested in a message type.

This will cause the caching topic to receive messages of the given message type. This enables internal filtering in the stasis message bus to reduce messages.

Parameters

caching_topic	The caching topic.
type	The message type we wish to receive.

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 90 of file stasis_cache.c.

{
    int res;
 
    if (!caching_topic) {
        return -1;
    }
 
    /* We wait to accept the stasis specific message types until now so that by default everything
     * will flow to us.
     */
    res = stasis_subscription_accept_message_type(caching_topic->sub, stasis_cache_clear_type());
    res |= stasis_subscription_accept_message_type(caching_topic->sub, stasis_subscription_change_type());
    res |= stasis_subscription_accept_message_type(caching_topic->sub, type);
 
    return res;
}

References stasis_cache_clear_type(), stasis_subscription_accept_message_type(), stasis_subscription_change_type(), stasis_caching_topic::sub, and type.

Referenced by ast_presence_state_engine_init(), devstate_init(), and stasis_cp_single_accept_message_type().

stasis_caching_get_topic()

struct stasis_topic * stasis_caching_get_topic

(

struct stasis_caching_topic *

caching_topic

)

Returns the topic of cached events from a caching topics.

Parameters

caching_topic

The caching topic.

Returns

The topic that publishes cache update events, along with passthrough events from the underlying topic.

Return values

NULL	if caching_topic is NULL.

Since

12

Definition at line 85 of file stasis_cache.c.

{
    return caching_topic->topic;
}

References stasis_caching_topic::topic.

Referenced by ast_device_state_topic_cached(), ast_mwi_topic_cached(), ast_presence_state_topic_cached(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), stasis_cp_single_create(), and stasis_cp_single_topic_cached().

stasis_caching_set_filter()

int stasis_caching_set_filter	(	struct stasis_caching_topic *	caching_topic,
		enum stasis_subscription_message_filter	filter
	)

Set the message type filtering level on a cache.

This will cause the underlying 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

caching_topic	The caching topic.
filter	What filter to use

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 109 of file stasis_cache.c.

{
    if (!caching_topic) {
        return -1;
    }
    return stasis_subscription_set_filter(caching_topic->sub, filter);
}

References filter(), stasis_subscription_set_filter(), and stasis_caching_topic::sub.

Referenced by ast_presence_state_engine_init(), devstate_init(), and stasis_cp_single_set_filter().

stasis_caching_topic_create()

struct stasis_caching_topic * stasis_caching_topic_create	(	struct stasis_topic *	original_topic,
		struct stasis_cache *	cache
	)

Create a topic which monitors and caches messages from another topic.

The idea is that some topics publish 'snapshots' of some other object's state that should be cached. When these snapshot messages are received, the cache is updated, and a stasis_cache_update() message is forwarded, which has both the original snapshot message and the new message.

The returned object is AO2 managed, so ao2_cleanup() when done with it.

Parameters

original_topic	Topic publishing snapshot messages.
cache	Backend cache in which to keep snapshots.

Returns

New topic which changes snapshot messages to stasis_cache_update() messages, and forwards all other messages from the original topic.

Return values

NULL

on error

Since

12

Definition at line 948 of file stasis_cache.c.

{
    struct stasis_caching_topic *caching_topic;
    static int caching_id;
    char *new_name;
    int ret;
 
    ret = ast_asprintf(&new_name, "cache:%d/%s", ast_atomic_fetchadd_int(&caching_id, +1), stasis_topic_name(original_topic));
    if (ret < 0) {
        return NULL;
    }
 
    caching_topic = ao2_alloc_options(sizeof(*caching_topic),
        stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (caching_topic == NULL) {
        ast_free(new_name);
 
        return NULL;
    }
 
    caching_topic->topic = stasis_topic_create(new_name);
    if (caching_topic->topic == NULL) {
        ao2_ref(caching_topic, -1);
        ast_free(new_name);
 
        return NULL;
    }
 
    ao2_ref(cache, +1);
    caching_topic->cache = cache;
    if (!cache->registered) {
        if (ao2_container_register(new_name, cache->entries, print_cache_entry)) {
            ast_log(LOG_ERROR, "Stasis cache container '%p' for '%s' did not register\n",
                cache->entries, new_name);
        } else {
            cache->registered = 1;
        }
    }
    ast_free(new_name);
 
    caching_topic->sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0, 0, __FILE__, __LINE__, __PRETTY_FUNCTION__);
    if (caching_topic->sub == NULL) {
        ao2_ref(caching_topic, -1);
 
        return NULL;
    }
 
    ao2_ref(original_topic, +1);
    caching_topic->original_topic = original_topic;
 
    /* The subscription holds the reference, so no additional ref bump. */
    return caching_topic;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_container_register(), ao2_ref, ast_asprintf, ast_atomic_fetchadd_int(), ast_free, ast_log, stasis_caching_topic::cache, cache, caching_topic_exec(), internal_stasis_subscribe(), LOG_ERROR, NULL, stasis_caching_topic::original_topic, print_cache_entry(), stasis_caching_topic_dtor(), stasis_topic_create(), stasis_topic_name(), stasis_caching_topic::sub, and stasis_caching_topic::topic.

Referenced by ast_presence_state_engine_init(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), devstate_init(), mwi_init(), and stasis_cp_sink_create().

stasis_caching_unsubscribe()

struct stasis_caching_topic * stasis_caching_unsubscribe

(

struct stasis_caching_topic *

caching_topic

)

Unsubscribes a caching topic from its upstream topic.

This function returns immediately, so be sure to cleanup when stasis_subscription_final_message() is received.

Parameters

caching_topic

Caching topic to unsubscribe

Return values

NULL	for convenience

Since

12

Definition at line 119 of file stasis_cache.c.

{
    if (!caching_topic) {
        return NULL;
    }
 
    /*
     * The subscription may hold the last reference to this caching
     * topic, but we want to make sure the unsubscribe finishes
     * before kicking of the caching topic's dtor.
     */
    ao2_ref(caching_topic, +1);
 
    if (stasis_subscription_is_subscribed(caching_topic->sub)) {
        /*
         * Increment the reference to hold on to it past the
         * unsubscribe. Will be cleaned up in dtor.
         */
        ao2_ref(caching_topic->sub, +1);
        stasis_unsubscribe(caching_topic->sub);
    } else {
        ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
    }
    ao2_cleanup(caching_topic);
    return NULL;
}

References ao2_cleanup, ao2_ref, ast_log, LOG_ERROR, NULL, stasis_subscription_is_subscribed(), stasis_unsubscribe(), and stasis_caching_topic::sub.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), stasis_caching_unsubscribe_and_join(), and stasis_cp_single_unsubscribe().

stasis_caching_unsubscribe_and_join()

struct stasis_caching_topic * stasis_caching_unsubscribe_and_join

(

struct stasis_caching_topic *

caching_topic

)

Unsubscribes a caching topic from its upstream topic, blocking until all messages have been forwarded.

See stasis_unsubscribe_and_join() for more info on when to use this as opposed to stasis_caching_unsubscribe().

Parameters

caching_topic

Caching topic to unsubscribe

Return values

NULL	for convenience

Since

12

Definition at line 146 of file stasis_cache.c.

{
    if (!caching_topic) {
        return NULL;
    }
 
    /* Hold a ref past the unsubscribe */
    ao2_ref(caching_topic, +1);
    stasis_caching_unsubscribe(caching_topic);
    stasis_subscription_join(caching_topic->sub);
    ao2_cleanup(caching_topic);
    return NULL;
}

References ao2_cleanup, ao2_ref, NULL, stasis_caching_unsubscribe(), stasis_subscription_join(), and stasis_caching_topic::sub.

Referenced by AST_TEST_DEFINE(), devstate_cleanup(), mwi_cleanup(), and presence_state_engine_cleanup().

stasis_config_init()

int stasis_config_init

(

void

)

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(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), create_subscriptions(), create_subscriptions(), endpoint_internal_create(), forwards_create_bridge(), forwards_create_channel(), forwards_create_endpoint(), load_general_config(), load_module(), load_module(), 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(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), bridge_channel_control_thread(), cleanup_module(), destroy_subscriptions(), destroy_subscriptions(), forwards_unsubscribe(), load_general_config(), load_module(), 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(), unload_module(), unload_module(), 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_can_be_ami()

int stasis_message_can_be_ami

(

struct stasis_message *

msg

)

Determine if the given message can be converted to AMI.

Parameters

msg	Message to see if can be converted to AMI.

Return values

0	Cannot be converted
non-zero	Can be converted

Definition at line 251 of file stasis_message.c.

{
    return HAS_VIRTUAL(to_ami, msg);
}

References HAS_VIRTUAL, and to_ami().

Referenced by manager_default_msg_cb().

stasis_message_create()

struct stasis_message * stasis_message_create	(	struct stasis_message_type *	type,
		void *	data
	)

Create a new message.

This message is an ao2 object, and must be ao2_cleanup()'ed when you are done with it. Messages are also immutable, and must not be modified after they are initialized. Especially the data in the message.

Parameters

type	Type of the message
data	Immutable data that is the actual contents of the message

Returns

New message

Return values

NULL

on error

Since

12

Definition at line 174 of file stasis_message.c.

{
    return stasis_message_create_full(type, data, &ast_eid_default);
}

References ast_eid_default, stasis_message::data, stasis_message_create_full(), and type.

Referenced by aoc_publish_blob(), ast_bridge_blob_create(), ast_bridge_blob_create_from_snapshots(), ast_bridge_publish_attended_transfer(), ast_bridge_publish_blind_transfer(), ast_bridge_publish_merge(), ast_bridge_publish_state(), ast_cdr_engine_term(), ast_channel_publish_dial_internal(), ast_channel_publish_final_snapshot(), ast_channel_publish_snapshot(), ast_endpoint_blob_create(), ast_manager_publish_event(), ast_multi_object_blob_single_channel_publish(), ast_mwi_blob_create(), ast_refer_notify_transfer_request(), ast_rtp_publish_rtcp_message(), ast_system_publish_registry(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), bridge_publish_state_from_blob(), bridge_topics_destroy(), cc_publish(), cdr_prop_write(), cdr_read(), cdr_write(), create_channel_blob_message(), create_endpoint_snapshot_message(), endpoint_publish_snapshot(), forkcdr_exec(), handle_security_event(), local_optimization_finished_cb(), local_optimization_started_cb(), presence_state_event(), publish_acl_change(), publish_app_cdr_message(), publish_chanspy_message(), 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_parked_call(), publish_parked_call_failure(), queue_publish_member_blob(), queue_publish_multi_channel_snapshot_blob(), send_call_pickup_stasis_message(), send_msg(), send_start_msg_snapshots(), send_subscription_subscribe(), send_subscription_unsubscribe(), stasis_app_user_event(), stasis_cache_clear_create(), stasis_test_message_create(), stun_monitor_request(), and update_create().

stasis_message_create_full()

struct stasis_message * stasis_message_create_full	(	struct stasis_message_type *	type,
		void *	data,
		const struct ast_eid *	eid
	)

Create a new message for an entity.

This message is an ao2 object, and must be ao2_cleanup()'ed when you are done with it. Messages are also immutable, and must not be modified after they are initialized. Especially the data in the message.

Parameters

type	Type of the message
data	Immutable data that is the actual contents of the message
eid	What entity originated this message. (NULL for aggregate)

Note

An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Return values

New	message
NULL	on error

Since

12.2.0

Definition at line 140 of file stasis_message.c.

{
    struct stasis_message *message;
 
    if (type == NULL || data == NULL) {
        return NULL;
    }
 
    message = ao2_t_alloc_options(sizeof(*message), stasis_message_dtor,
        AO2_ALLOC_OPT_LOCK_NOLOCK, type->name);
    if (message == NULL) {
        return NULL;
    }
 
    message->timestamp = ast_tvnow();
    /*
     * XXX Normal ao2 ref counting rules says we should increment the message
     * type ref here and decrement it in stasis_message_dtor().  However, the
     * stasis message could be cached and legitimately cause the type ref count
     * to hit the excessive ref count assertion.  Since the message type
     * practically has to be a global object anyway, we can get away with not
     * holding a ref in the stasis message.
     */
    message->type = type;
    ao2_ref(data, +1);
    message->data = data;
    if (eid) {
        message->eid_ptr = &message->eid;
        message->eid = *eid;
    }
 
    return message;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_ref, ao2_t_alloc_options, ast_tvnow(), stasis_message::data, stasis_message::eid, NULL, stasis_message_dtor(), and type.

Referenced by ast_publish_device_state_full(), AST_TEST_DEFINE(), cache_test_message_create_full(), create_foo_type_message(), device_state_aggregate_calc(), mwi_state_create_message(), and stasis_message_create().

stasis_message_data()

void * stasis_message_data

(

const struct stasis_message *

msg

)

Get the data contained in a message.

Parameters

msg

Message.

Returns

Immutable data pointer

Return values

NULL	if msg is NULL.

Since

12

Definition at line 195 of file stasis_message.c.

{
    if (msg == NULL) {
        return NULL;
    }
    return msg->data;
}

References stasis_message::data, and NULL.

Referenced by agent_login_to_ami(), agent_logoff_to_ami(), agi_channel_to_ami(), aoc_to_ami(), appcdr_callback(), ari_transfer_to_json(), ast_ari_endpoints_list(), ast_ari_endpoints_list_by_tech(), ast_bridge_merge_message_to_json(), ast_bridge_publish_enter(), ast_bridge_publish_leave(), ast_channel_entered_bridge_to_json(), ast_channel_left_bridge_to_json(), ast_delete_mwi_state_full(), ast_endpoint_latest_snapshot(), ast_mwi_publish_by_mailbox(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), asterisk_publisher_devstate_cb(), asterisk_publisher_mwistate_cb(), attended_transfer_to_ami(), attended_transfer_to_json(), blind_transfer_to_ami(), blind_transfer_to_json(), bridge_attended_transfer_handler(), bridge_blind_transfer_handler(), bridge_merge_cb(), bridge_merge_handler(), bridge_snapshot_update(), cache_test_aggregate_calc_fn(), cache_test_data_id(), cache_update(), cache_update_cb(), caching_topic_exec(), call_forwarded_handler(), call_pickup_to_ami(), cc_available_to_ami(), cc_callerrecalling_to_ami(), cc_callerstartmonitoring_to_ami(), cc_callerstopmonitoring_to_ami(), cc_failure_to_ami(), cc_monitorfailed_to_ami(), cc_offertimerstart_to_ami(), cc_recallcomplete_to_ami(), cc_requestacknowledged_to_ami(), cc_requested_to_ami(), cdr_prop_write_callback(), cdr_read_callback(), cdr_write_callback(), cel_attended_transfer_cb(), cel_blind_transfer_cb(), cel_bridge_enter_cb(), cel_bridge_leave_cb(), cel_dial_cb(), cel_generic_cb(), cel_local_optimization_cb_helper(), cel_parking_cb(), cel_pickup_cb(), cel_snapshot_update_cb(), channel_blob_to_json(), channel_chanspy_start_cb(), channel_chanspy_stop_cb(), channel_dial_cb(), channel_dtmf_begin_cb(), channel_dtmf_end_cb(), channel_enter_cb(), channel_fax_cb(), channel_flash_cb(), channel_hangup_handler_cb(), channel_hangup_request_cb(), channel_hold_cb(), channel_leave_cb(), channel_mixmonitor_mute_cb(), channel_mixmonitor_start_cb(), channel_mixmonitor_stop_cb(), channel_moh_start_cb(), channel_moh_stop_cb(), channel_snapshot_update(), channel_unhold_cb(), channel_wink_cb(), check_cache_aggregate(), conf_send_event_to_participants(), confbridge_atxfer_cb(), confbridge_publish_manager_event(), confbridge_talking_cb(), consumer_exec(), contactstatus_to_ami(), contactstatus_to_json(), corosync_ping_to_event(), dahdichannel_to_ami(), device_state_aggregate_calc(), device_state_cb(), device_state_cb(), device_state_cb(), device_state_cb(), device_state_get_id(), devstate_cached(), devstate_change_cb(), devstate_to_ami(), devstate_to_event(), dial_to_json(), dtmf_end_to_json(), dump_cache_load(), dump_cache_unload(), dump_consumer(), endpoint_snapshot_get_id(), endpoints_scrape_cb(), explicit_publish_cb(), fake_ami(), fake_json(), find_route(), forkcdr_callback(), generic_agent_devstate_cb(), generic_monitor_devstate_cb(), get_bool_header(), handle_attended_transfer(), handle_blind_transfer(), handle_bridge_enter(), handle_bridge_enter_message(), handle_bridge_leave_message(), handle_channel_snapshot_update_message(), handle_dial_message(), handle_hangup(), handle_hint_change_message_type(), handle_local_optimization_begin(), handle_local_optimization_end(), handle_masquerade(), handle_mwi_state(), handle_parked_call_message(), has_voicemail(), hold_to_json(), implicit_publish_cb(), is_msg(), local_message_to_ami(), manager_generic_msg_cb(), meetme_stasis_cb(), moh_post_start(), moh_post_stop(), multi_user_event_to_ami(), multi_user_event_to_json(), mwi_app_event_cb(), mwi_startup_event_cb(), mwi_state_get_id(), mwi_to_event(), mwi_update_cb(), mwi_update_cb(), park_announce_update_cb(), parker_update_cb(), parking_event_cb(), peerstatus_to_ami(), peerstatus_to_json(), playback_to_json(), presence_state_cached(), presence_state_cb(), presence_state_get_id(), presence_state_to_ami(), queue_agent_cb(), queue_channel_to_ami(), queue_member_to_ami(), queue_multi_channel_to_ami(), recording_to_json(), refer_progress_bridge(), remove_device_states_cb(), rtcp_report_to_ami(), rtcp_report_to_json(), security_event_to_ami(), security_stasis_cb(), stasis_end_to_json(), stasis_start_to_json(), stasis_state_subscriber_data(), stasis_subscription_final_message(), sub_bridge_update_handler(), sub_channel_update_handler(), sub_endpoint_update_handler(), subscription_persistence_event_cb(), system_registry_to_ami(), talking_start_to_ami(), talking_stop_to_ami(), unhold_to_json(), unistim_send_mwi_to_peer(), update_registry(), updates(), varset_to_ami(), xmpp_pubsub_devstate_cb(), and xmpp_pubsub_mwi_cb().

stasis_message_eid()

const struct ast_eid * stasis_message_eid

(

const struct stasis_message *

msg

)

Get the entity id for a stasis_message.

Since

12.2.0

Parameters

msg	Message to get eid.

Return values

Entity	id of msg
NULL	if msg is an aggregate or msg is NULL.

Definition at line 179 of file stasis_message.c.

{
    if (msg == NULL) {
        return NULL;
    }
    return msg->eid_ptr;
}

References stasis_message::eid_ptr, and NULL.

Referenced by AST_TEST_DEFINE(), cache_entry_by_eid(), cache_entry_create(), cache_remove(), cache_update(), caching_topic_exec(), and clear_node_cache().

stasis_message_timestamp()

const struct timeval * stasis_message_timestamp

(

const struct stasis_message *

msg

)

Get the time when a message was created.

Parameters

msg

Message.

Returns

Pointer to the timeval when the message was created.

Return values

NULL	if msg is NULL.

Since

12

Definition at line 203 of file stasis_message.c.

{
    if (msg == NULL) {
        return NULL;
    }
    return &msg->timestamp;
}

References NULL, and stasis_message::timestamp.

Referenced by ari_transfer_to_json(), ast_bridge_merge_message_to_json(), ast_channel_entered_bridge_to_json(), ast_channel_left_bridge_to_json(), AST_TEST_DEFINE(), attended_transfer_to_json(), blind_transfer_to_json(), cel_attended_transfer_cb(), cel_blind_transfer_cb(), cel_bridge_enter_cb(), cel_bridge_leave_cb(), cel_dial_cb(), cel_generic_cb(), cel_local_optimization_cb_helper(), cel_parking_cb(), cel_pickup_cb(), cel_snapshot_update_cb(), channel_blob_to_json(), contactstatus_to_json(), dial_to_json(), dtmf_end_to_json(), handle_bridge_enter_message(), handle_bridge_leave_message(), handle_channel_snapshot_update_message(), handle_dial_message(), handle_parked_call_message(), hold_to_json(), multi_user_event_to_json(), pack_bridge_and_channel(), pack_bridge_and_channels(), peerstatus_to_json(), playback_to_json(), recording_to_json(), sub_bridge_update_handler(), sub_channel_update_handler(), sub_endpoint_update_handler(), unhold_to_json(), and updates().

stasis_message_to_ami()

struct ast_manager_event_blob * stasis_message_to_ami

(

struct stasis_message *

msg

)

Build the AMI representation of the message.

May return NULL, to indicate no representation. The returned object should be ao2_cleanup()'ed.

Parameters

msg	Message to convert to AMI.

Return values

NULL	if AMI format is not supported.

Definition at line 224 of file stasis_message.c.

{
    return INVOKE_VIRTUAL(to_ami, msg);
}

References INVOKE_VIRTUAL, and to_ami().

Referenced by action_devicestatelist(), action_presencestatelist(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and manager_default_msg_cb().

stasis_message_to_event()

struct ast_event * stasis_message_to_event

(

struct stasis_message *

msg

)

Build the Generic event system representation of the message.

May return NULL, to indicate no representation. The returned object should be disposed of via ast_event_destroy.

Parameters

msg	Message to convert to AMI.

Return values

NULL	if AMI format is not supported.

Definition at line 236 of file stasis_message.c.

{
    return INVOKE_VIRTUAL(to_event, msg);
}

References INVOKE_VIRTUAL.

Referenced by publish_to_corosync().

stasis_message_to_json()

struct ast_json * stasis_message_to_json	(	struct stasis_message *	msg,
		struct stasis_message_sanitizer *	sanitize
	)

Build the JSON representation of the message.

May return NULL, to indicate no representation. The returned object should be ast_json_unref()'ed.

Parameters

msg	Message to convert to JSON string.
sanitize	Snapshot sanitization callback.

Returns

Newly allocated string with JSON message.

Return values

NULL	if JSON format is not supported.

Definition at line 229 of file stasis_message.c.

{
    return INVOKE_VIRTUAL(to_json, msg, sanitize);
}

References INVOKE_VIRTUAL.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), and rtcp_message_handler().

stasis_message_type()

struct stasis_message_type * stasis_message_type

(

const struct stasis_message *

msg

)

Get the message type for a stasis_message.

Parameters

msg	Message to type

Returns

Type of msg

Return values

NULL	if msg is NULL.

Since

12

Definition at line 187 of file stasis_message.c.

{
    if (msg == NULL) {
        return NULL;
    }
    return msg->type;
}

References NULL, and stasis_message::type.

stasis_message_type_available_formatters()

enum stasis_subscription_message_formatters stasis_message_type_available_formatters

(

const struct stasis_message_type *

message_type

)

Get a bitmap of available formatters for a message type.

Parameters

message_type

Message type

Returns

A bitmap of stasis_subscription_message_formatters

Since

13.25.0

16.2.0

Definition at line 114 of file stasis_message.c.

{
    return type->available_formatters;
}

References type.

Referenced by dispatch_message().

stasis_message_type_create()

enum stasis_message_type_result stasis_message_type_create	(	const char *	name,
		struct stasis_message_vtable *	vtable,
		struct stasis_message_type **	result
	)

Create a new message type.

stasis_message_type is an AO2 object, so ao2_cleanup() when you're done with it.

Parameters

	name	Name of the new type.
	vtable	Virtual table of message methods. May be NULL.
[out]	result	The location where the new message type will be placed

Note

Stasis message type creation may be declined if the message type is disabled

Returns

A stasis_message_type_result enum

Since

12

Definition at line 56 of file stasis_message.c.

{
    struct stasis_message_type *type;
 
    /* Check for declination */
    if (name && stasis_message_type_declined(name)) {
        return STASIS_MESSAGE_TYPE_DECLINED;
    }
 
    type = ao2_t_alloc_options(sizeof(*type), message_type_dtor,
        AO2_ALLOC_OPT_LOCK_NOLOCK, name ?: "");
    if (!type) {
        return STASIS_MESSAGE_TYPE_ERROR;
    }
    if (!vtable) {
        /* Null object pattern, FTW! */
        vtable = &null_vtable;
    }
 
    type->name = ast_strdup(name);
    if (!type->name) {
        ao2_cleanup(type);
        return STASIS_MESSAGE_TYPE_ERROR;
    }
    type->hash = ast_hashtab_hash_string(name);
    type->vtable = vtable;
    if (vtable->to_json) {
        type->available_formatters |= STASIS_SUBSCRIPTION_FORMATTER_JSON;
    }
    if (vtable->to_ami) {
        type->available_formatters |= STASIS_SUBSCRIPTION_FORMATTER_AMI;
    }
    if (vtable->to_event) {
        type->available_formatters |= STASIS_SUBSCRIPTION_FORMATTER_EVENT;
    }
    type->id = ast_atomic_fetchadd_int(&message_type_id, +1);
    *result = type;
 
    return STASIS_MESSAGE_TYPE_SUCCESS;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_t_alloc_options, ast_atomic_fetchadd_int(), ast_hashtab_hash_string(), ast_strdup, message_type_dtor(), message_type_id, name, null_vtable, result, STASIS_MESSAGE_TYPE_DECLINED, stasis_message_type_declined(), STASIS_MESSAGE_TYPE_ERROR, STASIS_MESSAGE_TYPE_SUCCESS, STASIS_SUBSCRIPTION_FORMATTER_AMI, STASIS_SUBSCRIPTION_FORMATTER_EVENT, STASIS_SUBSCRIPTION_FORMATTER_JSON, stasis_message_vtable::to_ami, stasis_message_vtable::to_event, stasis_message_vtable::to_json, type, and stasis_message_type::vtable.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and create_message_types().

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

unsigned int stasis_message_type_hash

(

const struct stasis_message_type *

type

)

Gets the hash of a given message type.

Parameters

type	The type to get the hash of.

Returns

The hash

Since

13.24.0

Definition at line 104 of file stasis_message.c.

{
    return type->hash;
}

References type.

Referenced by cache_entry_compute_hash().

stasis_message_type_id()

int stasis_message_type_id

(

const struct stasis_message_type *

type

)

Gets the id of a given message type.

Parameters

type	The type to get the id of.

Returns

The id

Since

17.0.0

Definition at line 109 of file stasis_message.c.

{
    return type->id;
}

References type.

Referenced by dispatch_message(), publish_msg(), stasis_subscription_accept_message_type(), stasis_subscription_decline_message_type(), and subscription_invoke().

stasis_message_type_name()

const char * stasis_message_type_name

(

const struct stasis_message_type *

type

)

Gets the name of a given message type.

Parameters

type	The type to get.

Returns

Name of the type.

Return values

NULL	if type is NULL.

Since

12

Definition at line 99 of file stasis_message.c.

{
    return type->name;
}

References type.

Referenced by AST_TEST_DEFINE(), cache_find(), cache_simple(), cache_test_data_id(), caching_topic_exec(), dump_consumer(), print_cache_entry(), send_msg(), stasis_subscription_accept_message_type(), stasis_subscription_decline_message_type(), and statsmaker().

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(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), 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(), 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_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(), sub, and stasis_subscription::topic.

Referenced by AST_TEST_DEFINE(), and stasis_message_router_publish_sync().

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(), 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(), acl_change_stasis_subscribe(), ast_mwi_subscribe_pool(), ast_res_pjsip_initialize_configuration(), AST_TEST_DEFINE(), 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_module(), load_module(), load_module(), load_module(), load_module(), load_module(), 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_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(), 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, stasis_topic::subscribers, and stasis_subscription::topic.

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

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(), acl_change_stasis_subscribe(), ast_mwi_subscribe_pool(), ast_res_pjsip_initialize_configuration(), AST_TEST_DEFINE(), 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_module(), load_module(), load_module(), load_module(), load_module(), load_module(), 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, and stasis_subscription::uniqueid.

Referenced by AST_TEST_DEFINE(), 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_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), ast_test_init(), create_cts(), create_subscriptions(), devstate_init(), load_module(), 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_topic_uniqueid()

const char * stasis_topic_uniqueid

(

const struct stasis_topic *

topic

)

Return the uniqueid of a topic.

Parameters

topic

Topic.

Returns

Uniqueid of the topic.

Return values

NULL	if topic is NULL.

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, stasis_subscription::mailbox, NULL, send_subscription_unsubscribe(), sub, sub_cleanup(), stasis_subscription::topic, and topic_remove_subscription().

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), 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(), 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(), unload_module(), unload_module(), unload_module(), unload_module(), unload_module(), unload_module(), and unload_pbx().