Merge "mwi: Update the MWI core to use stasis_state API"

include/asterisk/mwi.h

@@ -19,7 +19,77 @@
 #ifndef _ASTERISK_MWI_H
 #define _ASTERISK_MWI_H
 
+/*! \file
+ *
+ * \brief Asterisk MWI API.
+ *
+ * \par Intro
+ *
+ * This module manages, and processes all things MWI. Defined are mechanisms for subscribing
+ * and publishing to MWI topics. User modules wishing to receive MWI updates for a particular
+ * mailbox should do so by adding an MWI subscriber to that mailbox, followed by subscribing
+ * to the mailbox's topic. Likewise, user modules that want to publish MWI updates about a
+ * particular mailbox should first add a publisher for that mailbox prior to publishing.
+ *
+ * MWI state is managed via an underlying \ref stasis_state_manager (if interested see the
+ * stasis_state.c module for the gory details). As such all last known mailbox state can be
+ * retrieve and iterated over by using the \ref ast_mwi_callback function.
+ *
+ * \par ast_mwi_subscriber
+ *
+ * Created via \ref ast_mwi_add_subscriber, a subscriber subscribes to a given mailbox in
+ * order to receive updates about the given mailbox. Adding a subscriber will create the
+ * underlying topic, and associated state data if those do not already exist for it. The
+ * topic, and last known state data is guaranteed to exist for the lifetime of the subscriber.
+ * State data can be NULL if nothing has been published to the mailbox's topic yet.
+ *
+ * NOTE, currently adding a subscriber here will either create, or add a reference to the
+ * underlying stasis state (and associated topic). However, it does not actually subscribe to
+ * the stasis topic itself. You still need to explicitly call \ref stasis_subscribe, or
+ * similar on the topic if you wish to receive published event updates.
+ *
+ * So given that when subscribing to an MWI topic the following order should be adhered to:
+ *
+ *   1. Add an MWI state subscriber using \ref ast_mwi_add_subscriber
+ *   2. Retrieve the topic from the subscriber using \ref ast_mwi_subscriber_topic
+ *   3. Subscribe to the topic itself using \ref stasis_subscribe or \ref stasis_subscribe_pool
+ *
+ * Or simply call \ref ast_mwi_subscribe_pool, which combines those steps into a single call and
+ * returns the subscriber that is now subscribed to both the stasis topic and state.
+ *
+ * Similarly, releasing the subscriber's reference removes a reference to the underlying state,
+ * but does not unsubscribe from the MWI topic. This should be done separately and prior to
+ * removing the subscriber's state reference:
+ *
+ *   1. Unsubscribe from the stasis topic subscription using \ref stasis_unsubscribe or
+ *      \ref stasis_unsubscribe_and_join
+ *   2. Remove the MWI subscriber reference
+ *
+ * Or call \ref ast_mwi_unsubscribe (or _and_join), which combines those two steps into a
+ * single call.
+ *
+ * \par ast_mwi_publisher
+ *
+ * Before publishing to a particular topic a publisher should be created. This can be achieved
+ * by using \ref ast_mwi_add_publisher. Publishing to a mailbox should then be done using the
+ * \ref ast_mwi_publish function. This ensures the message is published to the appropriate
+ * topic, and the last known state is maintained.
+ *
+ * Publishing by mailbox id alone is also allowed. However, it is not recommended to do so,
+ * and exists mainly for backwards compatibility, and legacy subsystems. If, and when this
+ * method of publishing is employed a call to one of the \ref ast_delete_mwi_state functions
+ * should also be called for a given mailbox id after no more publishing will be done for
+ * that id. Otherwise a memory leak on the underlying stasis_state object will occur.
+ *
+ * \par ast_mwi_observer
+ *
+ * Add an observer in order to watch for particular MWI module related events. For instance if
+ * a submodule needs to know when a subscription is added to any mailbox an observer can be
+ * added to watch for that.
+ */
+
 #include "asterisk/utils.h"
