stasis_state.c File Reference

#include "asterisk.h"
#include "asterisk/stasis_state.h"

Include dependency graph for stasis_state.c:

Go to the source code of this file.

Data Structures
struct	stasis_state
struct	stasis_state_manager
struct	stasis_state_proxy
struct	stasis_state_publisher
struct	stasis_state_subscriber

Macros
#define	STATE_BUCKETS   57
#define	state_find_or_add(manager, state_topic, id)   __state_find_or_add(manager, state_topic, id, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Functions
static struct stasis_state *	__state_find_or_add (struct stasis_state_manager *manager, struct stasis_topic *state_topic, const char *id, const char *file, int line, const char *func)
	AO2_STRING_FIELD_CMP_FN (stasis_state_proxy, id)
	AO2_STRING_FIELD_HASH_FN (stasis_state_proxy, id)
static int	handle_stasis_state (struct stasis_state *state, on_stasis_state handler, void *data)
static int	handle_stasis_state_proxy (void *obj, void *arg, void *data, int flags)
static int	handle_stasis_state_subscribed (void *obj, void *arg, void *data, int flags)
static void	publisher_dtor (void *obj)
int	stasis_state_add_observer (struct stasis_state_manager *manager, struct stasis_state_observer *observer)
	Add an observer to receive managed state related events. More...
struct stasis_state_publisher *	stasis_state_add_publisher (struct stasis_state_manager *manager, const char *id)
	Add a publisher to the managed state for the given id. More...
struct stasis_state_subscriber *	stasis_state_add_subscriber (struct stasis_state_manager *manager, const char *id)
	Add a subscriber to the managed stasis state for the given id. More...
struct stasis_topic *	stasis_state_all_topic (struct stasis_state_manager *manager)
	Retrieve the manager's topic (the topic that all state topics get forwarded to) More...
void	stasis_state_callback_all (struct stasis_state_manager *manager, on_stasis_state handler, void *data)
	For each managed state call the given handler. More...
void	stasis_state_callback_subscribed (struct stasis_state_manager *manager, on_stasis_state handler, void *data)
	For each managed, and explicitly subscribed state call the given handler. More...
struct stasis_state_manager *	stasis_state_manager_create (const char *topic_name)
	Create a stasis state manager. More...
void	stasis_state_publish (struct stasis_state_publisher *pub, struct stasis_message *msg)
	Publish to a managed state (topic) using a publisher. More...
void	stasis_state_publish_by_id (struct stasis_state_manager *manager, const char *id, const struct ast_eid *eid, struct stasis_message *msg)
	Publish to a managed named by id topic, and add an implicit subscriber. More...
const char *	stasis_state_publisher_id (const struct stasis_state_publisher *pub)
	Retrieve the publisher's underlying state's unique id. More...
struct stasis_topic *	stasis_state_publisher_topic (struct stasis_state_publisher *pub)
	Retrieve the publisher's topic. More...
void	stasis_state_remove_observer (struct stasis_state_manager *manager, struct stasis_state_observer *observer)
	Remove an observer (will no longer receive managed state related events). More...
void	stasis_state_remove_publish_by_id (struct stasis_state_manager *manager, const char *id, const struct ast_eid *eid, struct stasis_message *msg)
	Publish to a managed named by id topic, and remove an implicit publisher. More...
struct stasis_state_subscriber *	stasis_state_subscribe_pool (struct stasis_state_manager *manager, const char *id, stasis_subscription_cb callback, void *data)
	Add a subscriber, and subscribe to its underlying stasis topic. More...
void *	stasis_state_subscriber_data (struct stasis_state_subscriber *sub)
	Retrieve the last known state stasis message payload for the subscriber. More...
const char *	stasis_state_subscriber_id (const struct stasis_state_subscriber *sub)
	Retrieve the underlying subscribed to state's unique id. More...
struct stasis_subscription *	stasis_state_subscriber_subscription (struct stasis_state_subscriber *sub)
	Retrieve the stasis topic subscription if available. More...
struct stasis_topic *	stasis_state_subscriber_topic (struct stasis_state_subscriber *sub)
	Retrieve the subscriber's topic. More...
struct stasis_topic *	stasis_state_topic (struct stasis_state_manager *manager, const char *id)
	Retrieve a managed topic creating one if not currently managed. More...
void *	stasis_state_unsubscribe (struct stasis_state_subscriber *sub)
	Unsubscribe from the stasis topic and stasis state. More...
void *	stasis_state_unsubscribe_and_join (struct stasis_state_subscriber *sub)
	Unsubscribe from the stasis topic, block until the final message is received, and then unsubscribe from stasis state. More...
static struct stasis_state *	state_alloc (struct stasis_state_manager *manager, struct stasis_topic *state_topic, const char *id, const char *file, int line, const char *func)
static void	state_dtor (void *obj)
static void	state_find_and_remove_eid (struct stasis_state *state, const struct ast_eid *eid)
static void	state_find_or_add_eid (struct stasis_state *state, const struct ast_eid *eid)
static const char *	state_id_by_topic (struct stasis_topic *manager_topic, const struct stasis_topic *state_topic)
static void	state_manager_dtor (void *obj)
static void	state_proxy_dtor (void *obj)
static void	state_proxy_sub_cb (void *obj, void *data)
static void	subscriber_dtor (void *obj)

