Asterisk - The Open Source Telephony Project GIT-master-2de1a68
Data Structures | Macros | Functions
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_publisherstasis_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_subscriberstasis_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_topicstasis_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_managerstasis_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_topicstasis_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_subscriberstasis_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_subscriptionstasis_state_subscriber_subscription (struct stasis_state_subscriber *sub)
 Retrieve the stasis topic subscription if available. More...
 
struct stasis_topicstasis_state_subscriber_topic (struct stasis_state_subscriber *sub)
 Retrieve the subscriber's topic. More...
 
struct stasis_topicstasis_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_statestate_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.

275{
276 struct stasis_state *state;
277
279 if (ast_strlen_zero(id)) {
280 id = state_id_by_topic(manager->all_topic, state_topic);
281 }
282
284 if (!state) {
285 state = state_alloc(manager, state_topic, id, file, line, func);
286 }
287
289
290 return state;
291}
#define ao2_unlock(a)
Definition: astobj2.h:729
#define ao2_weakproxy_find(c, arg, flags, tag)
Perform an ao2_find on a container with ao2_weakproxy objects, returning the real object.
Definition: astobj2.h:1748
#define ao2_lock(a)
Definition: astobj2.h:717
@ OBJ_NOLOCK
Assume that the ao2_container is already locked.
Definition: astobj2.h:1063
@ OBJ_SEARCH_KEY
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
enum cc_state state
Definition: ccss.c:393
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)
Definition: stasis_state.c:168
static const char * state_id_by_topic(struct stasis_topic *manager_topic, const struct stasis_topic *state_topic)
Definition: stasis_state.c:104
static force_inline int attribute_pure ast_strlen_zero(const char *s)
Definition: strings.h:65
struct ao2_container * states
Definition: stasis_state.c:81
struct stasis_topic * all_topic
Definition: stasis_state.c:83
struct stasis_state_manager * manager
The manager that owns and handles this state.
Definition: stasis_state.c:57

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.

710{
711 struct stasis_message *msg;
712 int res;
713
714 /*
715 * State needs to be locked here while we retrieve and bump the reference on its message
716 * object. Doing so guarantees the message object will live throughout its handling.
717 */
719 msg = ao2_bump(state->msg);
721
722 res = handler(state->id, msg, data);
723 ao2_cleanup(msg);
724 return res;
725}
#define ao2_cleanup(obj)
Definition: astobj2.h:1934
#define ao2_bump(obj)
Bump refcount on an AO2 object by one, returning the object.
Definition: astobj2.h:480
static void handler(const char *name, int response_code, struct ast_variable *get_params, struct ast_variable *path_vars, struct ast_variable *headers, struct ast_json *body, struct ast_ari_response *response)
Definition: test_ari.c:59

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.

728{
730
731 if (state) {
732 int res;
733 res = handle_stasis_state(state, arg, data);
734 ao2_ref(state, -1);
735 return res;
736 }
737
738 return 0;
739}
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ao2_weakproxy_get_object(weakproxy, flags)
Get the object associated with weakproxy.
Definition: astobj2.h:621
static int handle_stasis_state(struct stasis_state *state, on_stasis_state handler, void *data)
Definition: stasis_state.c:709

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.

751{
753 int res = 0;
754
755 if (state && state->num_subscribers) {
756 res = handle_stasis_state(state, arg, data);
757 }
758
760
761 return res;
762}

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.

526{
527 struct stasis_state_publisher *pub = obj;
528
529 ao2_ref(pub->state, -1);
530}
struct stasis_state * state
Definition: stasis_state.c:522

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
managerThe state manager
observerThe observer handling events
Return values
0if successfully registered
-1on failure
Since
13.28.0
16.5.0

Definition at line 689 of file stasis_state.c.

691{
692 int res;
693
694 AST_VECTOR_RW_WRLOCK(&manager->observers);
695 res = AST_VECTOR_APPEND(&manager->observers, observer);
696 AST_VECTOR_RW_UNLOCK(&manager->observers);
697
698 return res;
699}
struct ast_sorcery_instance_observer observer
#define AST_VECTOR_RW_WRLOCK(vec)
Obtain write lock on vector.
Definition: vector.h:887
#define AST_VECTOR_RW_UNLOCK(vec)
Unlock vector.
Definition: vector.h:897
#define AST_VECTOR_APPEND(vec, elem)
Append an element to a vector, growing the vector if needed.
Definition: vector.h:256

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
managerThe manager object
idThe unique id of a managed state
Return values
Astasis state publisher
NULLif an error occurred
Since
13.28.0
16.5.0

