Asterisk - The Open Source Telephony Project  21.4.1
Data Structures | Typedefs | Functions
stasis_state.h File Reference

Stasis State API. More...

#include "asterisk/stasis.h"

Go to the source code of this file.

Data Structures

struct  stasis_state_observer
 Managed stasis state event interface. More...
 

Typedefs

typedef int(* on_stasis_state) (const char *id, struct stasis_message *msg, void *user_data)
 The delegate called for each managed state. More...
 

Functions

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

Detailed Description

Stasis State API.

Intro

This module defines the data structures, and handling of "state" for topics within stasis. State is defined as the last stasis message, and its contained message data, published on a given topic.

Concepts to know:

stasis_state_manager

The manager stores and well, manages state data. Each state is an association of a unique stasis topic, and the last known published stasis message on that topic. There is only ever one managed state object per topic. For each topic all messages are forwarded to an "all" topic also maintained by the manager. This allows subscriptions to all managed topics, and their state. Managed state is created in one of several ways:

Adding an explicit subscriber Adding an explicit publisher Adding an implicit publisher Retrieving a stasis state topic from the manager via the stasis_state_topic function prior to doing one of the above (DO NOT DO THIS).

More on the first three options later (see relevant section descriptions below). The last option, creation through retrieving a topic is not only NOT recommended, but should NOT even BE DONE. Doing so will inevitably result in a memory leak. Why then is this even allowed? The short answer is backwards compatibility. The slightly longer answer is at the time of this module's creation that's how things were historically done using a combination of stasis topic management spread throughout various other modules, and stasis caching. And yes it did cause a memory leak.

Preferably, any new code wishing to track topics and states should do so by adding either an explicit subscriber and/or publisher.

stasis_state_subscriber

As mentioned, topic and state can be created, or referenced within the manager by adding a stasis_state_subscriber. When adding a subscriber if no state currently exists new managed state is immediately created. If managed state already exists then a new subscriber is created referencing that state. The managed state is guaranteed to live throughout the subscriber's lifetime. State is only removed from the manager when no other entities require it (no more subscribers, or publishers).

Subscribers are ao2 objects. Therefore there is no explicit cleanup required aside from dereferencing the subscriber object using normal ao2 dereferencing methods.

stasis_state_publisher

There are two ways of tracking publishers: explicitly and implicitly.

Topic and state can be created, or referenced within the manager by also explicitly adding a stasis_state_publisher. When adding a publisher if no state currently exists new managed state is created. If managed state already exists then a new publisher is created referencing that state. The managed state is guaranteed to live throughout the publisher's lifetime. State is only removed from the manager when no other entities require it (no more publishers, or subscribers).

Explicit publishers are ao2 objects. Therefore there is no cleanup required aside from dereferencing the publisher object using normal ao2 dereferencing methods.

When adding an explicit publisher, messages should be published using the stasis_state_publish function. This not only skips a lookup, but doesn't add an implicit publisher. They are not necessarily mutually exclusive it's just that the two ways exist to solve two different problems.

For example (using an explicit publisher):

// Add an explicit publisher to topic/state "8675309" within // a given manager context pub = stasis_state_add_publisher(manager, "8675309");

// Publish a stasis message to the topic/state stasis_state_publish(pub, msg);

// Publish another a stasis message to the topic/state stasis_state_publish(pub, msg);

// Done with the publisher release the reference ao2_ref(pub, -1);

An implicit publisher can also be created by calling stasis_state_publish_by_id. Calling this function not only publishes the message within stasis (creating managed state if needed) it also sets up internal tracking of the publishing module using an ast_eid. However, a final call to stasis_state_remove_publish_by_id must be done in order to remove the eid reference, which will subsequently allow the underlying managed state to be eventually deleted.

For example (using an implicit publisher):

// Publish a stasis message to topic/state 8675309 within a // given manager context and use the system's default eid stasis_state_publish_by_id(manager, "8675309", NULL, msg);

// Do some stuff and then publish again stasis_state_publish_by_id(manager, "8675309", NULL, msg);

// Done with all our publishing, so post a final clearing // message and remove the implicit publisher stasis_state_remove_publish_by_id(manager, "8675309", NULL, msg);

Explicit publisher/publishing is preferred. However, implicit publishing is allowed for those situations where it makes more sense to do so, but has been implemented mostly for backwards compatibility with some modules (using implicit publishing required less initial code changes to some legacy subsystems).

stasis_state_observer

Some modules may wish to watch for, and react to managed state events. By registering a state observer, and implementing handlers for the desired callbacks those modules can do so.

Definition in file stasis_state.h.

Typedef Documentation

typedef int(* on_stasis_state) (const char *id, struct stasis_message *msg, void *user_data)

The delegate called for each managed state.

Parameters
idThe unique id of a managed state object
msgThe last published message on the state, or NULL
user_dataData object the user passed into the manager callback
Return values
0to continue traversing
CMP_STOP(2) to stop traversing
Since
13.28.0
16.5.0