Macro Definition Documentation

STATE_BUCKETS

#define STATE_BUCKETS   57

The number of buckets to use for managed states

Definition at line 77 of file stasis_state.c.

state_find_or_add

#define state_find_or_add	(		manager,
			state_topic,
			id
	)		__state_find_or_add(manager, state_topic, id, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 271 of file stasis_state.c.

Function Documentation

__state_find_or_add()

static struct stasis_state * __state_find_or_add ( struct stasis_state_manager *  manager, struct stasis_topic *  state_topic, const char *  id, const char *  file, int  line, const char *  func  )

static

Definition at line 272 of file stasis_state.c.

{
    struct stasis_state *state;
 
    ao2_lock(manager->states);
    if (ast_strlen_zero(id)) {
        id = state_id_by_topic(manager->all_topic, state_topic);
    }
 
    state = ao2_weakproxy_find(manager->states, id, OBJ_SEARCH_KEY | OBJ_NOLOCK, "");
    if (!state) {
        state = state_alloc(manager, state_topic, id, file, line, func);
    }
 
    ao2_unlock(manager->states);
 
    return state;
}

References stasis_state_manager::all_topic, ao2_lock, ao2_unlock, ao2_weakproxy_find, ast_strlen_zero(), make_ari_stubs::file, stasis_state::manager, OBJ_NOLOCK, OBJ_SEARCH_KEY, state, state_alloc(), state_id_by_topic(), and stasis_state_manager::states.

AO2_STRING_FIELD_CMP_FN()

AO2_STRING_FIELD_CMP_FN	(	stasis_state_proxy	,
		id
	)

AO2_STRING_FIELD_HASH_FN()

AO2_STRING_FIELD_HASH_FN	(	stasis_state_proxy	,
		id
	)

handle_stasis_state()

static int handle_stasis_state ( struct stasis_state *  state, on_stasis_state  handler, void *  data  )

static

Definition at line 709 of file stasis_state.c.

{
    struct stasis_message *msg;
    int res;
 
    /*
     * State needs to be locked here while we retrieve and bump the reference on its message
     * object. Doing so guarantees the message object will live throughout its handling.
     */
    ao2_lock(state);
    msg = ao2_bump(state->msg);
    ao2_unlock(state);
 
    res = handler(state->id, msg, data);
    ao2_cleanup(msg);
    return res;
}

References ao2_bump, ao2_cleanup, ao2_lock, ao2_unlock, stasis_message::data, and handler().

Referenced by handle_stasis_state_proxy(), and handle_stasis_state_subscribed().

handle_stasis_state_proxy()

static int handle_stasis_state_proxy ( void *  obj, void *  arg, void *  data, int  flags  )

static

Definition at line 727 of file stasis_state.c.

{
    struct stasis_state *state = ao2_weakproxy_get_object(obj, 0);
 
    if (state) {
        int res;
        res = handle_stasis_state(state, arg, data);
        ao2_ref(state, -1);
        return res;
    }
 
    return 0;
}

References ao2_ref, ao2_weakproxy_get_object, and handle_stasis_state().

Referenced by stasis_state_callback_all().

handle_stasis_state_subscribed()

static int handle_stasis_state_subscribed ( void *  obj, void *  arg, void *  data, int  flags  )

static

Definition at line 750 of file stasis_state.c.

{
    struct stasis_state *state = ao2_weakproxy_get_object(obj, 0);
    int res = 0;
 
    if (state && state->num_subscribers) {
        res = handle_stasis_state(state, arg, data);
    }
 
    ao2_cleanup(state);
 
    return res;
}

References ao2_cleanup, ao2_weakproxy_get_object, and handle_stasis_state().

Referenced by stasis_state_callback_subscribed().

publisher_dtor()

static void publisher_dtor ( void *  obj )

static

Definition at line 525 of file stasis_state.c.

{
    struct stasis_state_publisher *pub = obj;
 
    ao2_ref(pub->state, -1);
}

References ao2_ref, and stasis_state_publisher::state.

Referenced by stasis_state_add_publisher().

stasis_state_add_observer()

int stasis_state_add_observer	(	struct stasis_state_manager *	manager,
		struct stasis_state_observer *	observer
	)

Add an observer to receive managed state related events.

Parameters

manager	The state manager
observer	The observer handling events

Return values

0	if successfully registered
-1	on failure

Since

13.28.0

16.5.0

Definition at line 689 of file stasis_state.c.

{
    int res;
 
    AST_VECTOR_RW_WRLOCK(&manager->observers);
    res = AST_VECTOR_APPEND(&manager->observers, observer);
    AST_VECTOR_RW_UNLOCK(&manager->observers);
 
    return res;
}

References AST_VECTOR_APPEND, AST_VECTOR_RW_UNLOCK, AST_VECTOR_RW_WRLOCK, stasis_state::manager, and observer.

Referenced by ast_mwi_add_observer(), and subscriptions_create().

stasis_state_add_publisher()

struct stasis_state_publisher * stasis_state_add_publisher	(	struct stasis_state_manager *	manager,
		const char *	id
	)

Add a publisher to the managed state for the given id.

Adds a publisher to a managed state based on id. If managed state does not already exists for the given id then new managed state is created. Otherwise the existing state is used.

Parameters

manager	The manager object
id	The unique id of a managed state

Return values

A	stasis state publisher
NULL	if an error occurred

Since

13.28.0

16.5.0

Definition at line 532 of file stasis_state.c.

{
    struct stasis_state_publisher *pub = ao2_alloc_options(
        sizeof(*pub), publisher_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
 
    if (!pub) {
        ast_log(LOG_ERROR, "Unable to create publisher to %s/%s\n",
            stasis_topic_name(manager->all_topic), id);
        return NULL;
    }
 
    pub->state = state_find_or_add(manager, NULL, id);
    if (!pub->state) {
        ao2_ref(pub, -1);
        return NULL;
    }
 
    return pub;
}

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_ref, ast_log, LOG_ERROR, NULL, publisher_dtor(), stasis_topic_name(), stasis_state_publisher::state, and state_find_or_add.

Referenced by ast_mwi_add_publisher(), and publishers_create().

stasis_state_add_subscriber()

struct stasis_state_subscriber * stasis_state_add_subscriber	(	struct stasis_state_manager *	manager,
		const char *	id
	)

Add a subscriber to the managed stasis state for the given id.

Adds a subscriber to a managed state based on id. If managed state does not already exists for the given id then new managed state is created. Otherwise the existing state is subscribed to.

Parameters

manager	The manager object
id	The unique id of a managed state

Return values

A	stasis state subscriber
NULL	if an error occurred

Since

13.28.0

16.5.0

Definition at line 413 of file stasis_state.c.

{
    size_t i;
    struct stasis_state_subscriber *sub = ao2_alloc_options(
        sizeof(*sub), subscriber_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
 
    if (!sub) {
        ast_log(LOG_ERROR, "Unable to create subscriber to %s/%s\n",
            stasis_topic_name(manager->all_topic), id);
        return NULL;
    }
 
    sub->state = state_find_or_add(manager, NULL, id);
    if (!sub->state) {
        ao2_ref(sub, -1);
        return NULL;
    }
 
    ao2_lock(sub->state);
    ++sub->state->num_subscribers;
    ao2_unlock(sub->state);
 
    AST_VECTOR_RW_RDLOCK(&manager->observers);
    for (i = 0; i < AST_VECTOR_SIZE(&manager->observers); ++i) {
        if (AST_VECTOR_GET(&manager->observers, i)->on_subscribe) {
            AST_VECTOR_GET(&manager->observers, i)->on_subscribe(id, sub);
        }
    }
    AST_VECTOR_RW_UNLOCK(&manager->observers);
 
    return sub;
}

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_lock, ao2_ref, ao2_unlock, ast_log, AST_VECTOR_GET, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, LOG_ERROR, NULL, stasis_topic_name(), state_find_or_add, sub, and subscriber_dtor().

Referenced by ast_mwi_add_subscriber(), and stasis_state_subscribe_pool().

stasis_state_all_topic()

struct stasis_topic * stasis_state_all_topic

(

struct stasis_state_manager *

manager

)

Retrieve the manager's topic (the topic that all state topics get forwarded to)

Parameters

manager

The manager object

Return values

The	manager's topic.

Since

13.28.0

16.5.0

Definition at line 365 of file stasis_state.c.

{
    return manager->all_topic;
}

References stasis_state_manager::all_topic.

Referenced by ast_mwi_topic_all().

stasis_state_callback_all()

void stasis_state_callback_all	(	struct stasis_state_manager *	manager,
		on_stasis_state	handler,
		void *	data
	)

For each managed state call the given handler.

Parameters

manager	The state manager
handler	The handler to call for each managed state
data	User to data to pass on to the handler

Since

13.28.0

16.5.0

Definition at line 741 of file stasis_state.c.

{
    ast_assert(handler != NULL);
 
    ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
        handle_stasis_state_proxy, handler, data);
}

References ao2_callback_data, ast_assert, handle_stasis_state_proxy(), handler(), stasis_state::manager, NULL, OBJ_MULTIPLE, OBJ_NODATA, and stasis_state_manager::states.

Referenced by ast_mwi_state_callback_all(), and publish().

stasis_state_callback_subscribed()

void stasis_state_callback_subscribed	(	struct stasis_state_manager *	manager,
		on_stasis_state	handler,
		void *	data
	)

For each managed, and explicitly subscribed state call the given handler.

Parameters

manager	The state manager
handler	The handler to call for each managed state
data	User to data to pass on to the handler

Since

13.28.0

16.5.0

Definition at line 764 of file stasis_state.c.

{
    ast_assert(handler != NULL);
 
    ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
        handle_stasis_state_subscribed, handler, data);
}

References ao2_callback_data, ast_assert, handle_stasis_state_subscribed(), handler(), stasis_state::manager, NULL, OBJ_MULTIPLE, OBJ_NODATA, and stasis_state_manager::states.

Referenced by ast_mwi_state_callback_subscribed().

stasis_state_manager_create()

struct stasis_state_manager * stasis_state_manager_create

(

const char *

topic_name

)

Create a stasis state manager.

Note

The state manager is an ao2_object. When done simply decrement its reference for object cleanup.

Parameters

topic_name

The name of the topic to create that all state topics get forwarded to

Return values

A	stasis state manager
NULL	if an error occurred

Since

13.28.0

16.5.0

Definition at line 325 of file stasis_state.c.

{
    struct stasis_state_manager *manager;
 
    manager = ao2_alloc_options(sizeof(*manager), state_manager_dtor,
        AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!manager) {
        return NULL;
    }
 
    manager->states = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
        STATE_BUCKETS, stasis_state_proxy_hash_fn, NULL, stasis_state_proxy_cmp_fn);
    if (!manager->states) {
        ao2_ref(manager, -1);
        return NULL;
    }
 
    manager->all_topic = stasis_topic_create(topic_name);
    if (!manager->all_topic) {
        ao2_ref(manager, -1);
        return NULL;
    }
 
    if (AST_VECTOR_RW_INIT(&manager->observers, 2) != 0) {
        ao2_ref(manager, -1);
        return NULL;
    }
 
#ifdef AO2_DEBUG
    {
        char *container_name =
            ast_alloca(strlen(stasis_topic_name(manager->all_topic)) + strlen("-manager") + 1);
        sprintf(container_name, "%s-manager", stasis_topic_name(manager->all_topic));
        ao2_container_register(container_name, manager->states, state_prnt_obj);
    }
#endif
 
    return manager;
}

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_MUTEX, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_container_alloc_hash, ao2_container_register(), ao2_ref, ast_alloca, AST_VECTOR_RW_INIT, NULL, stasis_topic_create(), stasis_topic_name(), STATE_BUCKETS, state_manager_dtor(), and stasis_state_manager::states.

