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(mgr, top, id)   __state_find_or_add(mgr, top, 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.

Referenced by stasis_state_manager_create().

state_find_or_add

#define state_find_or_add	(		mgr,
			top,
			id
	)		__state_find_or_add(mgr, top, id, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 270 of file stasis_state.c.

Referenced by stasis_state_add_publisher(), stasis_state_add_subscriber(), stasis_state_publish_by_id(), and stasis_state_topic().

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 271 of file stasis_state.c.

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

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

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 708 of file stasis_state.c.

References ao2_bump, ao2_cleanup, ao2_lock, ao2_unlock, handler(), and stasis_state::msg.

Referenced by handle_stasis_state_proxy(), and handle_stasis_state_subscribed().

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

handle_stasis_state_proxy()

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

static

Definition at line 726 of file stasis_state.c.

References ao2_ref, ao2_weakproxy_get_object, and handle_stasis_state().

Referenced by stasis_state_callback_all().

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

handle_stasis_state_subscribed()

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

static

Definition at line 749 of file stasis_state.c.

References ao2_cleanup, ao2_weakproxy_get_object, handle_stasis_state(), and stasis_state::num_subscribers.

Referenced by stasis_state_callback_subscribed().

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

publisher_dtor()

static void publisher_dtor ( void *  obj )

static

Definition at line 524 of file stasis_state.c.

References ao2_ref, and stasis_state_publisher::state.

Referenced by stasis_state_add_publisher().

{
   struct stasis_state_publisher *pub = obj;

   ao2_ref(pub->state, -1);
}

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 688 of file stasis_state.c.

References AST_VECTOR_APPEND, AST_VECTOR_RW_UNLOCK, and AST_VECTOR_RW_WRLOCK.

Referenced by ast_mwi_add_observer(), and subscriptions_create().

{
   int res;

   AST_VECTOR_RW_WRLOCK(&manager->observers);
   res = AST_VECTOR_APPEND(&manager->observers, observer);
   AST_VECTOR_RW_UNLOCK(&manager->observers);

   return res;
}

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 531 of file stasis_state.c.

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

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

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 412 of file stasis_state.c.

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_state::num_subscribers, stasis_topic_name(), stasis_state_subscriber::state, state_find_or_add, sub, and subscriber_dtor().

Referenced by ast_mwi_add_subscriber(), and stasis_state_subscribe_pool().

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

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 364 of file stasis_state.c.

References stasis_state_manager::all_topic.

Referenced by ast_mwi_topic_all().

{
   return manager->all_topic;
}

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 740 of file stasis_state.c.

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

Referenced by ast_mwi_state_callback_all(), and publish().

{
   ast_assert(handler != NULL);

   ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
      handle_stasis_state_proxy, handler, data);
}

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 763 of file stasis_state.c.

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

Referenced by ast_mwi_state_callback_subscribed().

{
   ast_assert(handler != NULL);

   ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
      handle_stasis_state_subscribed, handler, data);
}

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 324 of file stasis_state.c.

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

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

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 562 of file stasis_state.c.

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

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

   stasis_publish(pub->state->topic, msg);
}

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 638 of file stasis_state.c.

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

Referenced by ast_mwi_publish_by_mailbox(), and implicit_publish_cb().

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

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 552 of file stasis_state.c.

References stasis_state_publisher::state.

Referenced by ast_mwi_publish(), and explicit_publish_cb().

{
   return pub->state->id;
}

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 557 of file stasis_state.c.

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

{
   return pub->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 700 of file stasis_state.c.

References AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_REMOVE_ELEM_UNORDERED, AST_VECTOR_RW_UNLOCK, and AST_VECTOR_RW_WRLOCK.

Referenced by ast_mwi_remove_observer(), and subscriptions_destroy().

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

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 658 of file stasis_state.c.

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

Referenced by ast_delete_mwi_state_full(), and publishers_destroy().

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

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 446 of file stasis_state.c.

References ao2_ref, ast_debug, NULL, stasis_state_add_subscriber(), stasis_state_subscriber::stasis_sub, stasis_subscribe_pool, stasis_topic_name(), stasis_state_subscriber::state, sub, and stasis_state::topic.

Referenced by ast_mwi_subscribe_pool(), and subscriptions_create().

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

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 497 of file stasis_state.c.

References ao2_bump, ao2_lock, ao2_unlock, stasis_state::msg, stasis_message_data(), and stasis_state_subscriber::state.

Referenced by ast_mwi_subscriber_data(), and handle_validate().

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

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 487 of file stasis_state.c.

References stasis_state_subscriber::state.

Referenced by ast_mwi_subscriber_data().

{
   return sub->state->id;
}

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 513 of file stasis_state.c.

References stasis_state_subscriber::stasis_sub.

Referenced by ast_mwi_subscriber_subscription().

{
   return sub->stasis_sub;
}

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 492 of file stasis_state.c.

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

Referenced by ast_mwi_subscriber_topic().

{
   return sub->state->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 369 of file stasis_state.c.

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

Referenced by ast_mwi_topic().

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

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 470 of file stasis_state.c.

References ao2_ref, NULL, stasis_state_subscriber::stasis_sub, and stasis_unsubscribe().

Referenced by ast_mwi_unsubscribe().

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

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 477 of file stasis_state.c.

References ao2_ref, NULL, stasis_state_subscriber::stasis_sub, and stasis_unsubscribe_and_join().

Referenced by ast_mwi_unsubscribe_and_join(), and subscriptions_destroy().

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

   return NULL;
}

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 167 of file stasis_state.c.

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, stasis_state::forward, 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(), stasis_state_manager::states, and stasis_state::topic.

Referenced by __state_find_or_add().

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

state_dtor()

static void state_dtor ( void *  obj )

static

Definition at line 121 of file stasis_state.c.

References ao2_cleanup, ast_assert, AST_VECTOR_FREE, AST_VECTOR_SIZE, stasis_state::forward, stasis_state::msg, NULL, stasis_forward_cancel(), and stasis_state::topic.

Referenced by state_alloc().

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

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 620 of file stasis_state.c.

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

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

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 586 of file stasis_state.c.

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

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

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.

References ast_assert, ast_begins_with(), stasis_state_proxy::id, NULL, and stasis_topic_name().

Referenced by __state_find_or_add(), and state_alloc().

{
   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) != NULL);

   return (id + 1);
}

state_manager_dtor()

static void state_manager_dtor ( void *  obj )

static

Definition at line 292 of file stasis_state.c.

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

Referenced by stasis_state_manager_create().

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

state_proxy_dtor()

static void state_proxy_dtor ( void *  obj )

static

Definition at line 136 of file stasis_state.c.

References ao2_cleanup, and stasis_state_proxy::manager.

Referenced by state_alloc().

                                        {
   struct stasis_state_proxy *proxy = obj;

   ao2_cleanup(proxy->manager);
}

state_proxy_sub_cb()

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

static

Definition at line 142 of file stasis_state.c.

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

Referenced by state_alloc().

{
   struct stasis_state_proxy *proxy = obj;

   ao2_unlink(proxy->manager->states, proxy);
}

subscriber_dtor()

static void subscriber_dtor ( void *  obj )

static

Definition at line 391 of file stasis_state.c.

References ao2_lock, ao2_ref, ao2_unlock, AST_VECTOR_GET, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, stasis_state_proxy::manager, stasis_state::manager, stasis_state::num_subscribers, stasis_state_subscriber::state, and sub.

Referenced by stasis_state_add_subscriber().

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