Definition at line 532 of file stasis_state.c.

534{
537
538 if (!pub) {
539 ast_log(LOG_ERROR, "Unable to create publisher to %s/%s\n",
540 stasis_topic_name(manager->all_topic), id);
541 return NULL;
542 }
543
544 pub->state = state_find_or_add(manager, NULL, id);
545 if (!pub->state) {
546 ao2_ref(pub, -1);
547 return NULL;
548 }
549
550 return pub;
551}
#define ast_log
Definition: astobj2.c:42
@ AO2_ALLOC_OPT_LOCK_NOLOCK
Definition: astobj2.h:367
#define ao2_alloc_options(data_size, destructor_fn, options)
Definition: astobj2.h:404
#define LOG_ERROR
#define NULL
Definition: resample.c:96
const char * stasis_topic_name(const struct stasis_topic *topic)
Return the name of a topic.
Definition: stasis.c:627
#define state_find_or_add(manager, state_topic, id)
Definition: stasis_state.c:271
static void publisher_dtor(void *obj)
Definition: stasis_state.c:525

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
managerThe manager object
idThe unique id of a managed state
Return values
Astasis state subscriber
NULLif an error occurred
Since
13.28.0
16.5.0

Definition at line 413 of file stasis_state.c.

415{
416 size_t i;
419
420 if (!sub) {
421 ast_log(LOG_ERROR, "Unable to create subscriber to %s/%s\n",
422 stasis_topic_name(manager->all_topic), id);
423 return NULL;
424 }
425
426 sub->state = state_find_or_add(manager, NULL, id);
427 if (!sub->state) {
428 ao2_ref(sub, -1);
429 return NULL;
430 }
431
432 ao2_lock(sub->state);
433 ++sub->state->num_subscribers;
434 ao2_unlock(sub->state);
435
436 AST_VECTOR_RW_RDLOCK(&manager->observers);
437 for (i = 0; i < AST_VECTOR_SIZE(&manager->observers); ++i) {
438 if (AST_VECTOR_GET(&manager->observers, i)->on_subscribe) {
439 AST_VECTOR_GET(&manager->observers, i)->on_subscribe(id, sub);
440 }
441 }
442 AST_VECTOR_RW_UNLOCK(&manager->observers);
443
444 return sub;
445}
struct stasis_forward * sub
Definition: res_corosync.c:240
static void subscriber_dtor(void *obj)
Definition: stasis_state.c:392
#define AST_VECTOR_SIZE(vec)
Get the number of elements in a vector.
Definition: vector.h:609
#define AST_VECTOR_RW_RDLOCK(vec)
Obtain read lock on vector.
Definition: vector.h:877
#define AST_VECTOR_GET(vec, idx)
Get an element from a vector.
Definition: vector.h:680

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
managerThe manager object
Return values
Themanager's topic.
Since
13.28.0
16.5.0

Definition at line 365 of file stasis_state.c.

366{
367 return manager->all_topic;
368}

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
managerThe state manager
handlerThe handler to call for each managed state
dataUser to data to pass on to the handler
Since
13.28.0
16.5.0

Definition at line 741 of file stasis_state.c.

743{
745
748}
#define ao2_callback_data(container, flags, cb_fn, arg, data)
Definition: astobj2.h:1723
@ OBJ_NODATA
Definition: astobj2.h:1044
@ OBJ_MULTIPLE
Definition: astobj2.h:1049
static int handle_stasis_state_proxy(void *obj, void *arg, void *data, int flags)
Definition: stasis_state.c:727
#define ast_assert(a)
Definition: utils.h:739

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
managerThe state manager
handlerThe handler to call for each managed state
dataUser to data to pass on to the handler
Since
13.28.0
16.5.0

Definition at line 764 of file stasis_state.c.

766{
768
771}
static int handle_stasis_state_subscribed(void *obj, void *arg, void *data, int flags)
Definition: stasis_state.c:750

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_nameThe name of the topic to create that all state topics get forwarded to
Return values
Astasis state manager
NULLif an error occurred
Since
13.28.0
16.5.0

Definition at line 325 of file stasis_state.c.

