diff --git a/include/asterisk/stasis_state.h b/include/asterisk/stasis_state.h new file mode 100644 index 0000000000000000000000000000000000000000..a1d36253ad8215f7d9f19f2488a710a2c3b903c2 --- /dev/null +++ b/include/asterisk/stasis_state.h @@ -0,0 +1,549 @@ +/* + * 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. + */ + +#ifndef _STASIS_STATE_H +#define _STASIS_STATE_H + +/*! \file + * + * \brief Stasis State API. + * + * \par 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: + * - \ref stasis_state_manager + * - \ref stasis_state_subscriber + * - \ref stasis_state_publisher + * - \ref stasis_state_observer + * + * \par 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 \ref 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. + * + * \par stasis_state_subscriber + * + * As mentioned, topic and state can be created, or referenced within the manager by adding + * a \ref 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. + * + * \par 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 \ref 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 \ref + * 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 \ref 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 \ref ast_eid. However, a + * final call to \ref 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). + * + * \par 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. + */ + +#include "asterisk/stasis.h" + +struct ast_eid; + +/*! + * \brief Manages a collection of stasis states. + * + * Maintains data related to stasis state. Managed state is an association of a unique stasis + * topic (named by a given unique id), and the last known published message. + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_manager; + +/*! + * \brief Create a stasis state manager. + * + * \note The state manager is an ao2_object. When done simply decrement its reference + * for object cleanup. + * + * \param topic_name The name of the topic to create that all state topics + * get forwarded to + * + * \retval A stasis state manager + * \retval NULL if an error occurred + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_manager *stasis_state_manager_create(const char *topic_name); + +/*! + * \brief Retrieve the manager's topic (the topic that all state topics get forwarded to) + * + * \param manager The manager object + * + * \retval The manager's topic. + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_topic *stasis_state_all_topic(struct stasis_state_manager *manager); + +/*! + * \brief 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. + * + * \param manager The manager object + * \param id The unique id of/for the topic + * + * \retval A managed stasis topic. + * \retval NULL if an error occurred + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_topic *stasis_state_topic(struct stasis_state_manager *manager, const char *id); + +/*! + * \brief A stasis state subscriber + * + * A subscriber to a particular stasis state. As such it holds a reference to the + * underlying stasis state, so that managed state is guaranteed to exist for the + * lifetime of the subscriber. + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_subscriber; + +/*! + * \brief 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. + * + * \param manager The manager object + * \param id The unique id of a managed state + * + * \retval A stasis state subscriber + * \retval NULL if an error occurred + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_subscriber *stasis_state_add_subscriber( + struct stasis_state_manager *manager, const char *id); + +/*! + * \brief 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. + * + * \param manager The manager object + * \param id The unique id of a managed state + * \param callback The stasis subscription callback + * \param data A user data object passed to the stasis subscription + * + * \retval A stasis state subscriber + * \retval NULL if an error occurred + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_subscriber *stasis_state_subscribe_pool(struct stasis_state_manager *manager, + const char *id, stasis_subscription_cb callback, void *data); + +/*! + * \brief Unsubscribe from the stasis topic and stasis state. + * + * \param sub A stasis state subscriber + * + * \retval NULL + * + * \since 13.28.0 + * \since 16.5.0 + */ +void *stasis_state_unsubscribe(struct stasis_state_subscriber *sub); + +/*! + * \brief Unsubscribe from the stasis topic, block until the final message + * is received, and then unsubscribe from stasis state. + * + * \param sub A stasis state subscriber + * + * \retval NULL + * + * \since 13.28.0 + * \since 16.5.0 + */ +void *stasis_state_unsubscribe_and_join(struct stasis_state_subscriber *sub); + +/*! + * \brief Retrieve the underlying subscribed to state's unique id + * + * \param sub A stasis state subscriber + * + * \retval The managed state's id + * + * \since 13.28.0 + * \since 16.5.0 + */ +const char *stasis_state_subscriber_id(const struct stasis_state_subscriber *sub); + +/*! + * \brief 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. + * + * \param sub A stasis state subscriber + * + * \retval The subscriber's topic + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_topic *stasis_state_subscriber_topic(struct stasis_state_subscriber *sub); + +/*! + * \brief 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 + * + * \param sub A stasis state subscriber + * + * \retval The subscriber's state message data + * \retval NULL if no data has been published yet + * + * \since 13.28.0 + * \since 16.5.0 + */ +void *stasis_state_subscriber_data(struct stasis_state_subscriber *sub); + +/*! + * \brief Retrieve the stasis topic subscription if available. + * + * \param sub A stasis state subscriber + * + * \retval The subscriber's stasis subscription + * \retval NULL if no subscription available + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_subscription *stasis_state_subscriber_subscription( + struct stasis_state_subscriber *sub); + +/*! + * \brief A stasis state publisher + * + * A publisher to a particular stasis state and topic. As such it holds a reference to + * the underlying stasis state, so that managed state is guaranteed to exist for the + * lifetime of the publisher. + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_publisher; + +/*! + * \brief 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. + * + * \param manager The manager object + * \param id The unique id of a managed state + * + * \retval A stasis state publisher + * \retval NULL if an error occurred + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_state_publisher *stasis_state_add_publisher( + struct stasis_state_manager *manager, const char *id); + +/*! + * \brief Retrieve the publisher's underlying state's unique id + * + * \param pub A stasis state publisher + * + * \retval The managed state's id + * + * \since 13.28.0 + * \since 16.5.0 + */ +const char *stasis_state_publisher_id(const struct stasis_state_publisher *pub); + +/*! + * \brief 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. + * + * \param pub A stasis state publisher + * + * \retval The publisher's topic + * + * \since 13.28.0 + * \since 16.5.0 + */ +struct stasis_topic *stasis_state_publisher_topic(struct stasis_state_publisher *pub); + +/*! + * \brief Publish to a managed state (topic) using a publisher. + * + * \param pub The publisher to use to publish the message + * \param msg The message to publish + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_publish(struct stasis_state_publisher *pub, struct stasis_message *msg); + +/*! + * \brief 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. + * + * \param manager The state manager + * \param id A state's unique id + * \param eid The unique system id + * \param msg The message to publish + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_publish_by_id(struct stasis_state_manager *manager, const char *id, + const struct ast_eid *eid, struct stasis_message *msg); + +/*! + * \brief 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. + * + * \param manager The state manager + * \param id A state's unique id + * \param eid The unique system id + * \param msg The message to publish (can be NULL) + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_remove_publish_by_id(struct stasis_state_manager *manager, + const char *id, const struct ast_eid *eid, struct stasis_message *msg); + +/*! \brief Managed stasis state event interface */ +struct stasis_state_observer { + /*! + * \brief Raised when any managed state is being subscribed. + * + * \param id The unique id of the managed state + * \param sub The subscriber subscribed + */ + void (*on_subscribe)(const char *id, struct stasis_state_subscriber *sub); + + /*! + * \brief Raised when any managed state is being unsubscribed. + * + * \param id The unique id of the managed state + * \param sub The subscriber to unsubscribe + */ + void (*on_unsubscribe)(const char *id, struct stasis_state_subscriber *sub); +}; + +/*! + * \brief Add an observer to receive managed state related events. + * + * \param manager The state manager + * \param observer The observer handling events + * + * \retval 0 if successfully registered + * \retval -1 on failure + * + * \since 13.28.0 + * \since 16.5.0 + */ +int stasis_state_add_observer(struct stasis_state_manager *manager, + struct stasis_state_observer *observer); + +/*! + * \brief Remove an observer (will no longer receive managed state related events). + * + * \param manager The state manager + * \param observer The observer being removed + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_remove_observer(struct stasis_state_manager *manager, + struct stasis_state_observer *observer); + +/*! + * \brief The delegate called for each managed state. + * + * \param id The unique id of a managed state object + * \param msg The last published message on the state, or NULL + * \param user_data Data object the user passed into the manager callback + * + * \retval 0 to continue traversing + * \retval CMP_STOP (2) to stop traversing + * + * \since 13.28.0 + * \since 16.5.0 + */ +typedef int (*on_stasis_state)(const char *id, struct stasis_message *msg, void *user_data); + +/*! + * \brief For each managed state call the given handler. + * + * \param manager The state manager + * \param handler The handler to call for each managed state + * \param data User to data to pass on to the handler + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_callback_all(struct stasis_state_manager *manager, on_stasis_state handler, + void *data); + +/*! + * \brief For each managed, and explicitly subscribed state call the given handler. + * + * \param manager The state manager + * \param handler The handler to call for each managed state + * \param data User to data to pass on to the handler + * + * \since 13.28.0 + * \since 16.5.0 + */ +void stasis_state_callback_subscribed(struct stasis_state_manager *manager, on_stasis_state handler, + void *data); + +#endif /* _STASIS_STATE_H */ diff --git a/main/stasis_state.c b/main/stasis_state.c new file mode 100644 index 0000000000000000000000000000000000000000..2a9afcb717fc3b320cf0bf1ace9633a7ee25bc52 --- /dev/null +++ b/main/stasis_state.c @@ -0,0 +1,791 @@ +/* + * 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 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; + /*! The manager that owns and handles this state */ + 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[0]; +}; + +AO2_STRING_FIELD_HASH_FN(stasis_state, id); +AO2_STRING_FIELD_CMP_FN(stasis_state, 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; + ao2_cleanup(state->manager); + state->manager = NULL; + + /* All eids should have been removed */ + ast_assert(AST_VECTOR_SIZE(&state->eids) == 0); + AST_VECTOR_FREE(&state->eids); +} + +/*! + * \internal + * \brief Allocate a stasis state object. + * + * 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 + */ +static struct stasis_state *state_alloc(struct stasis_state_manager *manager, + struct stasis_topic *state_topic, const char *id) +{ + struct stasis_state *state; + + if (!state_topic) { + char *name; + + /* If not given a state topic, then an id is required */ + ast_assert(id != NULL); + + /* + * 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) { + ast_log(LOG_ERROR, "Unable to create state topic name '%s/%s'\n", + stasis_topic_name(manager->all_topic), id); + return NULL; + } + + state_topic = stasis_topic_create(name); + + if (!state_topic) { + ast_log(LOG_ERROR, "Unable to create state topic '%s'\n", name); + ast_free(name); + return NULL; + } + ast_free(name); + } + + 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); + } + + /* + * Since the state topic could have been 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 = ao2_alloc(sizeof(*state) + strlen(id) + 1, state_dtor); + if (!state) { + ast_log(LOG_ERROR, "Unable to allocate state '%s' in manager '%s'\n", + id, stasis_topic_name(manager->all_topic)); + ao2_ref(state_topic, -1); + return NULL; + } + + strcpy(state->id, id); /* Safe */ + state->topic = state_topic; /* ref already bumped above */ + + state->forward = stasis_forward_all(state->topic, manager->all_topic); + if (!state->forward) { + ast_log(LOG_ERROR, "Unable to add state '%s' forward in manager '%s'\n", + id, stasis_topic_name(manager->all_topic)); + ao2_ref(state, -1); + return NULL; + } + + if (AST_VECTOR_INIT(&state->eids, 2)) { + ast_log(LOG_ERROR, "Unable to initialize eids for state '%s' in manager '%s'\n", + id, stasis_topic_name(manager->all_topic)); + ao2_ref(state, -1); + return NULL; + } + + state->manager = ao2_bump(manager); + + return state; +} + +/*! + * \internal + * \brief Create a state object, and add it to the manager. + * + * \note Locking on the states container is specifically not done here, thus + * appropriate locks should be applied prior to this function being called. + * + * \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 + */ +static struct stasis_state *state_add(struct stasis_state_manager *manager, + struct stasis_topic *state_topic, const char *id) +{ + struct stasis_state *state = state_alloc(manager, state_topic, id); + + if (!state) { + return NULL; + } + + if (!ao2_link_flags(manager->states, state, OBJ_NOLOCK)) { + ast_log(LOG_ERROR, "Unable to add state '%s' to manager '%s'\n", + state->id ? state->id : "", stasis_topic_name(manager->all_topic)); + ao2_ref(state, -1); + return NULL; + } + + return state; +} + +/*! + * \internal + * \brief Find a state by id, or create one if not found and add it to the manager. + * + * \note Locking on the states container is specifically not done here, thus + * appropriate locks should be applied prior to this function being called. + * + * \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 + */ +static struct stasis_state *state_find_or_add(struct stasis_state_manager *manager, + struct stasis_topic *state_topic, const char *id) +{ + struct stasis_state *state; + + if (ast_strlen_zero(id)) { + id = state_id_by_topic(manager->all_topic, state_topic); + } + + state = ao2_find(manager->states, id, OBJ_SEARCH_KEY | OBJ_NOLOCK); + + return state ? state : state_add(manager, state_topic, id); +} + +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_hash_fn, NULL, stasis_state_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; + + ao2_lock(manager->states); + state = state_find_or_add(manager, NULL, id); + ao2_unlock(manager->states); + + if (!state) { + return NULL; + } + + topic = state->topic; + ao2_ref(state, -1); + return topic; +} + +/*! + * \internal + * \brief Remove a state from the stasis manager + * + * State should only be removed from the manager under the following conditions: + * + * There are no more subscribers to it + * There are no more explicit publishers publishing to it + * There are no more implicit publishers publishing to it + * + * Subscribers and explicit publishers hold a reference to the state object itself, so + * once a state's reference count drops to 2 (1 for the manager, 1 passed in) then we + * know there are no more subscribers or explicit publishers. Implicit publishers are + * tracked by eids, so once that container is empty no more implicit publishers exist + * for the state either. Only then can a state be removed. + * + * \param state The state to remove + */ +static void state_remove(struct stasis_state *state) +{ + ao2_lock(state); + + /* + * The manager's state container also needs to be locked here prior to checking + * the state's reference count, and potentially removing since we don't want its + * count to possibly increase between the check and unlinking. + */ + ao2_lock(state->manager->states); + + /* + * If there are only 2 references left then it's the one owned by the manager, + * and the one passed in to this function. However, before removing it from the + * manager we need to also check that no eid is associated with the given state. + * If an eid still remains then this means that an implicit publisher is still + * publishing to this state. + */ + if (ao2_ref(state, 0) == 2 && AST_VECTOR_SIZE(&state->eids) == 0) { + ao2_unlink_flags(state->manager->states, state, 0); + } + + ao2_unlock(state->manager->states); + ao2_unlock(state); + + /* Now it's safe to remove the reference that is held on the given state */ + ao2_ref(state, -1); +} + +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); + + state_remove(sub->state); +} + +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; + } + + ao2_lock(manager->states); + sub->state = state_find_or_add(manager, NULL, id); + if (!sub->state) { + ao2_unlock(manager->states); + ao2_ref(sub, -1); + return NULL; + } + ao2_unlock(manager->states); + + 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) +{ + 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; + + state_remove(pub->state); +} + +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; + } + + ao2_lock(manager->states); + pub->state = state_find_or_add(manager, NULL, id); + if (!pub->state) { + ao2_unlock(manager->states); + ao2_ref(pub, -1); + return NULL; + } + ao2_unlock(manager->states); + + 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)) { + AST_VECTOR_APPEND(&state->eids, *eid); + } +} + +/*! + * \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); + 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; + + ao2_lock(manager->states); + state = state_find_or_add(manager, NULL, id); + ao2_unlock(manager->states); + + 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_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); + + state_remove(state); +} + +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(void *obj, void *arg, void *data, int flags) +{ + struct stasis_state *state = obj; + on_stasis_state handler = arg; + 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; +} + +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, handler, data); +} + +static int handle_stasis_state_subscribed(void *obj, void *arg, void *data, int flags) +{ + struct stasis_state *state = obj; + + if (state->num_subscribers) { + return handle_stasis_state(obj, arg, data, flags); + } + + return 0; +} + +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); +} diff --git a/tests/test_stasis_state.c b/tests/test_stasis_state.c new file mode 100644 index 0000000000000000000000000000000000000000..3ad450d111100cccb601a90786815e2b3476d956 --- /dev/null +++ b/tests/test_stasis_state.c @@ -0,0 +1,466 @@ +/* + * 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 + <depend>TEST_FRAMEWORK</depend> + <support_level>core</support_level> + ***/ + +#include "asterisk.h" + +#include "asterisk/astobj2.h" +#include "asterisk/conversions.h" +#include "asterisk/module.h" +#include "asterisk/stasis_state.h" +#include "asterisk/test.h" + +#define test_category "/stasis/core/state/" + +#define TOPIC_COUNT 500 + +#define MANAGER_TOPIC "foo" + +struct stasis_message_type *foo_type(void); + +/*! foo stasis message type */ +STASIS_MESSAGE_TYPE_DEFN(foo_type); + +/*! foo_type data */ +struct foo_data { + size_t bar; +}; + +AST_VECTOR(subscriptions, struct stasis_state_subscriber *); +AST_VECTOR(publishers, struct stasis_state_publisher *); + +/*! + * For testing purposes each subscribed state's id is a number. This value is + * the summation of all id's. + */ +static size_t sum_total; + +/*! Test variable that tracks the running total of state ids */ +static size_t running_total; + +/*! This value is set to check if state data is NULL before publishing */ +static int expect_null; + +static int validate_data(const char *id, struct foo_data *foo) +{ + size_t num; + + if (ast_str_to_umax(id, &num)) { + ast_log(LOG_ERROR, "Unable to convert the state's id '%s' to numeric\n", id); + return -1; + } + + running_total += num; + + if (!foo) { + if (expect_null) { + return 0; + } + + ast_log(LOG_ERROR, "Expected state data for '%s'\n", id); + return -1; + } + + if (expect_null) { + ast_log(LOG_ERROR, "Expected NULL state data for '%s'\n", id); + return -1; + } + + if (foo->bar != num) { + ast_log(LOG_ERROR, "Unexpected state data for '%s'\n", id); + return -1; + } + + return 0; +} + +static void handle_validate(const char *id, struct stasis_state_subscriber *sub) +{ + struct foo_data *foo = stasis_state_subscriber_data(sub); + validate_data(id, foo); + ao2_cleanup(foo); +} + +struct stasis_state_observer foo_observer = { + .on_subscribe = handle_validate, + .on_unsubscribe = handle_validate +}; + +static void foo_type_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message) +{ + /* No op since we are not really testing stasis topic handling here */ +} + +static int subscriptions_destroy(struct stasis_state_manager *manager, struct subscriptions *subs) +{ + running_total = expect_null = 0; + + AST_VECTOR_CALLBACK_VOID(subs, stasis_state_unsubscribe_and_join); + AST_VECTOR_FREE(subs); + + stasis_state_remove_observer(manager, &foo_observer); + + if (running_total != sum_total) { + ast_log(LOG_ERROR, "Failed to destroy all subscriptions: running=%zu, sum=%zu\n", + running_total, sum_total); + return -1; + } + + return 0; +} + +static int subscriptions_create(struct stasis_state_manager *manager, + struct subscriptions *subs) +{ + size_t i; + + if (stasis_state_add_observer(manager, &foo_observer) || + AST_VECTOR_INIT(subs, TOPIC_COUNT)) { + return -1; + } + + sum_total = running_total = 0; + expect_null = 1; + + for (i = 0; i < TOPIC_COUNT; ++i) { + struct stasis_state_subscriber *sub; + char id[32]; + + if (snprintf(id, 10, "%zu", i) == -1) { + ast_log(LOG_ERROR, "Unable to convert subscriber id to string\n"); + break; + } + + sub = stasis_state_subscribe_pool(manager, id, foo_type_cb, NULL); + if (!sub) { + ast_log(LOG_ERROR, "Failed to create a state subscriber for id '%s'\n", id); + ao2_ref(sub, -1); + break; + } + + if (AST_VECTOR_APPEND(subs, sub)) { + ast_log(LOG_ERROR, "Failed to add to foo_sub to vector for id '%s'\n", id); + ao2_ref(sub, -1); + break; + } + + sum_total += i; + } + + if (i != TOPIC_COUNT || running_total != sum_total) { + ast_log(LOG_ERROR, "Failed to create all subscriptions: running=%zu, sum=%zu\n", + running_total, sum_total); + subscriptions_destroy(manager, subs); + return -1; + } + + return 0; +} + +static int publishers_destroy(struct stasis_state_manager *manager, struct publishers *pubs) +{ + size_t i; + + if (pubs) { + /* Remove explicit publishers */ + AST_VECTOR_CALLBACK_VOID(pubs, ao2_cleanup); + AST_VECTOR_FREE(pubs); + return 0; + } + + for (i = 0; i < TOPIC_COUNT; ++i) { + char id[32]; + + /* Remove implicit publishers */ + if (snprintf(id, 10, "%zu", i) == -1) { + ast_log(LOG_ERROR, "Unable to convert publisher id to string\n"); + return -1; + } + + stasis_state_remove_publish_by_id(manager, id, NULL, NULL); + } + + return 0; +} + +static int publishers_create(struct stasis_state_manager *manager, + struct publishers *pubs) +{ + size_t i; + + if (AST_VECTOR_INIT(pubs, TOPIC_COUNT)) { + return -1; + } + + for (i = 0; i < TOPIC_COUNT; ++i) { + struct stasis_state_publisher *pub; + char id[32]; + + if (snprintf(id, 10, "%zu", i) == -1) { + ast_log(LOG_ERROR, "Unable to convert publisher id to string\n"); + break; + } + + /* Create the state publisher */ + pub = stasis_state_add_publisher(manager, id); + if (!pub) { + ast_log(LOG_ERROR, "Failed to create a state publisher for id '%s'\n", id); + break; + } + + if (AST_VECTOR_APPEND(pubs, pub)) { + ast_log(LOG_ERROR, "Failed to add to publisher to vector for id '%s'\n", id); + ao2_ref(pub, -1); + break; + } + } + + if (i != TOPIC_COUNT) { + ast_log(LOG_ERROR, "Failed to create all publishers: count=%zu\n", i); + publishers_destroy(manager, pubs); + return -1; + } + + return 0; +} + +static struct stasis_message *create_foo_type_message(const char *id) +{ + struct stasis_message *msg; + struct foo_data *foo; + + foo = ao2_alloc(sizeof(*foo), NULL); + if (!foo) { + ast_log(LOG_ERROR, "Failed to allocate foo data for '%s'\n", id); + return NULL; + } + + if (ast_str_to_umax(id, &foo->bar)) { + ast_log(LOG_ERROR, "Unable to convert the state's id '%s' to numeric\n", id); + ao2_ref(foo, -1); + return NULL; + } + + msg = stasis_message_create_full(foo_type(), foo, NULL); + if (!msg) { + ast_log(LOG_ERROR, "Failed to create stasis message for '%s'\n", id); + } + + ao2_ref(foo, -1); + return msg; +} + +static int implicit_publish_cb(const char *id, struct stasis_message *msg, void *user_data) +{ + /* For each state object create and publish new state data */ + struct foo_data *foo = stasis_message_data(msg); + + if (validate_data(id, foo)) { + return CMP_STOP; + } + + msg = create_foo_type_message(id); + if (!msg) { + return CMP_STOP; + } + + /* Now publish it on the managed state object */ + stasis_state_publish_by_id(user_data, id, NULL, msg); + ao2_ref(msg, -1); + + return 0; +} + +static int explicit_publish_cb(const char *id, struct stasis_message *msg, void *user_data) +{ + /* For each state object create and publish new state data */ + struct publishers *pubs = user_data; + struct stasis_state_publisher *pub = NULL; + struct foo_data *foo = stasis_message_data(msg); + size_t i; + + if (validate_data(id, foo)) { + return CMP_STOP; + } + + msg = create_foo_type_message(id); + if (!msg) { + return CMP_STOP; + } + + for (i = 0; i < AST_VECTOR_SIZE(pubs); ++i) { + if (!strcmp(stasis_state_publisher_id(AST_VECTOR_GET(pubs, i)), id)) { + pub = AST_VECTOR_GET(pubs, i); + break; + } + } + + if (!pub) { + ast_log(LOG_ERROR, "Unable to locate publisher for id '%s'\n", id); + return CMP_STOP; + } + + stasis_state_publish(pub, msg); + ao2_ref(msg, -1); + + return 0; +} + +static int publish(struct stasis_state_manager *manager, on_stasis_state cb, + void *user_data) +{ + /* First time there is no state data */ + expect_null = 1; + + running_total = 0; + stasis_state_callback_all(manager, cb, user_data); + + if (running_total != sum_total) { + ast_log(LOG_ERROR, "Failed manager_callback (1): running=%zu, sum=%zu\n", + running_total, sum_total); + return -1; + } + + /* Second time check valid state data exists */ + running_total = expect_null = 0; + stasis_state_callback_all(manager, cb, user_data); + + if (running_total != sum_total) { + ast_log(LOG_ERROR, "Failed manager_callback (2): running=%zu, sum=%zu\n", + running_total, sum_total); + return -1; + } + + return 0; +} + +AST_TEST_DEFINE(implicit_publish) +{ + RAII_VAR(struct stasis_state_manager *, manager, NULL, ao2_cleanup); + struct subscriptions subs; + int rc = AST_TEST_PASS; + + switch (cmd) { + case TEST_INIT: + info->name = __func__; + info->category = test_category; + info->summary = "Test implicit publishing of stasis state"; + info->description = info->summary; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + manager = stasis_state_manager_create(MANAGER_TOPIC); + ast_test_validate(test, manager != NULL); + + ast_test_validate(test, !subscriptions_create(manager, &subs)); + + ast_test_validate_cleanup(test, !publish(manager, implicit_publish_cb, manager), + rc, cleanup); + +cleanup: + if (subscriptions_destroy(manager, &subs) || publishers_destroy(manager, NULL)) { + return AST_TEST_FAIL; + } + + /* + * State subscriptions add a ref a state. The state in turn adds a ref + * to the manager. So if more than one ref is held on the manager before + * exiting, there is a ref leak some place. + */ + if (ao2_ref(manager, 0) != 1) { + ast_log(LOG_ERROR, "Memory leak - Too many references held on manager\n"); + return AST_TEST_FAIL; + } + + return rc; +} + +AST_TEST_DEFINE(explicit_publish) +{ + RAII_VAR(struct stasis_state_manager *, manager, NULL, ao2_cleanup); + struct subscriptions subs; + struct publishers pubs; + int rc = AST_TEST_PASS; + + switch (cmd) { + case TEST_INIT: + info->name = __func__; + info->category = test_category; + info->summary = "Test explicit publishing of stasis state"; + info->description = info->summary; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + manager = stasis_state_manager_create(MANAGER_TOPIC); + ast_test_validate(test, manager != NULL); + + ast_test_validate(test, !subscriptions_create(manager, &subs)); + ast_test_validate_cleanup(test, !publishers_create(manager, &pubs), rc, cleanup); + + ast_test_validate_cleanup(test, !publish(manager, explicit_publish_cb, &pubs), + rc, cleanup); + +cleanup: + if (subscriptions_destroy(manager, &subs) || publishers_destroy(manager, &pubs)) { + return AST_TEST_FAIL; + } + + /* + * State subscriptions add a ref a state. The state in turn adds a ref + * to the manager. So if more than one ref is held on the manager before + * exiting, there is a ref leak some place. + */ + if (ao2_ref(manager, 0) != 1) { + ast_log(LOG_ERROR, "Memory leak - Too many references held on manager\n"); + return AST_TEST_FAIL; + } + + return rc; +} + +static int unload_module(void) +{ + AST_TEST_UNREGISTER(implicit_publish); + AST_TEST_UNREGISTER(explicit_publish); + + STASIS_MESSAGE_TYPE_CLEANUP(foo_type); + + return 0; +} + +static int load_module(void) +{ + if (STASIS_MESSAGE_TYPE_INIT(foo_type) != 0) { + return -1; + } + + AST_TEST_REGISTER(implicit_publish); + AST_TEST_REGISTER(explicit_publish); + + return AST_MODULE_LOAD_SUCCESS; +} + +AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Stasis state testing");