Referenced by AST_TEST_DEFINE(), and mwi_init().

stasis_state_publish()

void stasis_state_publish	(	struct stasis_state_publisher *	pub,
		struct stasis_message *	msg
	)

Publish to a managed state (topic) using a publisher.

Parameters

pub	The publisher to use to publish the message
msg	The message to publish

Since

13.28.0

16.5.0

Definition at line 563 of file stasis_state.c.

{
    ao2_lock(pub->state);
    ao2_replace(pub->state->msg, msg);
    ao2_unlock(pub->state);
 
    stasis_publish(pub->state->topic, msg);
}

References ao2_lock, ao2_replace, ao2_unlock, stasis_state::msg, stasis_publish(), stasis_state_publisher::state, and stasis_state::topic.

Referenced by ast_mwi_publish(), and explicit_publish_cb().

stasis_state_publish_by_id()

void stasis_state_publish_by_id	(	struct stasis_state_manager *	manager,
		const char *	id,
		const struct ast_eid *	eid,
		struct stasis_message *	msg
	)

Publish to a managed named by id topic, and add an implicit subscriber.

Note

It is recommended when adding new publisher functionality within a module to create and use an explicit publisher instead of using this method.

This creates an implicit publisher keyed off the eid. This ability was mainly implemented in order to maintain compatibility with already established code. Allowing the creation of an implicit publisher made is so less changes were required when stasis state module was initially added.