Definition at line 521 of file stasis_state.h.

Function Documentation

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.

References AST_VECTOR_APPEND, AST_VECTOR_RW_UNLOCK, and AST_VECTOR_RW_WRLOCK.

Referenced by ast_mwi_add_observer().

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 }
#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
#define AST_VECTOR_RW_WRLOCK(vec)
Obtain write lock on vector.
Definition: vector.h:887
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.

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_ref, stasis_topic_name(), and stasis_state_publisher::state.

Referenced by ast_mwi_add_publisher().

534 {
535  struct stasis_state_publisher *pub = ao2_alloc_options(
536  sizeof(*pub), publisher_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
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 }
struct stasis_topic * all_topic
Definition: stasis_state.c:83
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
const char * stasis_topic_name(const struct stasis_topic *topic)
Return the name of a topic.
Definition: stasis.c:627
struct stasis_state * state
Definition: stasis_state.c:522
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.

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_ref, AST_VECTOR_GET, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, stasis_state::num_subscribers, stasis_topic_name(), and stasis_state_subscriber::state.

Referenced by ast_mwi_add_subscriber(), and stasis_state_subscribe_pool().

415 {
416  size_t i;
417  struct stasis_state_subscriber *sub = ao2_alloc_options(
418  sizeof(*sub), subscriber_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
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_state * state
Definition: stasis_state.c:387
#define AST_VECTOR_RW_UNLOCK(vec)
Unlock vector.
Definition: vector.h:897
#define AST_VECTOR_RW_RDLOCK(vec)
Obtain read lock on vector.
Definition: vector.h:877
struct stasis_topic * all_topic
Definition: stasis_state.c:83
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
const char * stasis_topic_name(const struct stasis_topic *topic)
Return the name of a topic.
Definition: stasis.c:627
#define AST_VECTOR_GET(vec, idx)
Get an element from a vector.
Definition: vector.h:680
unsigned int num_subscribers
Definition: stasis_state.c:52
#define AST_VECTOR_SIZE(vec)
Get the number of elements in a vector.
Definition: vector.h:609
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.

References stasis_state_manager::all_topic.

Referenced by ast_mwi_topic_all().

366 {
367  return manager->all_topic;
368 }
struct stasis_topic * all_topic
Definition: stasis_state.c:83
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.

References OBJ_MULTIPLE, OBJ_NODATA, and stasis_state_manager::states.

Referenced by ast_mwi_state_callback_all().

743 {
744  ast_assert(handler != NULL);
745 
746  ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
747  handle_stasis_state_proxy, handler, data);
748 }
struct ao2_container * states
Definition: stasis_state.c:81
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.

References OBJ_MULTIPLE, OBJ_NODATA, and stasis_state_manager::states.

Referenced by ast_mwi_state_callback_subscribed().

766 {
767  ast_assert(handler != NULL);
768 
769  ao2_callback_data(manager->states, OBJ_MULTIPLE | OBJ_NODATA,
770  handle_stasis_state_subscribed, handler, data);
771 }
struct ao2_container * states
Definition: stasis_state.c:81
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.

References stasis_state_manager::all_topic, AO2_ALLOC_OPT_LOCK_MUTEX, AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_container_alloc_hash, ao2_container_register(), ao2_ref, ast_alloca, AST_VECTOR_RW_INIT, stasis_state::manager, stasis_topic_create(), stasis_topic_name(), and stasis_state_manager::states.

Referenced by mwi_init().

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_VECTOR_RW_INIT(vec, size)
Initialize a vector with a read/write lock.
Definition: vector.h:158
struct stasis_topic * all_topic
Definition: stasis_state.c:83
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_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
struct stasis_topic * stasis_topic_create(const char *name)
Create a new topic.
Definition: stasis.c:617
#define ast_alloca(size)
call __builtin_alloca to ensure we get gcc builtin semantics
Definition: astmm.h:288
#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
const char * stasis_topic_name(const struct stasis_topic *topic)
Return the name of a topic.
Definition: stasis.c:627
struct ao2_container * states
Definition: stasis_state.c:81
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.

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

Referenced by ast_mwi_publish().

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 }
struct stasis_topic * topic
Definition: stasis_state.c:61
struct stasis_message * msg
Definition: stasis_state.c:63
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1511
#define ao2_replace(dst, src)
Replace one object reference with another cleaning up the original.
Definition: astobj2.h:501
struct stasis_state * state
Definition: stasis_state.c:522
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.

References ao2_ref, ao2_replace, stasis_state::msg, stasis_publish(), and stasis_state::topic.

Referenced by ast_mwi_publish_by_mailbox().

641 {
642  struct stasis_state *state;
643 
644  state = state_find_or_add(manager, NULL, id);
645  if (!state) {
646  return;
647  }
648 
649  ao2_lock(state);
650  state_find_or_add_eid(state, eid);
651  ao2_replace(state->msg, msg);
652  ao2_unlock(state);
653 
654  stasis_publish(state->topic, msg);
655 
656  ao2_ref(state, -1);
657 }
struct stasis_topic * topic
Definition: stasis_state.c:61
struct stasis_message * msg
Definition: stasis_state.c:63
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1511
#define ao2_replace(dst, src)
Replace one object reference with another cleaning up the original.
Definition: astobj2.h:501
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.

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

Referenced by ast_mwi_publish().

554 {
555  return pub->state->id;
556 }
struct stasis_state * state
Definition: stasis_state.c:522
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.

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

559 {
560  return pub->state->topic;
561 }
struct stasis_topic * topic
Definition: stasis_state.c:61
struct stasis_state * state
Definition: stasis_state.c:522
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.

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

703 {
704  AST_VECTOR_RW_WRLOCK(&manager->observers);
705  AST_VECTOR_REMOVE_ELEM_UNORDERED(&manager->observers, observer, AST_VECTOR_ELEM_CLEANUP_NOOP);
706  AST_VECTOR_RW_UNLOCK(&manager->observers);
707 }
#define AST_VECTOR_REMOVE_ELEM_UNORDERED(vec, elem, cleanup)
Remove an element from a vector.
Definition: vector.h:583
#define AST_VECTOR_RW_UNLOCK(vec)
Unlock vector.
Definition: vector.h:897
#define AST_VECTOR_ELEM_CLEANUP_NOOP(elem)
Vector element cleanup that does nothing.
Definition: vector.h:571
#define AST_VECTOR_RW_WRLOCK(vec)
Obtain write lock on vector.
Definition: vector.h:887
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.

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

Referenced by ast_delete_mwi_state_full().

661 {
662  struct stasis_state *state = ao2_weakproxy_find(manager->states, id, OBJ_SEARCH_KEY, "");
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 
682  ao2_lock(state);
683  state_find_and_remove_eid(state, eid);
684  ao2_unlock(state);
685 
686  ao2_ref(state, -1);
687 }
struct stasis_topic * topic
Definition: stasis_state.c:61
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ast_debug(level,...)
Log a DEBUG message.
#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
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1511
struct ao2_container * states
Definition: stasis_state.c:81
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.

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

Referenced by ast_mwi_subscribe_pool().

449 {
450  struct stasis_topic *topic;
451  struct stasis_state_subscriber *sub = stasis_state_add_subscriber(manager, id);
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 }
struct stasis_state * state
Definition: stasis_state.c:387
struct stasis_topic * topic
Definition: stasis_state.c:61
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
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ast_debug(level,...)
Log a DEBUG message.
const char * stasis_topic_name(const struct stasis_topic *topic)
Return the name of a topic.
Definition: stasis.c:627
struct stasis_subscription * stasis_sub
Definition: stasis_state.c:389
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.

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

Referenced by ast_mwi_subscriber_data().

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 }
struct stasis_state * state
Definition: stasis_state.c:387
struct stasis_message * msg
Definition: stasis_state.c:63
#define ao2_bump(obj)
Bump refcount on an AO2 object by one, returning the object.
Definition: astobj2.h:480
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
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.

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

