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