There should only ever be one publisher for a specifically named managed topic within the system. This being the case we can use the eid to implicitly track the publisher. However once publishing is no longer needed for a topic a call to stasis_state_remove_publish_by_id is required in order to remove the implicit publisher. Thus allowing for its eventual destruction. Without the call to remove a memory leak will occur.

Parameters

manager	The state manager
id	A state's unique id
eid	The unique system id
msg	The message to publish

Since

13.28.0

16.5.0

Definition at line 639 of file stasis_state.c.

{
    struct stasis_state *state;
 
    state = state_find_or_add(manager, NULL, id);
    if (!state) {
        return;
    }
 
    ao2_lock(state);
    state_find_or_add_eid(state, eid);
    ao2_replace(state->msg, msg);
    ao2_unlock(state);
 
    stasis_publish(state->topic, msg);
 
    ao2_ref(state, -1);
}

References ao2_lock, ao2_ref, ao2_replace, ao2_unlock, stasis_state::manager, stasis_state::msg, NULL, stasis_publish(), state, state_find_or_add, and state_find_or_add_eid().

Referenced by ast_mwi_publish_by_mailbox(), and implicit_publish_cb().

stasis_state_publisher_id()

const char * stasis_state_publisher_id

(

const struct stasis_state_publisher *

pub

)

Retrieve the publisher's underlying state's unique id.

Parameters

pub	A stasis state publisher

Return values

The	managed state's id

Since

13.28.0

16.5.0