326{
327 struct stasis_state_manager *manager;
328
329 manager = ao2_alloc_options(sizeof(*manager), state_manager_dtor,
331 if (!manager) {
332 return NULL;
333 }
334
336 STATE_BUCKETS, stasis_state_proxy_hash_fn, NULL, stasis_state_proxy_cmp_fn);
337 if (!manager->states) {
338 ao2_ref(manager, -1);
339 return NULL;
340 }
341
342 manager->all_topic = stasis_topic_create(topic_name);
343 if (!manager->all_topic) {
344 ao2_ref(manager, -1);
345 return NULL;
346 }
347
348 if (AST_VECTOR_RW_INIT(&manager->observers, 2) != 0) {
349 ao2_ref(manager, -1);
350 return NULL;
351 }
352
353#ifdef AO2_DEBUG
354 {
355 char *container_name =
356 ast_alloca(strlen(stasis_topic_name(manager->all_topic)) + strlen("-manager") + 1);
357 sprintf(container_name, "%s-manager", stasis_topic_name(manager->all_topic));
358 ao2_container_register(container_name, manager->states, state_prnt_obj);
359 }
360#endif
361
362 return manager;
363}
#define ast_alloca(size)
call __builtin_alloca to ensure we get gcc builtin semantics
Definition: astmm.h:288
@ AO2_ALLOC_OPT_LOCK_MUTEX
Definition: astobj2.h:363
int ao2_container_register(const char *name, struct ao2_container *self, ao2_prnt_obj_fn *prnt_obj)
Register a container for CLI stats and integrity check.
#define ao2_container_alloc_hash(ao2_options, container_options, n_buckets, hash_fn, sort_fn, cmp_fn)
Allocate and initialize a hash container with the desired number of buckets.
Definition: astobj2.h:1303
struct stasis_topic * stasis_topic_create(const char *name)
Create a new topic.
Definition: stasis.c:617
static void state_manager_dtor(void *obj)
Definition: stasis_state.c:293
#define STATE_BUCKETS
Definition: stasis_state.c:77
#define AST_VECTOR_RW_INIT(vec, size)
Initialize a vector with a read/write lock.
Definition: vector.h:158

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
pubThe publisher to use to publish the message
msgThe message to publish
Since
13.28.0
16.5.0

Definition at line 563 of file stasis_state.c.

564{
565 ao2_lock(pub->state);
566 ao2_replace(pub->state->msg, msg);
567 ao2_unlock(pub->state);
568
569 stasis_publish(pub->state->topic, msg);
570}
#define ao2_replace(dst, src)
Replace one object reference with another cleaning up the original.
Definition: astobj2.h:501
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1511
struct stasis_topic * topic
Definition: stasis_state.c:61
struct stasis_message * msg
Definition: stasis_state.c:63

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
managerThe state manager
idA state's unique id
eidThe unique system id
msgThe message to publish
Since
13.28.0
16.5.0

Definition at line 639 of file stasis_state.c.

641{
642 struct stasis_state *state;
643
645 if (!state) {
646 return;
647 }
648
651 ao2_replace(state->msg, msg);
653
654 stasis_publish(state->topic, msg);
655
656 ao2_ref(state, -1);
657}
static void state_find_or_add_eid(struct stasis_state *state, const struct ast_eid *eid)
Definition: stasis_state.c:587

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
pubA stasis state publisher
Return values
Themanaged state's id
Since
13.28.0
16.5.0

Definition at line 553 of file stasis_state.c.

554{
555 return pub->state->id;
556}

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
pubA stasis state publisher
Return values
Thepublisher's topic
Since
13.28.0
16.5.0

Definition at line 558 of file stasis_state.c.

559{
560 return pub->state->topic;
561}

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
managerThe state manager
observerThe observer being removed
Since
13.28.0
16.5.0

Definition at line 701 of file stasis_state.c.

703{
704 AST_VECTOR_RW_WRLOCK(&manager->observers);
706 AST_VECTOR_RW_UNLOCK(&manager->observers);
707}
#define AST_VECTOR_ELEM_CLEANUP_NOOP(elem)
Vector element cleanup that does nothing.
Definition: vector.h:571
#define AST_VECTOR_REMOVE_ELEM_UNORDERED(vec, elem, cleanup)
Remove an element from a vector.
Definition: vector.h:583

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
managerThe state manager
idA state's unique id
eidThe unique system id
msgThe message to publish (can be NULL)
Since
13.28.0
16.5.0

Definition at line 659 of file stasis_state.c.

