stasis_state.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2019, Sangoma Technologies Corporation
 *
 * Kevin Harwell <[email protected]>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */

/*** MODULEINFO
 <support_level>core</support_level>
 ***/

#include "asterisk.h"

#include "asterisk/stasis_state.h"

/*!
 * \internal
 * \brief Used to link a stasis_state to it's manager
 */
struct stasis_state_proxy {
 AO2_WEAKPROXY();
 /*! The manager that owns and handles this state */
 struct stasis_state_manager *manager;
 /*! A unique id for this state object. */
 char id[0];
};

/*!
 * \internal
 * \brief Associates a stasis topic to its last known published message
 *
 * This object's lifetime is tracked by the number of publishers and subscribers to it.
 * Once all publishers and subscribers have been removed this object is removed from the
 * manager's collection and destroyed. While a single object type (namely this one) could
 * be utilized for both publishers, and subscribers this implementation purposely keeps
 * them separated. This was done to maintain readability, make debugging easier, and allow
 * for better logging and future enhancements.
 */
struct stasis_state {
 /*! The number of state subscribers */
 unsigned int num_subscribers;
 /*!
 * \brief The manager that owns and handles this state
 * \note This reference is owned by stasis_state_proxy
 */
 struct stasis_state_manager *manager;
 /*! Forwarding information, i.e. this topic to manager's topic */
 struct stasis_forward *forward;
 /*! The managed topic */
 struct stasis_topic *topic;
 /*! The actual state data */
 struct stasis_message *msg;
 /*!
 * A container of eids. It's assumed that there is only a single publisher per
 * eid per topic. Thus the publisher is tracked by the system's eid.
 */
 AST_VECTOR(, struct ast_eid) eids;
 /*! A unique id for this state object. */
 char *id;
};

AO2_STRING_FIELD_HASH_FN(stasis_state_proxy, id);
AO2_STRING_FIELD_CMP_FN(stasis_state_proxy, id);

/*! The number of buckets to use for managed states */
#define STATE_BUCKETS 57

struct stasis_state_manager {
 /*! Holds all state objects handled by this manager */
 struct ao2_container *states;
 /*! The manager's topic. All state topics are forward to this one */
 struct stasis_topic *all_topic;
 /*! A collection of manager event handlers */
 AST_VECTOR_RW(, struct stasis_state_observer *) observers;
};

/*!
 * \internal
 * \brief Retrieve a state's topic name without the manager topic.
 *
 * State topics have names that consist of the manager's topic name
 * combined with a unique id separated by a slash. For instance:
 *
 * manager topic's name/unique id
 *
 * This method retrieves the unique id part from the state's topic name.
 *
 * \param manager_topic The manager's topic
 * \param state_topic A state topic
 *
 * \return The state's topic unique id
 */
static const char *state_id_by_topic(struct stasis_topic *manager_topic,
 const struct stasis_topic *state_topic)
{
 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);
}

static void state_dtor(void *obj)
{
 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);
}

static void state_proxy_dtor(void *obj) {
 struct stasis_state_proxy *proxy = obj;

 ao2_cleanup(proxy->manager);
}

static void state_proxy_sub_cb(void *obj, void *data)
{
 struct stasis_state_proxy *proxy = obj;

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

/*!
 * \internal
 * \brief Allocate a stasis state object and add it to the manager.
 *
 * Create and initialize a state structure. It's required that either a state
 * topic, or an id is specified. If a state topic is not given then one will be
 * created using the given id.
 *
 * \param manager The owning manager
 * \param state_topic A state topic to be managed
 * \param id The unique id for the state
 *
 * \return A stasis_state object or NULL
 * \return NULL on error
 *
 * \pre manager->states must be locked.
 * \pre manager->states does not contain an object matching key \a id.
 */
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)
{
 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;
}

/*!
 * \internal
 * \brief Find a state by id, or create one if not found and add it to the manager.
 *
 * \param manager The manager to be added to
 * \param state_topic A state topic to be managed (if NULL id is required)
 * \param id The unique id for the state (if NULL state_topic is required)
 *
 * \return The added state object
 * \return NULL on error
 */
#define state_find_or_add(mgr, top, id) __state_find_or_add(mgr, top, id, __FILE__, __LINE__, __PRETTY_FUNCTION__)
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)
{
 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;
}

static void state_manager_dtor(void *obj)
{
 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);
}

#ifdef AO2_DEBUG
static void state_prnt_obj(void *v_obj, void *where, ao2_prnt_fn *prnt)
{
 struct stasis_state *state = v_obj;

 if (!state) {
 return;
 }
 prnt(where, "%s", stasis_topic_name(state->topic));
}
#endif

struct stasis_state_manager *stasis_state_manager_create(const char *topic_name)
{
 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;
}

struct stasis_topic *stasis_state_all_topic(struct stasis_state_manager *manager)
{
 return manager->all_topic;
}