Definition at line 553 of file stasis_state.c.

{
    return pub->state->id;
}

References stasis_state::id, and stasis_state_publisher::state.

Referenced by ast_mwi_publish(), and explicit_publish_cb().

stasis_state_publisher_topic()

struct stasis_topic * stasis_state_publisher_topic

(

struct stasis_state_publisher *

pub

)

Retrieve the publisher's topic.

Note

Returned topic's reference count is NOT incremented. However, the topic is guaranteed to live for the lifetime of the publisher.

Parameters

pub	A stasis state publisher

Return values

The	publisher's topic

Since

13.28.0

16.5.0

Definition at line 558 of file stasis_state.c.

{
    return pub->state->topic;
}

References stasis_state_publisher::state, and stasis_state::topic.

stasis_state_remove_observer()

void stasis_state_remove_observer	(	struct stasis_state_manager *	manager,
		struct stasis_state_observer *	observer
	)

Remove an observer (will no longer receive managed state related events).

Parameters

manager	The state manager
observer	The observer being removed

Since

13.28.0

16.5.0

Definition at line 701 of file stasis_state.c.

{
    AST_VECTOR_RW_WRLOCK(&manager->observers);
    AST_VECTOR_REMOVE_ELEM_UNORDERED(&manager->observers, observer, AST_VECTOR_ELEM_CLEANUP_NOOP);
    AST_VECTOR_RW_UNLOCK(&manager->observers);
}

References AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_REMOVE_ELEM_UNORDERED, AST_VECTOR_RW_UNLOCK, AST_VECTOR_RW_WRLOCK, stasis_state::manager, and observer.

Referenced by ast_mwi_remove_observer(), and subscriptions_destroy().

stasis_state_remove_publish_by_id()

void stasis_state_remove_publish_by_id	(	struct stasis_state_manager *	manager,
		const char *	id,
		const struct ast_eid *	eid,
		struct stasis_message *	msg
	)

Publish to a managed named by id topic, and remove an implicit publisher.

This function should be called after calling stasis_state_publish_by_id at least once for the same manager, id, and eid. If the given stasis message is NULL then the implicit publisher is removed, but no last message is published.

See note and description on stasis_state_publish_by_id for more details about if, and when this function should be used.

Parameters

manager	The state manager
id	A state's unique id
eid	The unique system id
msg	The message to publish (can be NULL)

Since

13.28.0

16.5.0

Definition at line 659 of file stasis_state.c.

{
    struct stasis_state *state = ao2_weakproxy_find(manager->states, id, OBJ_SEARCH_KEY, "");
 
    if (!state) {
        /*
         * In most circumstances state should already exist here. However, if there is no
         * state then it can mean one of a few things:
         *
         * 1. This function was called prior to an implicit publish for the same given
         *    manager, and id.
         * 2. This function was called more than once for the same manager, and id.
         * 3. There is ref count problem with the explicit subscribers, and publishers.
         */
        ast_debug(5, "Attempted to remove state for id '%s', but state not found\n", id);
        return;
    }
 
    if (msg) {
        stasis_publish(state->topic, msg);
    }
 
    ao2_lock(state);
    state_find_and_remove_eid(state, eid);
    ao2_unlock(state);
 
    ao2_ref(state, -1);
}

References ao2_lock, ao2_ref, ao2_unlock, ao2_weakproxy_find, ast_debug, stasis_state::manager, stasis_state::msg, OBJ_SEARCH_KEY, stasis_publish(), state_find_and_remove_eid(), and stasis_state_manager::states.

Referenced by ast_delete_mwi_state_full(), and publishers_destroy().

stasis_state_subscribe_pool()

struct stasis_state_subscriber * stasis_state_subscribe_pool	(	struct stasis_state_manager *	manager,
		const char *	id,
		stasis_subscription_cb	callback,
		void *	data
	)

Add a subscriber, and subscribe to its underlying stasis topic.

Adds a subscriber to a managed state based on id. If managed state does not already exists for the given id then new managed state is created. Otherwise the existing state is subscribed to. If the state is successfully subscribed to then a stasis subscription is subsequently created as well.

Parameters

manager	The manager object
id	The unique id of a managed state
callback	The stasis subscription callback
data	A user data object passed to the stasis subscription

Return values

A	stasis state subscriber
NULL	if an error occurred

Since

13.28.0

16.5.0

Definition at line 447 of file stasis_state.c.

{
    struct stasis_topic *topic;
    struct stasis_state_subscriber *sub = stasis_state_add_subscriber(manager, id);
 
    if (!sub) {
        return NULL;
    }
 
    topic = sub->state->topic;
    ast_debug(3, "Creating stasis state subscription to id '%s'. Topic: '%s':%p %d\n",
        id, stasis_topic_name(topic), topic, (int)ao2_ref(topic, 0));
 
    sub->stasis_sub = stasis_subscribe_pool(topic, callback, data);
 
    if (!sub->stasis_sub) {
        ao2_ref(sub, -1);
        return NULL;
    }
 
    return sub;
}

