[Asterisk-code-review] stasis_state: Add new stasis_state module (...asterisk[master])
Friendly Automation
asteriskteam at digium.com
Tue Jul 2 09:30:35 CDT 2019
Friendly Automation has submitted this change and it was merged. ( https://gerrit.asterisk.org/c/asterisk/+/11462 )
Change subject: stasis_state: Add new stasis_state module
......................................................................
stasis_state: Add new stasis_state module
This new module describes an API that can be thought of as a combination of
stasis topic pools, and caching. Except, hopefully done in a more efficient
and less memory "leaky" manner.
The API defines methods, and data structures for managing, and tracking
published message state through stasis. By adding a subscriber or publisher,
consumers can more easily track the lifetime of the contained state. For
instance, when no more publishers and/or subscribers have need of the topic,
and associated state its data is removed from the managed container.
* 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.
* stasis_state_subscriber *
Topic and state can be created, or referenced within the manager by adding a
stasis_state_subscriber. When adding a subscriber if no state currently exists
new managed state is immediately created. If managed state already exists then
a new subscriber is created referencing that state. The managed state is
guaranteed to live throughout the subscriber's lifetime. State is only removed
from the manager when no other entities require it.
* stasis_state_publisher *
Topic and state can be created, or referenced within the manager by also adding
a stasis_state_publisher. When adding a publisher if no state currently exists
new managed state is created. If managed state already exists then a new
publisher is created referencing that state. The managed state is guaranteed to
live throughout the publisher's lifetime. State is only removed from the
manager when no other entities require it.
* 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.
* other *
Callbacks also exist that allow consumers to iterate over all, or some of the
managed state.
ASTERISK-28442
Change-Id: I7a4a06685a96e511da9f5bd23f9601642d7bd8e5
---
A include/asterisk/stasis_state.h
A main/stasis_state.c
A tests/test_stasis_state.c
3 files changed, 1,806 insertions(+), 0 deletions(-)
Approvals:
Joshua Colp: Looks good to me, but someone else must approve
George Joseph: Looks good to me, approved
Friendly Automation: Approved for Submit
diff --git a/include/asterisk/stasis_state.h b/include/asterisk/stasis_state.h
new file mode 100644
index 0000000..a1d3625
--- /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 at 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 0000000..2a9afcb
--- /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 at 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 0000000..3ad450d
--- /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 at 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");
--
To view, visit https://gerrit.asterisk.org/c/asterisk/+/11462
To unsubscribe, or for help writing mail filters, visit https://gerrit.asterisk.org/settings
Gerrit-Project: asterisk
Gerrit-Branch: master
Gerrit-Change-Id: I7a4a06685a96e511da9f5bd23f9601642d7bd8e5
Gerrit-Change-Number: 11462
Gerrit-PatchSet: 4
Gerrit-Owner: Kevin Harwell <kharwell at digium.com>
Gerrit-Reviewer: Benjamin Keith Ford <bford at digium.com>
Gerrit-Reviewer: Friendly Automation
Gerrit-Reviewer: George Joseph <gjoseph at digium.com>
Gerrit-Reviewer: Joshua Colp <jcolp at digium.com>
Gerrit-Reviewer: Kevin Harwell <kharwell at digium.com>
Gerrit-MessageType: merged
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.digium.com/pipermail/asterisk-code-review/attachments/20190702/5220edb2/attachment-0001.html>
More information about the asterisk-code-review
mailing list