Initial support for endpoints.

An endpoint is an external device/system that may offer/accept
channels to/from Asterisk. While this is a very useful concept for end
users, it is surprisingly not a core concept within Asterisk itself.

This patch defines ast_endpoint as a separate object, which channel
drivers may use to expose their concept of an endpoint. As the channel
driver creates channels, it can use ast_endpoint_add_channel() to
associate channels to the endpoint. This updated the endpoint
appropriately, and forwards all of the channel's events to the
endpoint's topic.

In order to avoid excessive locking on the endpoint object itself, the
mutable state is not accessible via getters. Instead, you can create a
snapshot using ast_endpoint_snapshot_create() to get a consistent
snapshot of the internal state.

This patch also includes a set of topics and messages associated with
endpoints, and implementations of the endpoint-related RESTful
API. chan_sip was updated to create endpoints with SIP peers, but the
state of the endpoints is not updated with the state of the peer.

Along for the ride in this patch is a Stasis test API. This is a
stasis_message_sink object, which can be subscribed to a Stasis
topic. It has functions for blocking while waiting for conditions in
the message sink to be fulfilled.

(closes issue ASTERISK-21421)
Review: https://reviewboard.asterisk.org/r/2492/



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

1 files changed, 291 insertions, 0 deletions

diff --git a/res/res_stasis_test.c b/res/res_stasis_test.c
new file mode 100644
index 000000000..7b5aece4f
--- /dev/null
+++ b/res/res_stasis_test.c
@@ -0,0 +1,291 @@
+/*
+ * 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 Test infrastructure for dealing with Stasis.
+ *
+ * \author David M. Lee, II <dlee@digium.com>
+ */
+
+/*** MODULEINFO
+	<depend>TEST_FRAMEWORK</depend>
+	<support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+ASTERISK_FILE_VERSION(__FILE__, "$Revision$");
+
+#include "asterisk/astobj2.h"
+#include "asterisk/module.h"
+#include "asterisk/stasis_test.h"
+
+static struct stasis_message_type *test_message_type;
+
+static void stasis_message_sink_dtor(void *obj)
+{
+	struct stasis_message_sink *sink = obj;
+
+	{
+		SCOPED_MUTEX(lock, &sink->lock);
+		while (!sink->is_done) {
+			/* Normally waiting forever is bad, but if we're not
+			 * done, we're not done. */
+			ast_cond_wait(&sink->cond, &sink->lock);
+		}
+	}
+
+	ast_mutex_destroy(&sink->lock);
+	ast_cond_destroy(&sink->cond);
+
+	while (sink->num_messages > 0) {
+		ao2_cleanup(sink->messages[--sink->num_messages]);
+	}
+	ast_free(sink->messages);
+	sink->messages = NULL;
+	sink->max_messages = 0;
+}
+
+static struct timespec make_deadline(int timeout_millis)
+{
+	struct timeval start = ast_tvnow();
+	struct timeval delta = {
+		.tv_sec = timeout_millis / 1000,
+		.tv_usec = (timeout_millis % 1000) * 1000,
+	};
+	struct timeval deadline_tv = ast_tvadd(start, delta);
+	struct timespec deadline = {
+		.tv_sec = deadline_tv.tv_sec,
+		.tv_nsec = 1000 * deadline_tv.tv_usec,
+	};
+
+	return deadline;
+}
+
+struct stasis_message_sink *stasis_message_sink_create(void)
+{
+	RAII_VAR(struct stasis_message_sink *, sink, NULL, ao2_cleanup);
+
+	sink = ao2_alloc(sizeof(*sink), stasis_message_sink_dtor);
+	if (!sink) {
+		return NULL;
+	}
+	ast_mutex_init(&sink->lock);
+	ast_cond_init(&sink->cond, NULL);
+	sink->max_messages = 4;
+	sink->messages =
+		ast_malloc(sizeof(*sink->messages) * sink->max_messages);
+	if (!sink->messages) {
+		return NULL;
+	}
+	ao2_ref(sink, +1);
+	return sink;
+}
+
+/*!
+ * \brief Implementation of the stasis_message_sink_cb() callback.
+ *
+ * Why the roundabout way of exposing this via stasis_message_sink_cb()? Well,
+ * it has to do with how we load modules.
+ *
+ * Modules have their own metadata compiled into them in the AST_MODULE_INFO()
+ * block. This includes dependency information in the \c nonoptreq field.
+ *
+ * Asterisk loads the module, inspects the field, then loads any needed
+ * dependencies. This works because Asterisk passes \c RTLD_LAZY to the initial
+ * dlopen(), which defers binding function references until they are called.
+ *
+ * But when you take the address of a function, that function needs to be
+ * available at load time. So if some module used the address of
+ * message_sink_cb() directly, and \c res_stasis_test.so wasn't loaded yet, then
+ * that module would fail to load.
+ *
+ * The stasis_message_sink_cb() function gives us a layer of indirection so that
+ * the initial lazy binding will still work as expected.
+ */
+static void message_sink_cb(void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic, struct stasis_message *message)
+{
+	struct stasis_message_sink *sink = data;
+
+	SCOPED_MUTEX(lock, &sink->lock);
+
+	if (stasis_subscription_final_message(sub, message)) {
+		sink->is_done = 1;
+		ast_cond_signal(&sink->cond);
+		return;
+	}
+
+	if (stasis_subscription_change_type() == stasis_message_type(message)) {
+		/* Ignore subscription changes */
+		return;
+	}
+
+	if (sink->num_messages == sink->max_messages) {
+		size_t new_max_messages = sink->max_messages * 2;
+		struct stasis_message **new_messages = ast_realloc(
+			sink->messages,
+			sizeof(*new_messages) * new_max_messages);
+		if (!new_messages) {
+			return;
+		}
+		sink->max_messages = new_max_messages;
+		sink->messages = new_messages;
+	}
+
+	ao2_ref(message, +1);
+	sink->messages[sink->num_messages++] = message;
+	ast_cond_signal(&sink->cond);
+}
+
+stasis_subscription_cb stasis_message_sink_cb(void)
+{
+	return message_sink_cb;
+}
+
+
+int stasis_message_sink_wait_for_count(struct stasis_message_sink *sink,
+	int num_messages, int timeout_millis)
+{
+	struct timespec deadline = make_deadline(timeout_millis);
+
+	SCOPED_MUTEX(lock, &sink->lock);
+	while (sink->num_messages < num_messages) {
+		int r = ast_cond_timedwait(&sink->cond, &sink->lock, &deadline);
+
+		if (r == ETIMEDOUT) {
+			break;
+		}
+		if (r != 0) {
+			ast_log(LOG_ERROR, "Unexpected condition error: %s\n",
+				strerror(r));
+			break;
+		}
+	}
+	return sink->num_messages;
+}
+
+int stasis_message_sink_should_stay(struct stasis_message_sink *sink,
+	int num_messages, int timeout_millis)
+{
+	struct timespec deadline = make_deadline(timeout_millis);
+
+	SCOPED_MUTEX(lock, &sink->lock);
+	while (sink->num_messages == num_messages) {
+		int r = ast_cond_timedwait(&sink->cond, &sink->lock, &deadline);
+
+		if (r == ETIMEDOUT) {
+			break;
+		}
+		if (r != 0) {
+			ast_log(LOG_ERROR, "Unexpected condition error: %s\n",
+				strerror(r));
+			break;
+		}
+	}
+	return sink->num_messages;
+}
+
+int stasis_message_sink_wait_for(struct stasis_message_sink *sink, int start,
+	stasis_wait_cb cmp_cb, const void *data, int timeout_millis)
+{
+	struct timespec deadline = make_deadline(timeout_millis);
+
+	SCOPED_MUTEX(lock, &sink->lock);
+
+	/* wait for the start */
+	while (sink->num_messages < start + 1) {
+		int r = ast_cond_timedwait(&sink->cond, &sink->lock, &deadline);
+
+		if (r == ETIMEDOUT) {
+			/* Timed out waiting for the start */
+			return -1;
+		}
+		if (r != 0) {
+			ast_log(LOG_ERROR, "Unexpected condition error: %s\n",
+				strerror(r));
+			return -2;
+		}
+	}
+
+
+	while (!cmp_cb(sink->messages[start], data)) {
+		++start;
+
+		while (sink->num_messages < start + 1) {
+			int r = ast_cond_timedwait(&sink->cond,
+				&sink->lock, &deadline);
+
+			if (r == ETIMEDOUT) {
+				return -1;
+			}
+			if (r != 0) {
+				ast_log(LOG_ERROR,
+					"Unexpected condition error: %s\n",
+					strerror(r));
+				return -2;
+			}
+		}
+	}
+
+	return start;
+}
+
+struct stasis_message *stasis_test_message_create(void)
+{
+	RAII_VAR(void *, data, NULL, ao2_cleanup);
+
+	/* We just need the unique pointer; don't care what's in it */
+	data = ao2_alloc(1, NULL);
+	if (!data) {
+		return NULL;
+	}
+
+	return stasis_message_create(stasis_test_message_type(), data);
+}
+
+struct stasis_message_type *stasis_test_message_type(void)
+{
+	return test_message_type;
+}
+
+static int unload_module(void)
+{
+	ao2_cleanup(test_message_type);
+	test_message_type = NULL;
+	return 0;
+}
+
+static int load_module(void)
+{
+	test_message_type = stasis_message_type_create(
+		"stasis_test_message");
+	if (!test_message_type) {
+		return AST_MODULE_LOAD_FAILURE;
+	}
+
+	return AST_MODULE_LOAD_SUCCESS;
+}
+
+AST_MODULE_INFO(ASTERISK_GPL_KEY,
+	AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER,
+	"Stasis test utilities",
+	.load = load_module,
+	.unload = unload_module,
+	.load_pri = AST_MODPRI_APP_DEPEND,
+	);