References ao2_ref, ast_debug, NULL, stasis_state_add_subscriber(), stasis_subscribe_pool, stasis_topic_name(), and sub.

Referenced by ast_mwi_subscribe_pool(), and subscriptions_create().

stasis_state_subscriber_data()

void * stasis_state_subscriber_data

(

struct stasis_state_subscriber *

sub

)

Retrieve the last known state stasis message payload for the subscriber.

If a stasis message has been published to this state, this function returns that message's payload object. If no stasis message has been published on the state, or the message's payload does not exist then NULL is returned.

Note

Returned data's reference count is incremented

Parameters

sub	A stasis state subscriber

Return values

The	subscriber's state message data
NULL	if no data has been published yet

Since

13.28.0

16.5.0

Definition at line 498 of file stasis_state.c.

{
    void *res;
 
    /*
     * The data's reference needs to be bumped before returning so it doesn't disappear
     * for the caller. Lock state, so the underlying message data is not replaced while
     * retrieving.
     */
    ao2_lock(sub->state);
    res = ao2_bump(stasis_message_data(sub->state->msg));
    ao2_unlock(sub->state);
 
    return res;
}

References ao2_bump, ao2_lock, ao2_unlock, stasis_message_data(), and sub.

Referenced by ast_mwi_subscriber_data(), and handle_validate().

stasis_state_subscriber_id()

const char * stasis_state_subscriber_id

(

const struct stasis_state_subscriber *

sub

)

Retrieve the underlying subscribed to state's unique id.

Parameters

sub	A stasis state subscriber

Return values

The	managed state's id

Since

13.28.0

16.5.0

Definition at line 488 of file stasis_state.c.

{
    return sub->state->id;
}

References sub.

Referenced by ast_mwi_subscriber_data().

stasis_state_subscriber_subscription()

struct stasis_subscription * stasis_state_subscriber_subscription

(

struct stasis_state_subscriber *

sub

)

Retrieve the stasis topic subscription if available.

Parameters

sub	A stasis state subscriber

Return values

The	subscriber's stasis subscription
NULL	if no subscription available

Since

13.28.0

16.5.0

Definition at line 514 of file stasis_state.c.

{
    return sub->stasis_sub;
}

References sub.

Referenced by ast_mwi_subscriber_subscription().

stasis_state_subscriber_topic()

struct stasis_topic * stasis_state_subscriber_topic

(

struct stasis_state_subscriber *

sub

)

Retrieve the subscriber's topic.

Note

Returned topic's reference count is NOT incremented. However, the topic is guaranteed to live for the lifetime of the subscriber.

Parameters

sub	A stasis state subscriber

Return values

The	subscriber's topic

Since

13.28.0

16.5.0

Definition at line 493 of file stasis_state.c.

{
    return sub->state->topic;
}

References sub.

Referenced by ast_mwi_subscriber_topic().

stasis_state_topic()

struct stasis_topic * stasis_state_topic	(	struct stasis_state_manager *	manager,
		const char *	id
	)

Retrieve a managed topic creating one if not currently managed.

WARNING This function should not be called before adding a publisher or subscriber or it will cause a memory leak within the stasis state manager. This function is here in order to allow for compatibility with how things used to work.

Also much like the similar functionality from before it returns the stasis topic, but does not bump its reference.

Parameters

manager	The manager object
id	The unique id of/for the topic

Return values

A	managed stasis topic.
NULL	if an error occurred

Since

13.28.0

16.5.0

Definition at line 370 of file stasis_state.c.

{
    struct stasis_topic *topic;
    struct stasis_state *state;
 
    state = state_find_or_add(manager, NULL, id);
    if (!state) {
        return NULL;
    }
 
    topic = state->topic;
    ao2_ref(state, -1);
    return topic;
}

References ao2_ref, stasis_state::manager, NULL, state, state_find_or_add, and stasis_state::topic.

Referenced by ast_mwi_topic().

stasis_state_unsubscribe()

void * stasis_state_unsubscribe

(

struct stasis_state_subscriber *

sub

)

Unsubscribe from the stasis topic and stasis state.

Parameters

sub	A stasis state subscriber

Return values

NULL

Since

13.28.0

16.5.0

Definition at line 471 of file stasis_state.c.

{
    sub->stasis_sub = stasis_unsubscribe(sub->stasis_sub);
    ao2_ref(sub, -1);
    return NULL;
}

References ao2_ref, NULL, stasis_unsubscribe(), and sub.

Referenced by ast_mwi_unsubscribe().

stasis_state_unsubscribe_and_join()

void * stasis_state_unsubscribe_and_join

(

struct stasis_state_subscriber *

sub

)

Unsubscribe from the stasis topic, block until the final message is received, and then unsubscribe from stasis state.

Parameters

sub	A stasis state subscriber

Return values

NULL

Since

13.28.0

16.5.0

Definition at line 478 of file stasis_state.c.

{
    if (sub) {
        sub->stasis_sub = stasis_unsubscribe_and_join(sub->stasis_sub);
        ao2_ref(sub, -1);
    }
 
    return NULL;
}