Referenced by ast_mwi_subscriber_data().

489 {
490  return sub->state->id;
491 }
struct stasis_state * state
Definition: stasis_state.c:387
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.

References stasis_state_subscriber::stasis_sub.

Referenced by ast_mwi_subscriber_subscription().

516 {
517  return sub->stasis_sub;
518 }
struct stasis_subscription * stasis_sub
Definition: stasis_state.c:389
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.

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

Referenced by ast_mwi_subscriber_topic().

494 {
495  return sub->state->topic;
496 }
struct stasis_state * state
Definition: stasis_state.c:387
struct stasis_topic * topic
Definition: stasis_state.c:61
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.

References ao2_ref, and stasis_state::topic.

Referenced by ast_mwi_topic().

371 {
372  struct stasis_topic *topic;
373  struct stasis_state *state;
374 
375  state = state_find_or_add(manager, NULL, id);
376  if (!state) {
377  return NULL;
378  }
379 
380  topic = state->topic;
381  ao2_ref(state, -1);
382  return topic;
383 }
struct stasis_topic * topic
Definition: stasis_state.c:61
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
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.

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

Referenced by ast_mwi_unsubscribe().

472 {
474  ao2_ref(sub, -1);
475  return NULL;
476 }
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
struct stasis_subscription * stasis_sub
Definition: stasis_state.c:389
struct stasis_subscription * stasis_unsubscribe(struct stasis_subscription *subscription)
Cancel a subscription.
Definition: stasis.c:971
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.

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

Referenced by ast_mwi_unsubscribe_and_join().

479 {
480  if (sub) {
482  ao2_ref(sub, -1);
483  }
484 
485  return NULL;
486 }
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
struct stasis_subscription * stasis_sub
Definition: stasis_state.c:389
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