661{
663
664 if (!state) {
665 /*
666 * In most circumstances state should already exist here. However, if there is no
667 * state then it can mean one of a few things:
668 *
669 * 1. This function was called prior to an implicit publish for the same given
670 * manager, and id.
671 * 2. This function was called more than once for the same manager, and id.
672 * 3. There is ref count problem with the explicit subscribers, and publishers.
673 */
674 ast_debug(5, "Attempted to remove state for id '%s', but state not found\n", id);
675 return;
676 }
677
678 if (msg) {
679 stasis_publish(state->topic, msg);
680 }
681
685
686 ao2_ref(state, -1);
687}
#define ast_debug(level,...)
Log a DEBUG message.
static void state_find_and_remove_eid(struct stasis_state *state, const struct ast_eid *eid)
Definition: stasis_state.c:621

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
managerThe manager object
idThe unique id of a managed state
callbackThe stasis subscription callback
dataA user data object passed to the stasis subscription
Return values
Astasis state subscriber
NULLif an error occurred
Since
13.28.0
16.5.0

Definition at line 447 of file stasis_state.c.

449{
450 struct stasis_topic *topic;
452
453 if (!sub) {
454 return NULL;
455 }
456
457 topic = sub->state->topic;
458 ast_debug(3, "Creating stasis state subscription to id '%s'. Topic: '%s':%p %d\n",
459 id, stasis_topic_name(topic), topic, (int)ao2_ref(topic, 0));
460
461 sub->stasis_sub = stasis_subscribe_pool(topic, callback, data);
462
463 if (!sub->stasis_sub) {
464 ao2_ref(sub, -1);
465 return NULL;
466 }
467
468 return sub;
469}
#define stasis_subscribe_pool(topic, callback, data)
Definition: stasis.h:680
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.
Definition: stasis_state.c:413

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
subA stasis state subscriber
Return values
Thesubscriber's state message data
NULLif no data has been published yet
Since
13.28.0
16.5.0

Definition at line 498 of file stasis_state.c.

499{
500 void *res;
501
502 /*
503 * The data's reference needs to be bumped before returning so it doesn't disappear
504 * for the caller. Lock state, so the underlying message data is not replaced while
505 * retrieving.
506 */
507 ao2_lock(sub->state);
508 res = ao2_bump(stasis_message_data(sub->state->msg));
509 ao2_unlock(sub->state);
510
511 return res;
512}
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.

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
subA stasis state subscriber
Return values
Themanaged state's id
Since
13.28.0
16.5.0

Definition at line 488 of file stasis_state.c.

489{
490 return sub->state->id;
491}

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
subA stasis state subscriber
Return values
Thesubscriber's stasis subscription
NULLif no subscription available
Since
13.28.0
16.5.0

Definition at line 514 of file stasis_state.c.

516{
517 return sub->stasis_sub;
518}

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
subA stasis state subscriber
Return values
Thesubscriber's topic
Since
13.28.0
16.5.0

Definition at line 493 of file stasis_state.c.

494{
495 return sub->state->topic;
496}

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
managerThe manager object
idThe unique id of/for the topic
Return values
Amanaged stasis topic.
NULLif an error occurred
Since
13.28.0
16.5.0

Definition at line 370 of file stasis_state.c.

371{
372 struct stasis_topic *topic;
373 struct stasis_state *state;
374
376 if (!state) {
377 return NULL;
378 }
379
380 topic = state->topic;
381 ao2_ref(state, -1);
382 return topic;
383}

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
subA stasis state subscriber
Return values
NULL
Since
13.28.0
16.5.0

Definition at line 471 of file stasis_state.c.

472{
473 sub->stasis_sub = stasis_unsubscribe(sub->stasis_sub);
474 ao2_ref(sub, -1);
475 return NULL;
476}
struct stasis_subscription * stasis_unsubscribe(struct stasis_subscription *subscription)
Cancel a subscription.
Definition: stasis.c:971

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
subA stasis state subscriber
Return values
NULL
Since
13.28.0
16.5.0

Definition at line 478 of file stasis_state.c.

479{
480 if (sub) {
481 sub->stasis_sub = stasis_unsubscribe_and_join(sub->stasis_sub);
482 ao2_ref(sub, -1);
483 }
484
485 return NULL;
486}
struct stasis_subscription * stasis_unsubscribe_and_join(struct stasis_subscription *subscription)
Cancel a subscription, blocking until the last message is processed.
Definition: stasis.c:1134

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.