References ao2_ref, NULL, stasis_unsubscribe_and_join(), and sub.

Referenced by ast_mwi_unsubscribe_and_join(), and subscriptions_destroy().

state_alloc()

static struct stasis_state * state_alloc ( struct stasis_state_manager *  manager, struct stasis_topic *  state_topic, const char *  id, const char *  file, int  line, const char *  func  )

static

Definition at line 168 of file stasis_state.c.

{
    struct stasis_state_proxy *proxy = NULL;
    struct stasis_state *state = NULL;
 
    if (!id) {
        /* If not given an id, then a state topic is required */
        ast_assert(state_topic != NULL);
 
        /* Get the id we'll key off of from the state topic */
        id = state_id_by_topic(manager->all_topic, state_topic);
    }
 
    state = __ao2_alloc(sizeof(*state), state_dtor, AO2_ALLOC_OPT_LOCK_MUTEX, id, file, line, func);
    if (!state) {
        goto error_return;
    }
 
    if (!state_topic) {
        char *name;
 
        /*
         * To provide further detail and to ensure that the topic is unique within the
         * scope of the system we prefix it with the manager's topic name, which should
         * itself already be unique.
         */
        if (ast_asprintf(&name, "%s/%s", stasis_topic_name(manager->all_topic), id) < 0) {
            goto error_return;
        }
 
        state->topic = stasis_topic_create(name);
 
        ast_free(name);
        if (!state->topic) {
            goto error_return;
        }
    } else {
        /*
         * Since the state topic was passed in, go ahead and bump its reference.
         * By doing this here first, it allows us to consistently decrease the reference on
         * state allocation error.
         */
        ao2_ref(state_topic, +1);
        state->topic = state_topic;
    }
 
    proxy = ao2_t_weakproxy_alloc(sizeof(*proxy) + strlen(id) + 1, state_proxy_dtor, id);
    if (!proxy) {
        goto error_return;
    }
 
    strcpy(proxy->id, id); /* Safe */
 
    state->id = proxy->id;
    proxy->manager = ao2_bump(manager);
    state->manager = proxy->manager; /* state->manager is owned by the proxy */
 
    state->forward = stasis_forward_all(state->topic, manager->all_topic);
    if (!state->forward) {
        goto error_return;
    }
 
    if (AST_VECTOR_INIT(&state->eids, 2)) {
        goto error_return;
    }
 
    if (ao2_t_weakproxy_set_object(proxy, state, OBJ_NOLOCK, "weakproxy link")) {
        goto error_return;
    }
 
    if (ao2_weakproxy_subscribe(proxy, state_proxy_sub_cb, NULL, OBJ_NOLOCK)) {
        goto error_return;
    }
 
    if (!ao2_link_flags(manager->states, proxy, OBJ_NOLOCK)) {
        goto error_return;
    }
 
    ao2_ref(proxy, -1);
 
    return state;
 
error_return:
    ast_log(LOG_ERROR, "Unable to allocate state '%s' in manager '%s'\n",
            id, stasis_topic_name(manager->all_topic));
    ao2_cleanup(state);
    ao2_cleanup(proxy);
    return NULL;
}

References __ao2_alloc(), stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_MUTEX, ao2_bump, ao2_cleanup, ao2_link_flags, ao2_ref, ao2_t_weakproxy_alloc, ao2_t_weakproxy_set_object, ao2_weakproxy_subscribe(), ast_asprintf, ast_assert, ast_free, ast_log, AST_VECTOR_INIT, make_ari_stubs::file, stasis_state_proxy::id, LOG_ERROR, stasis_state_proxy::manager, stasis_state::manager, name, NULL, OBJ_NOLOCK, stasis_forward_all(), stasis_topic_create(), stasis_topic_name(), state, state_dtor(), state_id_by_topic(), state_proxy_dtor(), state_proxy_sub_cb(), and stasis_state_manager::states.

Referenced by __state_find_or_add().

state_dtor()

static void state_dtor ( void *  obj )

static

Definition at line 121 of file stasis_state.c.

{
    struct stasis_state *state = obj;
 
    state->forward = stasis_forward_cancel(state->forward);
    ao2_cleanup(state->topic);
    state->topic = NULL;
    ao2_cleanup(state->msg);
    state->msg = NULL;
 
    /* All eids should have been removed */
    ast_assert(AST_VECTOR_SIZE(&state->eids) == 0);
    AST_VECTOR_FREE(&state->eids);
}

References ao2_cleanup, ast_assert, AST_VECTOR_FREE, AST_VECTOR_SIZE, NULL, and stasis_forward_cancel().

Referenced by state_alloc().

state_find_and_remove_eid()

static void state_find_and_remove_eid ( struct stasis_state *  state, const struct ast_eid *  eid  )

static

Definition at line 621 of file stasis_state.c.

