This patch adds a new message bus API to Asterisk.

For the initial use of this bus, I took some work kmoore did creating
channel snapshots. So rather than create AMI events directly in the
channel code, this patch generates Stasis events, which manager.c uses
to then publish the AMI event.

This message bus provides a generic publish/subscribe mechanism within
Asterisk. This message bus is:

 - Loosely coupled; new message types can be added in seperate modules.
 - Easy to use; publishing and subscribing are straightforward
   operations.

In addition to basic publish/subscribe, the patch also provides
mechanisms for message forwarding, and for message caching.

(issue ASTERISK-20887)
(closes issue ASTERISK-20959)
Review: https://reviewboard.asterisk.org/r/2339/


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@382685 65c4cc65-6c06-0410-ace0-fbb531ad65f3

1 files changed, 443 insertions, 0 deletions

diff --git a/main/stasis_cache.c b/main/stasis_cache.c
new file mode 100644
index 000000000..2f4cf52fd
--- /dev/null
+++ b/main/stasis_cache.c
@@ -0,0 +1,443 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2013, Digium, Inc.
+ *
+ * David M. Lee, II <dlee@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.
+ */
+
+/*! \file
+ *
+ * \brief Stasis Message API.
+ *
+ * \author David M. Lee, II <dlee@digium.com>
+ */
+
+/*** MODULEINFO
+	<support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
+
+#include "asterisk/astobj2.h"
+#include "asterisk/hashtab.h"
+#include "asterisk/stasis.h"
+#include "asterisk/utils.h"
+
+#ifdef LOW_MEMORY
+#define NUM_CACHE_BUCKETS 17
+#else
+#define NUM_CACHE_BUCKETS 563
+#endif
+
+struct stasis_caching_topic {
+	struct ao2_container *cache;
+	struct stasis_topic *topic;
+	struct stasis_subscription *sub;
+	snapshot_get_id id_fn;
+};
+
+static void stasis_caching_topic_dtor(void *obj) {
+	struct stasis_caching_topic *caching_topic = obj;
+	ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
+	caching_topic->sub = NULL;
+	ao2_cleanup(caching_topic->cache);
+	caching_topic->cache = NULL;
+	ao2_cleanup(caching_topic->topic);
+	caching_topic->topic = NULL;
+}
+
+struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
+{
+	return caching_topic->topic;
+}
+
+void stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
+{
+	if (caching_topic) {
+		if (stasis_subscription_is_subscribed(caching_topic->sub)) {
+			stasis_unsubscribe(caching_topic->sub);
+		} else {
+			ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
+		}
+	}
+}
+
+struct cache_entry {
+	struct stasis_message_type *type;
+	char *id;
+	struct stasis_message *snapshot;
+};
+
+static void cache_entry_dtor(void *obj)
+{
+	struct cache_entry *entry = obj;
+	ao2_cleanup(entry->type);
+	entry->type = NULL;
+	ast_free(entry->id);
+	entry->id = NULL;
+	ao2_cleanup(entry->snapshot);
+	entry->snapshot = NULL;
+}
+
+static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
+{
+	RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
+
+	ast_assert(type != NULL);
+	ast_assert(id != NULL);
+
+	entry = ao2_alloc(sizeof(*entry), cache_entry_dtor);
+	if (!entry) {
+		return NULL;
+	}
+
+	entry->id = ast_strdup(id);
+	if (!entry->id) {
+		return NULL;
+	}
+
+	ao2_ref(type, +1);
+	entry->type = type;
+	if (snapshot != NULL) {
+		ao2_ref(snapshot, +1);
+		entry->snapshot = snapshot;
+	}
+
+	ao2_ref(entry, +1);
+	return entry;
+}
+
+static int cache_entry_hash(const void *obj, int flags)
+{
+	const struct cache_entry *entry = obj;
+	int hash = 0;
+
+	ast_assert(!(flags & OBJ_KEY));
+
+	hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
+	hash += ast_hashtab_hash_string(entry->id);
+	return hash;
+}
+
+static int cache_entry_cmp(void *obj, void *arg, int flags)
+{
+	const struct cache_entry *left = obj;
+	const struct cache_entry *right = arg;
+
+	ast_assert(!(flags & OBJ_KEY));
+
+	if (left->type == right->type && strcmp(left->id, right->id) == 0) {
+		return CMP_MATCH | CMP_STOP;
+	}
+
+	return 0;
+}
+
+static struct stasis_message *cache_put(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id, struct stasis_message *new_snapshot)
+{
+	RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
+	RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
+	struct stasis_message *old_snapshot = NULL;
+
+	ast_assert(caching_topic->cache != NULL);
+
+	new_entry = cache_entry_create(type, id, new_snapshot);
+
+	if (new_snapshot == NULL) {
+		/* Remove entry from cache */
+		cached_entry = ao2_find(caching_topic->cache, new_entry, OBJ_POINTER | OBJ_UNLINK);
+		if (cached_entry) {
+			old_snapshot = cached_entry->snapshot;
+			cached_entry->snapshot = NULL;
+		}
+	} else {
+		/* Insert/update cache */
+		SCOPED_AO2LOCK(lock, caching_topic->cache);
+
+		cached_entry = ao2_find(caching_topic->cache, new_entry, OBJ_POINTER | OBJ_NOLOCK);
+		if (cached_entry) {
+			/* Update cache. Because objects are moving, no need to update refcounts. */
+			old_snapshot = cached_entry->snapshot;
+			cached_entry->snapshot = new_entry->snapshot;
+			new_entry->snapshot = NULL;
+		} else {
+			/* Insert into the cache */
+			ao2_link_flags(caching_topic->cache, new_entry, OBJ_NOLOCK);
+		}
+
+	}
+
+	return old_snapshot;
+}
+
+struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id)
+{
+	RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
+	RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
+
+	ast_assert(caching_topic->cache != NULL);
+
+	search_entry = cache_entry_create(type, id, NULL);
+	if (search_entry == NULL) {
+		return NULL;
+	}
+
+	cached_entry = ao2_find(caching_topic->cache, search_entry, OBJ_POINTER);
+	if (cached_entry == NULL) {
+		return NULL;
+	}
+
+	ast_assert(cached_entry->snapshot != NULL);
+	ao2_ref(cached_entry->snapshot, +1);
+	return cached_entry->snapshot;
+}
+
+static struct stasis_message_type *__cache_clear_data;
+
+static struct stasis_message_type *cache_clear_data(void)
+{
+	ast_assert(__cache_clear_data != NULL);
+	return __cache_clear_data;
+}
+
+static struct stasis_message_type *__cache_update;
+
+struct stasis_message_type *stasis_cache_update(void)
+{
+	ast_assert(__cache_update != NULL);
+	return __cache_update;
+}
+
+struct cache_clear_data {
+	struct stasis_message_type *type;
+	char *id;
+};
+
+static void cache_clear_data_dtor(void *obj)
+{
+	struct cache_clear_data *ev = obj;
+	ast_free(ev->id);
+	ev->id = NULL;
+	ao2_cleanup(ev->type);
+	ev->type = NULL;
+}
+
+struct stasis_message *stasis_cache_clear_create(struct stasis_message_type *type, const char *id)
+{
+	RAII_VAR(struct cache_clear_data *, ev, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+	ev = ao2_alloc(sizeof(*ev), cache_clear_data_dtor);
+	if (!ev) {
+		return NULL;
+	}
+
+	ev->id = ast_strdup(id);
+	if (!ev->id) {
+		return NULL;
+	}
+	ao2_ref(type, +1);
+	ev->type = type;
+
+	msg = stasis_message_create(cache_clear_data(), ev);
+
+	if (!msg) {
+		return NULL;
+	}
+
+	ao2_ref(msg, +1);
+	return msg;
+}
+
+static void stasis_cache_update_dtor(void *obj)
+{
+	struct stasis_cache_update *update = obj;
+	ao2_cleanup(update->topic);
+	update->topic = NULL;
+	ao2_cleanup(update->old_snapshot);
+	update->old_snapshot = NULL;
+	ao2_cleanup(update->new_snapshot);
+	update->new_snapshot = NULL;
+	ao2_cleanup(update->type);
+	update->type = NULL;
+}
+
+static struct stasis_message *update_create(struct stasis_topic *topic, struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
+{
+	RAII_VAR(struct stasis_cache_update *, update, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+	ast_assert(topic != NULL);
+	ast_assert(old_snapshot != NULL || new_snapshot != NULL);
+
+	update = ao2_alloc(sizeof(*update), stasis_cache_update_dtor);
+	if (!update) {
+		return NULL;
+	}
+
+	ao2_ref(topic, +1);
+	update->topic = topic;
+	if (old_snapshot) {
+		ao2_ref(old_snapshot, +1);
+		update->old_snapshot = old_snapshot;
+		if (!new_snapshot) {
+			ao2_ref(stasis_message_type(old_snapshot), +1);
+			update->type = stasis_message_type(old_snapshot);
+		}
+	}
+	if (new_snapshot) {
+		ao2_ref(new_snapshot, +1);
+		update->new_snapshot = new_snapshot;
+		ao2_ref(stasis_message_type(new_snapshot), +1);
+		update->type = stasis_message_type(new_snapshot);
+	}
+
+	msg = stasis_message_create(stasis_cache_update(), update);
+	if (!msg) {
+		return NULL;
+	}
+
+	ao2_ref(msg, +1);
+	return msg;
+}
+
+static void caching_topic_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
+{
+	RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
+	struct stasis_caching_topic *caching_topic = data;
+	const char *id = NULL;
+
+	ast_assert(caching_topic->topic != NULL);
+	ast_assert(caching_topic->id_fn != NULL);
+
+	if (stasis_subscription_final_message(sub, message)) {
+		caching_topic_needs_unref = caching_topic;
+	}
+
+	/* Handle cache clear event */
+	if (cache_clear_data() == stasis_message_type(message)) {
+		RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
+		RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
+		struct cache_clear_data *clear = stasis_message_data(message);
+		ast_assert(clear->type != NULL);
+		ast_assert(clear->id != NULL);
+		old_snapshot = cache_put(caching_topic, clear->type, clear->id, NULL);
+		if (old_snapshot) {
+			update = update_create(topic, old_snapshot, NULL);
+			stasis_publish(caching_topic->topic, update);
+		} else {
+			ast_log(LOG_ERROR,
+				"Attempting to remove an item from the cache that isn't there: %s %s\n",
+				stasis_message_type_name(clear->type), clear->id);
+		}
+		return;
+	}
+
+	id = caching_topic->id_fn(message);
+	if (id == NULL) {
+		/* Object isn't cached; forward */
+		stasis_forward_message(caching_topic->topic, topic, message);
+	} else {
+		/* Update the cache */
+		RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
+		RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
+
+		old_snapshot = cache_put(caching_topic, stasis_message_type(message), id, message);
+
+		update = update_create(topic, old_snapshot, message);
+		if (update == NULL) {
+			return;
+		}
+
+		stasis_publish(caching_topic->topic, update);
+	}
+
+	if (stasis_subscription_final_message(sub, message)) {
+		ao2_cleanup(caching_topic);
+	}
+}
+
+struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, snapshot_get_id id_fn)
+{
+	RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
+	struct stasis_subscription *sub;
+	RAII_VAR(char *, new_name, NULL, free);
+	int ret;
+
+	ret = asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
+	if (ret < 0) {
+		return NULL;
+	}
+
+	caching_topic = ao2_alloc(sizeof(*caching_topic), stasis_caching_topic_dtor);
+	if (caching_topic == NULL) {
+		return NULL;
+	}
+
+	caching_topic->cache = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash, cache_entry_cmp);
+	if (!caching_topic->cache) {
+		ast_log(LOG_ERROR, "Stasis cache allocation failed\n");
+		return NULL;
+	}
+
+	caching_topic->topic = stasis_topic_create(new_name);
+	if (caching_topic->topic == NULL) {
+		return NULL;
+	}
+
+	caching_topic->id_fn = id_fn;
+
+	sub = stasis_subscribe(original_topic, caching_topic_exec, caching_topic);
+	if (sub == NULL) {
+		return NULL;
+	}
+	/* This is for the reference contained in the subscription above */
+	ao2_ref(caching_topic, +1);
+	caching_topic->sub = sub;
+
+	ao2_ref(caching_topic, +1);
+	return caching_topic;
+}
+
+static void stasis_cache_exit(void)
+{
+	ao2_cleanup(__cache_clear_data);
+	__cache_clear_data = NULL;
+	ao2_cleanup(__cache_update);
+	__cache_update = NULL;
+}
+
+int stasis_cache_init(void)
+{
+	ast_register_atexit(stasis_cache_exit);
+
+	if (__cache_clear_data || __cache_update) {
+		ast_log(LOG_ERROR, "Stasis cache double initialized\n");
+		return -1;
+	}
+
+	__cache_update = stasis_message_type_create("stasis_cache_update");
+	if (!__cache_update) {
+		return -1;
+	}
+
+	__cache_clear_data = stasis_message_type_create("StasisCacheClear");
+	if (!__cache_clear_data) {
+		return -1;
+	}
+	return 0;
+}
+