171{
172 struct stasis_state_proxy *proxy = NULL;
173 struct stasis_state *state = NULL;
174
175 if (!id) {
176 /* If not given an id, then a state topic is required */
177 ast_assert(state_topic != NULL);
178
179 /* Get the id we'll key off of from the state topic */
180 id = state_id_by_topic(manager->all_topic, state_topic);
181 }
182
183 state = __ao2_alloc(sizeof(*state), state_dtor, AO2_ALLOC_OPT_LOCK_MUTEX, id, file, line, func);
184 if (!state) {
185 goto error_return;
186 }
187
188 if (!state_topic) {
189 char *name;
190
191 /*
192 * To provide further detail and to ensure that the topic is unique within the
193 * scope of the system we prefix it with the manager's topic name, which should
194 * itself already be unique.
195 */
196 if (ast_asprintf(&name, "%s/%s", stasis_topic_name(manager->all_topic), id) < 0) {
197 goto error_return;
198 }
199
201
202 ast_free(name);
203 if (!state->topic) {
204 goto error_return;
205 }
206 } else {
207 /*
208 * Since the state topic was passed in, go ahead and bump its reference.
209 * By doing this here first, it allows us to consistently decrease the reference on
210 * state allocation error.
211 */
212 ao2_ref(state_topic, +1);
213 state->topic = state_topic;
214 }
215
216 proxy = ao2_t_weakproxy_alloc(sizeof(*proxy) + strlen(id) + 1, state_proxy_dtor, id);
217 if (!proxy) {
218 goto error_return;
219 }
220
221 strcpy(proxy->id, id); /* Safe */
222
223 state->id = proxy->id;
224 proxy->manager = ao2_bump(manager);
225 state->manager = proxy->manager; /* state->manager is owned by the proxy */
226
227 state->forward = stasis_forward_all(state->topic, manager->all_topic);
228 if (!state->forward) {
229 goto error_return;
230 }
231
232 if (AST_VECTOR_INIT(&state->eids, 2)) {
233 goto error_return;
234 }
235
236 if (ao2_t_weakproxy_set_object(proxy, state, OBJ_NOLOCK, "weakproxy link")) {
237 goto error_return;
238 }
239
241 goto error_return;
242 }
243
244 if (!ao2_link_flags(manager->states, proxy, OBJ_NOLOCK)) {
245 goto error_return;
246 }
247
248 ao2_ref(proxy, -1);
249
250 return state;
251
252error_return:
253 ast_log(LOG_ERROR, "Unable to allocate state '%s' in manager '%s'\n",
256 ao2_cleanup(proxy);
257 return NULL;
258}
#define ast_free(a)
Definition: astmm.h:180
#define ast_asprintf(ret, fmt,...)
A wrapper for asprintf()
Definition: astmm.h:267
int ao2_weakproxy_subscribe(void *weakproxy, ao2_weakproxy_notification_cb cb, void *data, int flags)
Request notification when weakproxy points to NULL.
Definition: astobj2.c:934
#define ao2_link_flags(container, obj, flags)
Add an object to a container.
Definition: astobj2.h:1554
#define ao2_t_weakproxy_alloc(data_size, destructor_fn, tag)
Definition: astobj2.h:553
void * __ao2_alloc(size_t data_size, ao2_destructor_fn destructor_fn, unsigned int options, const char *tag, const char *file, int line, const char *func) attribute_warn_unused_result
Definition: astobj2.c:768
#define ao2_t_weakproxy_set_object(weakproxy, obj, flags, tag)
Definition: astobj2.h:582
static const char name[]
Definition: format_mp3.c:68
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.
Definition: stasis.c:1578
static void state_proxy_dtor(void *obj)
Definition: stasis_state.c:136
static void state_dtor(void *obj)
Definition: stasis_state.c:121
static void state_proxy_sub_cb(void *obj, void *data)
Definition: stasis_state.c:142
struct stasis_state_manager * manager
Definition: stasis_state.c:34
#define AST_VECTOR_INIT(vec, size)
Initialize a vector.
Definition: vector.h:113

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.

122{
123 struct stasis_state *state = obj;
124
125 state->forward = stasis_forward_cancel(state->forward);
126 ao2_cleanup(state->topic);
127 state->topic = NULL;
128 ao2_cleanup(state->msg);
129 state->msg = NULL;
130
131 /* All eids should have been removed */
132 ast_assert(AST_VECTOR_SIZE(&state->eids) == 0);
133 AST_VECTOR_FREE(&state->eids);
134}
struct stasis_forward * stasis_forward_cancel(struct stasis_forward *forward)
Definition: stasis.c:1548
#define AST_VECTOR_FREE(vec)
Deallocates this vector.
Definition: vector.h:174

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.