{
    size_t i;
 
    if (!eid) {
        eid = &ast_eid_default;
    }
 
    for (i = 0; i < AST_VECTOR_SIZE(&state->eids); ++i) {
        if (!ast_eid_cmp(AST_VECTOR_GET_ADDR(&state->eids, i), eid)) {
            AST_VECTOR_REMOVE_UNORDERED(&state->eids, i);
            /* Balance the reference from state_find_or_add_eid */
            ao2_ref(state, -1);
            return;
        }
    }
}

References ao2_ref, ast_eid_cmp(), ast_eid_default, AST_VECTOR_GET_ADDR, AST_VECTOR_REMOVE_UNORDERED, and AST_VECTOR_SIZE.

Referenced by stasis_state_remove_publish_by_id().

state_find_or_add_eid()

static void state_find_or_add_eid ( struct stasis_state *  state, const struct ast_eid *  eid  )

static

Definition at line 587 of file stasis_state.c.

{
    size_t i;
 
    if (!eid) {
        eid = &ast_eid_default;
    }
 
    for (i = 0; i < AST_VECTOR_SIZE(&state->eids); ++i) {
        if (!ast_eid_cmp(AST_VECTOR_GET_ADDR(&state->eids, i), eid)) {
            break;
        }
    }
 
    if (i == AST_VECTOR_SIZE(&state->eids)) {
        if (!AST_VECTOR_APPEND(&state->eids, *eid)) {
            /* This ensures state cannot be freed if it has any eids */
            ao2_ref(state, +1);
        }
    }
}

References ao2_ref, ast_eid_cmp(), ast_eid_default, AST_VECTOR_APPEND, AST_VECTOR_GET_ADDR, and AST_VECTOR_SIZE.

Referenced by stasis_state_publish_by_id().

state_id_by_topic()

static const char * state_id_by_topic ( struct stasis_topic *  manager_topic, const struct stasis_topic *  state_topic  )

static

Definition at line 104 of file stasis_state.c.

{
    const char *id;
 
    /* This topic should always belong to the manager */
    ast_assert(ast_begins_with(stasis_topic_name(manager_topic),
        stasis_topic_name(state_topic)));
 
    id = strchr(stasis_topic_name(state_topic), '/');
 
    /* The state's unique id should always exist */
    ast_assert(id != NULL && *(id + 1) != '\0');
 
    return (id + 1);
}

References ast_assert, ast_begins_with(), id, manager_topic, NULL, and stasis_topic_name().

Referenced by __state_find_or_add(), and state_alloc().

state_manager_dtor()

static void state_manager_dtor ( void *  obj )

static

Definition at line 293 of file stasis_state.c.

{
    struct stasis_state_manager *manager = obj;
 
#ifdef AO2_DEBUG
    {
        char *container_name =
            ast_alloca(strlen(stasis_topic_name(manager->all_topic)) + strlen("-manager") + 1);
        sprintf(container_name, "%s-manager", stasis_topic_name(manager->all_topic));
        ao2_container_unregister(container_name);
    }
#endif
 
    ao2_cleanup(manager->states);
    manager->states = NULL;
    ao2_cleanup(manager->all_topic);
    manager->all_topic = NULL;
    AST_VECTOR_RW_FREE(&manager->observers);
}

References stasis_state_manager::all_topic, ao2_cleanup, ao2_container_unregister(), ast_alloca, AST_VECTOR_RW_FREE, NULL, stasis_topic_name(), and stasis_state_manager::states.

Referenced by stasis_state_manager_create().

state_proxy_dtor()

static void state_proxy_dtor ( void *  obj )

static

Definition at line 136 of file stasis_state.c.

                                        {
    struct stasis_state_proxy *proxy = obj;
 
    ao2_cleanup(proxy->manager);
}

References ao2_cleanup, and stasis_state_proxy::manager.

Referenced by state_alloc().

state_proxy_sub_cb()

static void state_proxy_sub_cb ( void *  obj, void *  data  )

static

Definition at line 142 of file stasis_state.c.

{
    struct stasis_state_proxy *proxy = obj;
 
    ao2_unlink(proxy->manager->states, proxy);
}

References ao2_unlink, stasis_state_proxy::manager, and stasis_state_manager::states.

Referenced by state_alloc().

subscriber_dtor()

static void subscriber_dtor ( void *  obj )

static

Definition at line 392 of file stasis_state.c.

{
    size_t i;
    struct stasis_state_subscriber *sub = obj;
    struct stasis_state_manager *manager = sub->state->manager;
 
    AST_VECTOR_RW_RDLOCK(&manager->observers);
    for (i = 0; i < AST_VECTOR_SIZE(&manager->observers); ++i) {
        if (AST_VECTOR_GET(&manager->observers, i)->on_unsubscribe) {
            AST_VECTOR_GET(&manager->observers, i)->on_unsubscribe(sub->state->id, sub);
        }
    }
    AST_VECTOR_RW_UNLOCK(&manager->observers);
 
    ao2_lock(sub->state);
    --sub->state->num_subscribers;
    ao2_unlock(sub->state);
 
    ao2_ref(sub->state, -1);
}

References ao2_lock, ao2_ref, ao2_unlock, AST_VECTOR_GET, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, and sub.

Referenced by stasis_state_add_subscriber().