+#include "asterisk/stasis_state.h"
 
 #if defined(__cplusplus) || defined(c_plusplus)
 extern "C" {
@@ -27,6 +97,273 @@ extern "C" {
 
 struct ast_json;
 struct stasis_message_type;
+struct ast_mwi_state;
+
+/*!
+ * \brief An MWI state subscriber
+ *
+ * An ao2 managed object. Holds a reference to the latest MWI state for its lifetime.
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_subscriber;
+
+/*!
+ * \brief Add an MWI state subscriber to the mailbox
+ *
+ * Adding a subscriber to a mailbox will create a stasis topic for the mailbox if one
+ * does not already exist. It does not however subscribe to the topic itself. This is
+ * done separately using a call to \ref stasis_subscribe or \ref stasis_subscribe_pool.
+ *
+ * A subscriber can be removed by releasing its reference. Doing so releases its underlying
+ * reference to the MWI state. It does not unsubscribe from the topic. Unsubscribing from
+ * a topic should be done prior to unsubscribing the state.
+ *
+ * \param mailbox The subscription state mailbox id
+ *
+ * \retval An MWI subscriber object
+ * \retval NULL on error
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_subscriber *ast_mwi_add_subscriber(const char *mailbox);
+
+/*!
+ * \brief Add an MWI state subscriber, and stasis subscription to the mailbox
+ *
+ * Adding a subscriber to a mailbox will create a stasis topic for the mailbox if one
+ * does not already exist. Once successfully create the underlying stasis topic is then
+ * subscribed to as well.
+ *
+ * A subscriber can be removed by releasing its reference. Doing so releases its underlying
+ * reference to the MWI state. It does not unsubscribe from the topic. Unsubscribing from
+ * a topic should be done prior to unsubscribing the state.
+ *
+ * \param mailbox The subscription state mailbox id
+ * \param callback The stasis subscription callback
+ * \param data A user data object passed to the stasis subscription
+ *
+ * \retval An MWI subscriber object
+ * \retval NULL on error
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_subscriber *ast_mwi_subscribe_pool(const char *mailbox,
+	stasis_subscription_cb callback, void *data);
+
+/*!
+ * \brief Unsubscribe from the stasis topic and MWI.
+ *
+ * \param sub An MWI subscriber
+ *
+ * \retval NULL
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+void *ast_mwi_unsubscribe(struct ast_mwi_subscriber *sub);
+
+/*!
+ * \brief Unsubscribe from the stasis topic, block until the final message
+ * is received, and then unsubscribe from MWI.
+ *
+ * \param sub An MWI subscriber
+ *
+ * \retval NULL
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+void *ast_mwi_unsubscribe_and_join(struct ast_mwi_subscriber *sub);
+
+/*!
+ * \brief Retrieves the MWI 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 An MWI subscriber
+ *
+ * \retval A stasis topic subscribed to by the subscriber
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct stasis_topic *ast_mwi_subscriber_topic(struct ast_mwi_subscriber *sub);
+
+/*!
+ * \brief Retrieves the state data object associated with the MWI subscriber
+ *
+ * \note Returned data's reference count is incremented
+ *
+ * \param sub An MWI subscriber
+ *
+ * \retval The state data object
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_state *ast_mwi_subscriber_data(struct ast_mwi_subscriber *sub);
+
+/*!
+ * \brief Retrieve the stasis MWI topic subscription if available.
+ *
+ * \param sub An MWI subscriber
+ *
+ * \retval The subscriber's stasis subscription
+ * \retval NULL if no subscription available
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct stasis_subscription *ast_mwi_subscriber_subscription(struct ast_mwi_subscriber *sub);
+
+/*!
+ * \brief An MWI state publisher
+ *
+ * An ao2 managed object. Holds a reference to the latest MWI state for its lifetime.
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_publisher;
+
+/*!
+ * \brief Add an MWI state publisher to the mailbox
+ *
+ * Adding a publisher to a mailbox will create a stasis topic for the mailbox if one
+ * does not already exist. A publisher can be removed by releasing its reference. Doing
+ * so releases its underlying reference to the MWI state.
+ *
+ * \param mailbox The mailbox id to publish to
+ *
+ * \retval An MWI publisher object
+ * \retval NULl on error
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+struct ast_mwi_publisher *ast_mwi_add_publisher(const char *mailbox);
+
+/*! \brief MWI state event interface */
+struct ast_mwi_observer {
+	/*!
+	 * \brief Raised when MWI is being subscribed
+	 *
+	 * \param mailbox The mailbox id subscribed
+	 * \param sub The subscriber subscribed
+	 */
+	void (*on_subscribe)(const char *mailbox, struct ast_mwi_subscriber *sub);
+
+	/*!
+	 * \brief Raised when MWI is being unsubscribed
+	 *
+	 * \param mailbox The mailbox id being unsubscribed
+	 * \param sub The subscriber to unsubscribe
+	 */
+	void (*on_unsubscribe)(const char *mailbox, struct ast_mwi_subscriber *sub);
+};
+
+/*!
+ * \brief Add an observer to receive MWI state related events.
+ *
+ * \param observer The observer handling events
+ *
+ * \retval 0 if successfully registered, -1 otherwise
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+int ast_mwi_add_observer(struct ast_mwi_observer *observer);
+
+/*!
+ * \brief Remove an MWI state observer.
+ *
+ * \param observer The observer being removed
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+void ast_mwi_remove_observer(struct ast_mwi_observer *observer);
+
+/*!
+ * \brief The delegate called for each managed mailbox state.
+ *
+ * \param mwi_state The mwi state object
+ * \param data User data passed in by the initiator
+ *
+ * \retval 0 to continue traversing, or CMP_STOP (2) to stop traversing
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+typedef int (*on_mwi_state)(struct ast_mwi_state *mwi_state, void *data);
+
+/*!
+ * \brief For each managed mailbox call the given handler.
+ *
+ * \param handler The mwi state handler to call for each managed mailbox
+ * \param data User to data to pass on to the handler
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+void ast_mwi_state_callback_all(on_mwi_state handler, void *data);
+
+/*!
+ * \brief For each managed mailbox that has a subscriber call the given handler.
+ *
+ * \param handler The mwi state handler to call for each managed mailbox
+ * \param data User to data to pass on to the handler
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+void ast_mwi_state_callback_subscribed(on_mwi_state handler, void *data);
+
+/*!
+ * \brief Publish MWI for the given mailbox.
+ *
+ * \param publisher The publisher to publish a mailbox update on
+ * \param urgent_msgs The number of urgent messages in this mailbox
+ * \param new_msgs The number of new messages in this mailbox
+ * \param old_msgs The number of old messages in this mailbox
+ * \param channel_id A unique identifier for a channel associated with this
+ *        change in mailbox state
+ * \param eid The EID of the server that originally published the message
+ *
+ * \retval 0 on success
+ * \retval -1 on failure
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+int ast_mwi_publish(struct ast_mwi_publisher *publisher, int urgent_msgs,
+	int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid);
+
+/*!
+ * \brief Publish MWI for the given mailbox.
+ *
+ * \param mailbox The mailbox identifier string.
+ * \param context The context this mailbox resides in (NULL or "" if only using mailbox)
+ * \param urgent_msgs The number of urgent messages in this mailbox
+ * \param new_msgs The number of new messages in this mailbox
+ * \param old_msgs The number of old messages in this mailbox
+ * \param channel_id A unique identifier for a channel associated with this
+ *        change in mailbox state
+ * \param eid The EID of the server that originally published the message
+ *
+ * \retval 0 on success
+ * \retval -1 on failure
+ *
+ * \since 13.28.0
+ * \since 16.5.0
+ */
+int ast_mwi_publish_by_mailbox(const char *mailbox, const char *context, int urgent_msgs,
+	int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid);
 
 /*!
  * \since 12
@@ -126,6 +463,7 @@ struct ast_mwi_state {
 	/*! If applicable, a snapshot of the channel that caused this MWI change */
 	struct ast_channel_snapshot *snapshot;
 	struct ast_eid eid;              /*!< The EID of the server where this message originated */
+	int urgent_msgs;                 /*!< The current number of urgent messages for this mailbox */
 };
 
 /*!
@@ -228,7 +566,7 @@ struct stasis_message_type *ast_mwi_vm_app_type(void);
  * \retval 0 Success
  * \retval -1 Failure
  *
- * \since 13.26.0
+ * \since 13.27.0
  * \since 16.4.0
  */
 int mwi_init(void);

main/mwi.c

@@ -22,16 +22,16 @@
 
 #include "asterisk.h"
 
+#include "asterisk/app.h"
 #include "asterisk/mwi.h"
 #include "asterisk/stasis_channels.h"
 
 /*
  * @{ \brief Define \ref stasis topic objects
  */
-static struct stasis_topic *mwi_topic_all;
+static struct stasis_state_manager *mwi_state_manager;
 static struct stasis_cache *mwi_state_cache;
 static struct stasis_caching_topic *mwi_topic_cached;
-static struct stasis_topic_pool *mwi_topic_pool;
 /* @} */
 
 /*! \brief Convert a MWI \ref stasis_message to a \ref ast_event */
@@ -84,7 +84,7 @@ static void mwi_state_dtor(void *obj)
 
 struct stasis_topic *ast_mwi_topic_all(void)
 {
-	return mwi_topic_all;
+	return stasis_state_all_topic(mwi_state_manager);
 }
 
 struct stasis_cache *ast_mwi_state_cache(void)
@@ -99,10 +99,11 @@ struct stasis_topic *ast_mwi_topic_cached(void)
 
 struct stasis_topic *ast_mwi_topic(const char *uniqueid)
 {
-	return stasis_topic_pool_get_topic(mwi_topic_pool, uniqueid);
+	return stasis_state_topic(mwi_state_manager, uniqueid);
 }
 
-struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
+static struct ast_mwi_state *mwi_create_state(const char *mailbox, const char *context,
+	int urgent_msgs, int new_msgs, int old_msgs)
 {
 	struct ast_mwi_state *mwi_state;
 
@@ -110,10 +111,14 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
 
 	mwi_state = ao2_alloc(sizeof(*mwi_state), mwi_state_dtor);
 	if (!mwi_state) {
+		ast_log(LOG_ERROR, "Unable to create MWI state for mailbox '%s@%s'\n",
+				mailbox, ast_strlen_zero(context) ? "" : context);
 		return NULL;
 	}
 
 	if (ast_string_field_init(mwi_state, 256)) {
+		ast_log(LOG_ERROR, "Unable to initialize MWI state for mailbox '%s@%s'\n",
+				mailbox, ast_strlen_zero(context) ? "" : context);
 		ao2_ref(mwi_state, -1);
 		return NULL;
 	}
@@ -123,9 +128,28 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
 		ast_string_field_set(mwi_state, uniqueid, mailbox);
 	}
 
+	mwi_state->urgent_msgs = urgent_msgs;
+	mwi_state->new_msgs = new_msgs;
+	mwi_state->old_msgs = old_msgs;
+
 	return mwi_state;
 }
 