622{
623 size_t i;
624
625 if (!eid) {
626 eid = &ast_eid_default;
627 }
628
629 for (i = 0; i < AST_VECTOR_SIZE(&state->eids); ++i) {
630 if (!ast_eid_cmp(AST_VECTOR_GET_ADDR(&state->eids, i), eid)) {
632 /* Balance the reference from state_find_or_add_eid */
633 ao2_ref(state, -1);
634 return;
635 }
636 }
637}
int ast_eid_cmp(const struct ast_eid *eid1, const struct ast_eid *eid2)
Compare two EIDs.
Definition: utils.c:3094
struct ast_eid ast_eid_default
Global EID.
Definition: options.c:93
#define AST_VECTOR_REMOVE_UNORDERED(vec, idx)
Remove an element from an unordered vector by index.
Definition: vector.h:438
#define AST_VECTOR_GET_ADDR(vec, idx)
Get an address of element in a vector.
Definition: vector.h:668

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.

588{
589 size_t i;
590
591 if (!eid) {
592 eid = &ast_eid_default;
593 }
594
595 for (i = 0; i < AST_VECTOR_SIZE(&state->eids); ++i) {
596 if (!ast_eid_cmp(AST_VECTOR_GET_ADDR(&state->eids, i), eid)) {
597 break;
598 }
599 }
600
601 if (i == AST_VECTOR_SIZE(&state->eids)) {
602 if (!AST_VECTOR_APPEND(&state->eids, *eid)) {
603 /* This ensures state cannot be freed if it has any eids */
604 ao2_ref(state, +1);
605 }
606 }
607}

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.

106{
107 const char *id;
108
109 /* This topic should always belong to the manager */
111 stasis_topic_name(state_topic)));
112
113 id = strchr(stasis_topic_name(state_topic), '/');
114
115 /* The state's unique id should always exist */
116 ast_assert(id != NULL && *(id + 1) != '\0');
117
118 return (id + 1);
119}
enum queue_result id
Definition: app_queue.c:1638
static struct stasis_topic * manager_topic
A stasis_topic that all topics AMI cares about will be forwarded to.
Definition: manager.c:1609
static int force_inline attribute_pure ast_begins_with(const char *str, const char *prefix)
Checks whether a string begins with another.
Definition: strings.h:97

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.

294{
295 struct stasis_state_manager *manager = obj;
296
297#ifdef AO2_DEBUG
298 {
299 char *container_name =
300 ast_alloca(strlen(stasis_topic_name(manager->all_topic)) + strlen("-manager") + 1);
301 sprintf(container_name, "%s-manager", stasis_topic_name(manager->all_topic));
302 ao2_container_unregister(container_name);
303 }
304#endif
305
306 ao2_cleanup(manager->states);
307 manager->states = NULL;
308 ao2_cleanup(manager->all_topic);
309 manager->all_topic = NULL;
310 AST_VECTOR_RW_FREE(&manager->observers);
311}
void ao2_container_unregister(const char *name)
Unregister a container for CLI stats and integrity check.
#define AST_VECTOR_RW_FREE(vec)
Deallocates this locked vector.
Definition: vector.h:202

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.

136 {
137 struct stasis_state_proxy *proxy = obj;
138
139 ao2_cleanup(proxy->manager);
140}

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.

143{
144 struct stasis_state_proxy *proxy = obj;
145
146 ao2_unlink(proxy->manager->states, proxy);
147}
#define ao2_unlink(container, obj)
Remove an object from a container.
Definition: astobj2.h:1578

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.

393{
394 size_t i;
395 struct stasis_state_subscriber *sub = obj;
396 struct stasis_state_manager *manager = sub->state->manager;
397
398 AST_VECTOR_RW_RDLOCK(&manager->observers);
399 for (i = 0; i < AST_VECTOR_SIZE(&manager->observers); ++i) {
400 if (AST_VECTOR_GET(&manager->observers, i)->on_unsubscribe) {
401 AST_VECTOR_GET(&manager->observers, i)->on_unsubscribe(sub->state->id, sub);
402 }
403 }
404 AST_VECTOR_RW_UNLOCK(&manager->observers);
405
406 ao2_lock(sub->state);
407 --sub->state->num_subscribers;
408 ao2_unlock(sub->state);
409
410 ao2_ref(sub->state, -1);
411}

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