struct stasis_topic *stasis_state_topic(struct stasis_state_manager *manager, const char *id)
{
 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;
}

struct stasis_state_subscriber {
 /*! The stasis state subscribed to */
 struct stasis_state *state;
 /*! The stasis subscription. */
 struct stasis_subscription *stasis_sub;
};

static void subscriber_dtor(void *obj)
{
 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);
}

struct stasis_state_subscriber *stasis_state_add_subscriber(
 struct stasis_state_manager *manager, const char *id)
{
 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;
}

struct stasis_state_subscriber *stasis_state_subscribe_pool(struct stasis_state_manager *manager,
 const char *id, stasis_subscription_cb callback, void *data)
{
 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;
}

void *stasis_state_unsubscribe(struct stasis_state_subscriber *sub)
{
 sub->stasis_sub = stasis_unsubscribe(sub->stasis_sub);
 ao2_ref(sub, -1);
 return NULL;
}

void *stasis_state_unsubscribe_and_join(struct stasis_state_subscriber *sub)
{
 if (sub) {
 sub->stasis_sub = stasis_unsubscribe_and_join(sub->stasis_sub);
 ao2_ref(sub, -1);
 }

 return NULL;
}

const char *stasis_state_subscriber_id(const struct stasis_state_subscriber *sub)
{
 return sub->state->id;
}

struct stasis_topic *stasis_state_subscriber_topic(struct stasis_state_subscriber *sub)
{
 return sub->state->topic;
}

void *stasis_state_subscriber_data(struct stasis_state_subscriber *sub)
{
 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;
}

struct stasis_subscription *stasis_state_subscriber_subscription(
 struct stasis_state_subscriber *sub)
{
 return sub->stasis_sub;
}

struct stasis_state_publisher {
 /*! The stasis state to publish to */
 struct stasis_state *state;
};

static void publisher_dtor(void *obj)
{
 struct stasis_state_publisher *pub = obj;

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

struct stasis_state_publisher *stasis_state_add_publisher(
 struct stasis_state_manager *manager, const char *id)
{
 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;
}

const char *stasis_state_publisher_id(const struct stasis_state_publisher *pub)
{
 return pub->state->id;
}

struct stasis_topic *stasis_state_publisher_topic(struct stasis_state_publisher *pub)
{
 return pub->state->topic;
}

void stasis_state_publish(struct stasis_state_publisher *pub, struct stasis_message *msg)
{
 ao2_lock(pub->state);
 ao2_replace(pub->state->msg, msg);
 ao2_unlock(pub->state);

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

/*!
 * \internal
 * \brief Find, or add the given eid to the state object
 *
 * Publishers can be tracked implicitly using eids. This allows us to add, and subsequently
 * remove state objects from the managed states container in a deterministic way. Using the
 * eids in this way is possible because it's guaranteed that there will only ever be a single
 * publisher for a uniquely named topic (topics tracked by this module) on a system.
 *
 * \note The vector does not use locking. Instead we use the state object for that, so it
 * needs to be locked prior to calling this method.
 *
 * \param state The state object
 * \param eid The system id to add to the state object
 */
static void state_find_or_add_eid(struct stasis_state *state, const struct ast_eid *eid)
{
 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);
 }
 }
}

/*!
 * \internal
 * \brief Find, and remove the given eid from the state object
 *
 * Used to remove an eid from an implicit publisher.
 *
 * \note The vector does not use locking. Instead we use the state object for that, so it
 * needs to be locked prior to calling this method.
 *
 * \param state The state object
 * \param eid The system id to remove from the state object
 */
static void state_find_and_remove_eid(struct stasis_state *state, const struct ast_eid *eid)
{
 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;
 }
 }
}

void stasis_state_publish_by_id(struct stasis_state_manager *manager, const char *id,
 const struct ast_eid *eid, struct stasis_message *msg)
{
 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);
}

void stasis_state_remove_publish_by_id(struct stasis_state_manager *manager,
 const char *id, const struct ast_eid *eid, struct stasis_message *msg)
{
 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);
}

int stasis_state_add_observer(struct stasis_state_manager *manager,
 struct stasis_state_observer *observer)
{
 int res;

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

 return res;
}

void stasis_state_remove_observer(struct stasis_state_manager *manager,
 struct stasis_state_observer *observer)
{
 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);
}

static int handle_stasis_state(struct stasis_state *state, on_stasis_state handler, void *data)
{
 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;
}

static int handle_stasis_state_proxy(void *obj, void *arg, void *data, int flags)
{
 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;
}

void stasis_state_callback_all(struct stasis_state_manager *manager, on_stasis_state handler,
 void *data)
{
 ast_assert(handler != NULL);

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

static int handle_stasis_state_subscribed(void *obj, void *arg, void *data, int flags)
{
 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;
}

void stasis_state_callback_subscribed(struct stasis_state_manager *manager, on_stasis_state handler,
 void *data)
{
 ast_assert(handler != NULL);

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