+static struct ast_mwi_state *mwi_retrieve_then_create_state(const char *mailbox)
+{
+	int urgent_msgs;
+	int new_msgs;
+	int old_msgs;
+
+	ast_app_inboxcount2(mailbox, &urgent_msgs, &new_msgs, &old_msgs);
+	return mwi_create_state(mailbox, NULL, urgent_msgs, new_msgs, old_msgs);
+}
+
+struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
+{
+	return mwi_create_state(mailbox, context, 0, 0, 0);
+}
+
 /*!
  * \internal
  * \brief Create a MWI state snapshot message.
@@ -145,6 +169,7 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
 static struct stasis_message *mwi_state_create_message(
 	const char *mailbox,
 	const char *context,
+	int urgent_msgs,
 	int new_msgs,
 	int old_msgs,
 	const char *channel_id,
@@ -157,14 +182,11 @@ static struct stasis_message *mwi_state_create_message(
 		return NULL;
 	}
 
-	mwi_state = ast_mwi_create(mailbox, context);
+	mwi_state = mwi_create_state(mailbox, context, urgent_msgs, new_msgs, old_msgs);
 	if (!mwi_state) {
 		return NULL;
 	}
 
-	mwi_state->new_msgs = new_msgs;
-	mwi_state->old_msgs = old_msgs;
-
 	if (!ast_strlen_zero(channel_id)) {
 		mwi_state->snapshot = ast_channel_snapshot_get_latest(channel_id);
 	}
@@ -186,6 +208,183 @@ static struct stasis_message *mwi_state_create_message(
 	return message;
 }
 
+/*!
+ * \internal
+ *
+ * This object currently acts as a typedef, but can also be thought of as a "child" object
+ * of the stasis_state_subscriber type. As such the "base" pointer should always be the
+ * first object attribute. Doing so allows this object to be easily type cast and used by
+ * the stasis_state code.
+ */
+struct ast_mwi_subscriber {
+	/*! The "base" state subscriber. (Must be first object attribute) */
+	struct stasis_state_subscriber *base;
+};
+
+struct ast_mwi_subscriber *ast_mwi_add_subscriber(const char *mailbox)
+{
+	return (struct ast_mwi_subscriber *)stasis_state_add_subscriber(
+		mwi_state_manager, mailbox);
+}
+
+struct ast_mwi_subscriber *ast_mwi_subscribe_pool(const char *mailbox,
+	stasis_subscription_cb callback, void *data)
+{
+	struct stasis_subscription *stasis_sub;
+	struct ast_mwi_subscriber *sub = (struct ast_mwi_subscriber *)stasis_state_subscribe_pool(
+		mwi_state_manager, mailbox, callback, data);
+
+	if (!sub) {
+		return NULL;
+	}
+
+	stasis_sub = ast_mwi_subscriber_subscription(sub);
+
+	stasis_subscription_accept_message_type(stasis_sub, ast_mwi_state_type());
+	stasis_subscription_set_filter(stasis_sub, STASIS_SUBSCRIPTION_FILTER_SELECTIVE);
+
+	return sub;
+}
+
+void *ast_mwi_unsubscribe(struct ast_mwi_subscriber *sub)
+{
+	return stasis_state_unsubscribe((struct stasis_state_subscriber *)sub);
+}
+
+void *ast_mwi_unsubscribe_and_join(struct ast_mwi_subscriber *sub)
+{
+	return stasis_state_unsubscribe_and_join((struct stasis_state_subscriber *)sub);
+}
+
+struct stasis_topic *ast_mwi_subscriber_topic(struct ast_mwi_subscriber *sub)
+{
+	return stasis_state_subscriber_topic((struct stasis_state_subscriber *)sub);
+}
+
+struct ast_mwi_state *ast_mwi_subscriber_data(struct ast_mwi_subscriber *sub)
+{
+	struct stasis_state_subscriber *s = (struct stasis_state_subscriber *)sub;
+	struct ast_mwi_state *mwi_state = stasis_state_subscriber_data(s);
+
+	return mwi_state ?: mwi_retrieve_then_create_state(stasis_state_subscriber_id(s));
+}
+
+struct stasis_subscription *ast_mwi_subscriber_subscription(struct ast_mwi_subscriber *sub)
+{
+	return stasis_state_subscriber_subscription((struct stasis_state_subscriber *)sub);
+}
+
+/*!
+ * \internal
+ *
+ * This object currently acts as a typedef, but can also be thought of as a "child" object
+ * of the stasis_state_publisher type. As such the "base" pointer should always be the
+ * first object attribute. Doing so allows this object to be easily type cast and used by
+ * the stasis_state code.
+ */
+struct ast_mwi_publisher {
+	/*! The "base" state publisher. (Must be first object attribute) */
+	struct stasis_state_publisher *base;
+};
+
+struct ast_mwi_publisher *ast_mwi_add_publisher(const char *mailbox)
+{
+	return (struct ast_mwi_publisher *)stasis_state_add_publisher(
+		mwi_state_manager, mailbox);
+}
+
+int ast_mwi_add_observer(struct ast_mwi_observer *observer)
+{
+	return stasis_state_add_observer(mwi_state_manager,
+		(struct stasis_state_observer *)observer);
+}
+
+void ast_mwi_remove_observer(struct ast_mwi_observer *observer)
+{
+	stasis_state_remove_observer(mwi_state_manager,
+		(struct stasis_state_observer *)observer);
+}
+
+struct mwi_handler_data {
+	on_mwi_state handler;
+	void *data;
+};
+
+static int handle_mwi_state(const char *id, struct stasis_message *msg, void *user_data)
+{
+	struct mwi_handler_data *d = user_data;
+	struct ast_mwi_state *mwi_state = stasis_message_data(msg);
+	int res;
+
+	if (mwi_state) {
+		return d->handler(mwi_state, d->data);
+	}
+
+	mwi_state = mwi_create_state(id, NULL, 0, 0, 0);
+	if (!mwi_state) {
+		return 0;
+	}
+
+	res = d->handler(mwi_state, d->data);
+	ao2_ref(mwi_state, -1);
+	return res;
+}
+
+void ast_mwi_state_callback_all(on_mwi_state handler, void *data)
+{
+	struct mwi_handler_data d = {
+		.handler = handler,
+		.data = data
+	};
+
+	stasis_state_callback_all(mwi_state_manager, handle_mwi_state, &d);
+}
+
+void ast_mwi_state_callback_subscribed(on_mwi_state handler, void *data)
+{
+	struct mwi_handler_data d = {
+		.handler = handler,
+		.data = data
+	};
+
+	stasis_state_callback_subscribed(mwi_state_manager, handle_mwi_state, &d);
+}
+
+int ast_mwi_publish(struct ast_mwi_publisher *pub, int urgent_msgs,
+	int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid)
+{
+	struct stasis_state_publisher *p = (struct stasis_state_publisher *)pub;
+	struct stasis_message *msg = mwi_state_create_message(stasis_state_publisher_id(p),
+		NULL, urgent_msgs, new_msgs, old_msgs, channel_id, eid);
+
+	if (!msg) {
+		return -1;
+	}
+
+	stasis_state_publish(p, msg);
+	ao2_ref(msg, -1);
+
+	return 0;
+}
+
+int ast_mwi_publish_by_mailbox(const char *mailbox, const char *context, int urgent_msgs,
+	int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid)
+{
+	struct ast_mwi_state *mwi_state;
+	struct stasis_message *msg = mwi_state_create_message(
+		mailbox, context, urgent_msgs, new_msgs, old_msgs, channel_id, eid);
+
+	if (!msg) {
+		return -1;
+	}
+
+	mwi_state = stasis_message_data(msg);
+	stasis_state_publish_by_id(mwi_state_manager, mwi_state->uniqueid, NULL, msg);
+	ao2_ref(msg, -1);
+
+	return 0;
+}
+
 int ast_publish_mwi_state_full(
 	const char *mailbox,
 	const char *context,
@@ -194,24 +393,7 @@ int ast_publish_mwi_state_full(
 	const char *channel_id,
 	struct ast_eid *eid)
 {
-	struct ast_mwi_state *mwi_state;
-	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
-	struct stasis_topic *mailbox_specific_topic;
-
-	message = mwi_state_create_message(mailbox, context, new_msgs, old_msgs, channel_id, eid);
-	if (!message) {
-		return -1;
-	}
-
-	mwi_state = stasis_message_data(message);
-	mailbox_specific_topic = ast_mwi_topic(mwi_state->uniqueid);
-	if (!mailbox_specific_topic) {
-		return -1;
-	}
-
-	stasis_publish(mailbox_specific_topic, message);
-
-	return 0;
+	return ast_mwi_publish_by_mailbox(mailbox, context, 0, new_msgs, old_msgs, channel_id, eid);
 }
 
 int ast_delete_mwi_state_full(const char *mailbox, const char *context, struct ast_eid *eid)
@@ -220,9 +402,8 @@ int ast_delete_mwi_state_full(const char *mailbox, const char *context, struct a
 	struct stasis_message *cached_msg;
 	struct stasis_message *clear_msg;
 	struct ast_mwi_state *mwi_state;
-	struct stasis_topic *mailbox_specific_topic;
 
-	msg = mwi_state_create_message(mailbox, context, 0, 0, NULL, eid);
+	msg = mwi_state_create_message(mailbox, context, 0, 0, 0, NULL, eid);
 	if (!msg) {
 		return -1;
 	}
@@ -244,22 +425,16 @@ int ast_delete_mwi_state_full(const char *mailbox, const char *context, struct a
 	cached_msg = stasis_cache_get_by_eid(ast_mwi_state_cache(),
 		ast_mwi_state_type(), mwi_state->uniqueid, &ast_eid_default);
 	if (!cached_msg) {
-		/* Nothing to clear */
+		/* Nothing to clear from the cache, but still need to remove state */
+		stasis_state_remove_publish_by_id(mwi_state_manager, mwi_state->uniqueid, eid, NULL);
 		return -1;
 	}
 	ao2_cleanup(cached_msg);
 
-	mailbox_specific_topic = ast_mwi_topic(mwi_state->uniqueid);
-	if (!mailbox_specific_topic) {
-		return -1;
-	}
-
 	clear_msg = stasis_cache_clear_create(msg);
-	if (clear_msg) {
-		stasis_publish(mailbox_specific_topic, clear_msg);
-	}
-
+	stasis_state_remove_publish_by_id(mwi_state_manager, mwi_state->uniqueid, eid, clear_msg);
 	ao2_cleanup(clear_msg);
+
 	return 0;
 }
 
@@ -315,13 +490,11 @@ struct stasis_message *ast_mwi_blob_create(struct ast_mwi_state *mwi_state,
 
 static void mwi_cleanup(void)
 {
-	ao2_cleanup(mwi_topic_pool);
-	mwi_topic_pool = NULL;
-	ao2_cleanup(mwi_topic_all);
-	mwi_topic_all = NULL;
 	ao2_cleanup(mwi_state_cache);
 	mwi_state_cache = NULL;
 	mwi_topic_cached = stasis_caching_unsubscribe_and_join(mwi_topic_cached);
+	ao2_cleanup(mwi_state_manager);
+	mwi_state_manager = NULL;
 	STASIS_MESSAGE_TYPE_CLEANUP(ast_mwi_state_type);
 	STASIS_MESSAGE_TYPE_CLEANUP(ast_mwi_vm_app_type);
 }
@@ -338,8 +511,8 @@ int mwi_init(void)
 		return -1;
 	}
 
-	mwi_topic_all = stasis_topic_create("mwi:all");
-	if (!mwi_topic_all) {
+	mwi_state_manager = stasis_state_manager_create("mwi:all");
+	if (!mwi_state_manager) {
 		return -1;
 	}
 
@@ -348,15 +521,10 @@ int mwi_init(void)
 		return -1;
 	}
 
-	mwi_topic_cached = stasis_caching_topic_create(mwi_topic_all, mwi_state_cache);
+	mwi_topic_cached = stasis_caching_topic_create(ast_mwi_topic_all(), mwi_state_cache);
 	if (!mwi_topic_cached) {
 		return -1;
 	}
 
-	mwi_topic_pool = stasis_topic_pool_create(mwi_topic_all);
-	if (!mwi_topic_pool) {
-		return -1;
-	}
-
 	return 0;
 }

tests/test_mwi.c

@@ -0,0 +1,407 @@
+/*
+ * 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/mwi.h"
+#include "asterisk/stasis.h"
+#include "asterisk/test.h"
+
+#define test_category "/mwi/"
+
+#define MAILBOX_PREFIX "test~" /* Hopefully sufficiently unlikely */
+#define MAILBOX_COUNT 500
+#define MAILBOX_SIZE 32
+
+AST_VECTOR(subscriptions, struct ast_mwi_subscriber *);
+AST_VECTOR(publishers, struct ast_mwi_publisher *);
+
+/*!
+ * For testing purposes each subscribed mailbox is a number. This value is
+ * the summation of all mailboxes.
+ */
+static size_t sum_total;
+
+/*! Test variable that tracks the running total of mailboxes */
+static size_t running_total;
+
+/*! This value is set to check if MWI data is zero before publishing */
+static int expect_zero;
+
+static int num_to_mailbox(char *mailbox, size_t size, size_t num)
+{
+	if (snprintf(mailbox, 10, MAILBOX_PREFIX "%zu", num) == -1) {
+		ast_log(LOG_ERROR, "Unable to convert mailbox to string\n");
+		return -1;
+	}
+
+	return 0;
+}
+
+static int mailbox_to_num(const char *mailbox, size_t *num)
+{
+	const char *p = strchr(mailbox, '~');
+
+	if (!p) {
+		ast_log(LOG_ERROR, "Prefix separator '~' not found in '%s'\n", mailbox);
+		return -1;
+	}
+
+	if (ast_str_to_umax(++p, num)) {
+		ast_log(LOG_ERROR, "Unable to convert mailbox '%s' to numeric\n", mailbox);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int validate_data(struct ast_mwi_state *mwi_state)
+{
+	size_t num;
+	size_t val;
+
+	if (mailbox_to_num(mwi_state->uniqueid, &num)) {
+		return -1;
+	}
+
+	running_total += num;
+
+	val = expect_zero ? 0 : num;
+
+	if (mwi_state->urgent_msgs != val || mwi_state->new_msgs != val ||
+			mwi_state->old_msgs != val) {
+		ast_log(LOG_ERROR, "Unexpected MWI state data for '%s', %d != %zu\n",
+				mwi_state->uniqueid, mwi_state->urgent_msgs, val);
+		return -1;
+	}
+
+	return num;
+}
+
+static void handle_validate(const char *mailbox, struct ast_mwi_subscriber *sub)
+{
+	struct ast_mwi_state *mwi_state = ast_mwi_subscriber_data(sub);
+
+	if (ast_begins_with(mwi_state->uniqueid, MAILBOX_PREFIX)) {
+		validate_data(mwi_state);
+	}
+
+	ao2_cleanup(mwi_state);
+}
+
+struct ast_mwi_observer mwi_observer = {
+	.on_subscribe = handle_validate,
+	.on_unsubscribe = handle_validate
+};
+
+static void mwi_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 subscriptions *subs)
+{
+	running_total = expect_zero = 0;
+
+	AST_VECTOR_CALLBACK_VOID(subs, ast_mwi_unsubscribe_and_join);
+	AST_VECTOR_FREE(subs);
+
+	ast_mwi_remove_observer(&mwi_observer);
+
+	if (running_total != sum_total) {
+		ast_log(LOG_ERROR, "Failed to destroy all MWI subscriptions: running=%zu, sum=%zu\n",
+				running_total, sum_total);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int subscriptions_create(struct subscriptions *subs)
+{
+	size_t i;
+
+	if (ast_mwi_add_observer(&mwi_observer) ||
+		AST_VECTOR_INIT(subs, MAILBOX_COUNT)) {
+		return -1;
+	}
+
+	sum_total = running_total = 0;
+	expect_zero = 1;
+
+	for (i = 0; i < MAILBOX_COUNT; ++i) {
+		struct ast_mwi_subscriber *sub;
+		char mailbox[MAILBOX_SIZE];
+
+		if (num_to_mailbox(mailbox, MAILBOX_SIZE, i)) {
+			break;
+		}
+
+		sub = ast_mwi_subscribe_pool(mailbox, mwi_type_cb, NULL);
+		if (!sub) {
+			ast_log(LOG_ERROR, "Failed to create a MWI subscriber for mailbox '%s'\n", mailbox);
+			break;
+		}
+
+		if (AST_VECTOR_APPEND(subs, sub)) {
+			ast_log(LOG_ERROR, "Failed to add to MWI sub to vector for mailbox '%s'\n", mailbox);
+			ao2_ref(sub, -1);
+			break;
+		}
+
+		sum_total += i;
+	}
+
+	if (i != MAILBOX_COUNT || running_total != sum_total) {
+		ast_log(LOG_ERROR, "Failed to create all MWI subscriptions: running=%zu, sum=%zu\n",
+				running_total, sum_total);
+		subscriptions_destroy(subs);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int publishers_destroy(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 < MAILBOX_COUNT; ++i) {
+		char mailbox[MAILBOX_SIZE];
+
+		/* Remove implicit publishers */
+		if (num_to_mailbox(mailbox, MAILBOX_SIZE, i)) {
+			return -1;
+		}
+
+		ast_delete_mwi_state(mailbox, NULL);
+	}
+
+	return 0;
+}
+
+static int publishers_create(struct publishers *pubs)
+{
+	size_t i;
+
+	if (AST_VECTOR_INIT(pubs, MAILBOX_COUNT)) {
+		return -1;
+	}
+
+	for (i = 0; i < MAILBOX_COUNT; ++i) {
+		struct ast_mwi_publisher *pub;
+		char mailbox[MAILBOX_SIZE];
+
+		if (num_to_mailbox(mailbox, MAILBOX_SIZE, i)) {
+			break;
+		}
+
+		/* Create the MWI publisher */
+		pub = ast_mwi_add_publisher(mailbox);
+		if (!pub) {
+			ast_log(LOG_ERROR, "Failed to create an MWI publisher for mailbox '%s'\n", mailbox);
+			break;
+		}
+
+		if (AST_VECTOR_APPEND(pubs, pub)) {
+			ast_log(LOG_ERROR, "Failed to add to an MWI publisher to vector for mailbox '%s'\n", mailbox);
+			ao2_ref(pub, -1);
+			break;
+		}
+	}
+
+	if (i != MAILBOX_COUNT) {
+		ast_log(LOG_ERROR, "Failed to create all MWI publishers: count=%zu\n", i);
+		publishers_destroy(pubs);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int implicit_publish_cb(struct ast_mwi_state *mwi_state, void *data)
+{
+	size_t num;
+
+	if (!ast_begins_with(mwi_state->uniqueid, MAILBOX_PREFIX)) {
+		/* Ignore any mailboxes not prefixed */
+		return 0;
+	}
+
+	num = validate_data(mwi_state);
+	if (num < 0) {
+		return CMP_STOP;
+	}
+
+	ast_mwi_publish_by_mailbox(mwi_state->uniqueid, NULL, num, num, num, NULL, NULL);
+
+	return 0;
+}
+
+static int explicit_publish_cb(struct ast_mwi_state *mwi_state, void *data)
+{
+	struct publishers *pubs = data;
+	struct ast_mwi_publisher *pub;
+	size_t num;
+
+	if (!ast_begins_with(mwi_state->uniqueid, MAILBOX_PREFIX)) {
+		/* Ignore any mailboxes not prefixed */
+		return 0;
+	}
+
+	num = validate_data(mwi_state);
+	if (num < 0) {
+		return CMP_STOP;
+	}
+
+	if (mailbox_to_num(mwi_state->uniqueid, &num)) {
+		return CMP_STOP;
+	}
+
+	/* Mailbox number will always be the index */
+	pub = AST_VECTOR_GET(pubs, num);
+
+	if (!pub) {
+		ast_log(LOG_ERROR, "Unable to locate MWI publisher for mailbox '%s'\n", mwi_state->uniqueid);
+		return CMP_STOP;
+	}
+
+	ast_mwi_publish(pub, num, num, num, NULL, NULL);
+
+	return 0;
+}
+
+static int publish(on_mwi_state cb, void *user_data)
+{
+	/* First time there is no state data */
+	expect_zero = 1;
+
+	running_total = 0;
+	ast_mwi_state_callback_all(cb, user_data);
+
+	if (running_total != sum_total) {
+		ast_log(LOG_ERROR, "Failed MWI state callback (1): running=%zu, sum=%zu\n",
+				running_total, sum_total);
+		return -1;
+	}
+
+	/* Second time check valid state data exists */
+	running_total = expect_zero = 0;
+	ast_mwi_state_callback_all(cb, user_data);
+
+	if (running_total != sum_total) {
+		ast_log(LOG_ERROR, "Failed MWI state callback (2): running=%zu, sum=%zu\n",
+				running_total, sum_total);
+		return -1;
+	}
+
+	return 0;
+}
+
+AST_TEST_DEFINE(implicit_publish)
+{
+	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 MWI state";
+		info->description = info->summary;
+		return AST_TEST_NOT_RUN;
+	case TEST_EXECUTE:
+		break;
+	}
+
+	ast_test_validate(test, !subscriptions_create(&subs));
+
+	ast_test_validate_cleanup(test, !publish(implicit_publish_cb, NULL),
+		rc, cleanup);
+
+cleanup:
+	if (subscriptions_destroy(&subs) || publishers_destroy(NULL)) {
+		return AST_TEST_FAIL;
+	}
+
+	return rc;
+}
+
+AST_TEST_DEFINE(explicit_publish)
+{
+	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 MWI state";
+		info->description = info->summary;
+		return AST_TEST_NOT_RUN;
+	case TEST_EXECUTE:
+		break;
+	}
+
+	ast_test_validate(test, !subscriptions_create(&subs));
+	ast_test_validate_cleanup(test, !publishers_create(&pubs), rc, cleanup);
+
+	ast_test_validate_cleanup(test, !publish(explicit_publish_cb, &pubs),
+		rc, cleanup);
+
+cleanup:
+	if (subscriptions_destroy(&subs) || publishers_destroy(&pubs)) {
+		return AST_TEST_FAIL;
+	}
+
+	return rc;
+}
+
+static int unload_module(void)
+{
+	AST_TEST_UNREGISTER(implicit_publish);
+	AST_TEST_UNREGISTER(explicit_publish);
+
+	return 0;
+}
+
+static int load_module(void)
+{
+	AST_TEST_REGISTER(implicit_publish);
+	AST_TEST_REGISTER(explicit_publish);
+
+	return AST_MODULE_LOAD_SUCCESS;
+}
+
+AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "MWI testing");