stasis_state.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2019, Sangoma Technologies Corporation
 *
 * Kevin Harwell <kharwell@digium.com>
 *
 * 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) != '\0');
 
    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
 * \param file, line, func
 *
 * \return A stasis_state object or NULL
 * \retval 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
 * \retval NULL on error
 */
#define state_find_or_add(manager, state_topic, id) __state_find_or_add(manager, state_topic, 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);
}