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

2 files changed, 141 insertions, 14 deletions

diff --git a/res/stasis_http/resource_endpoints.c b/res/stasis_http/resource_endpoints.c
index b2611bad8..52d05c093 100644
--- a/res/stasis_http/resource_endpoints.c
+++ b/res/stasis_http/resource_endpoints.c
@@ -1,4 +1,4 @@
-/* -*- C -*-
+/*
  * Asterisk -- An open source telephony toolkit.
  *
  * Copyright (C) 2012 - 2013, Digium, Inc.
@@ -18,26 +18,140 @@
 
 /*! \file
  *
- * \brief Implementation for stasis-http stubs.
+ * \brief /api-docs/endpoints.{format} implementation- Endpoint resources
  *
  * \author David M. Lee, II <dlee@digium.com>
  */
 
-/*** MODULEINFO
-	<support_level>core</support_level>
- ***/
-
 #include "asterisk.h"
 
 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 
 #include "resource_endpoints.h"
 
-void stasis_http_get_endpoint(struct ast_variable *headers, struct ast_get_endpoint_args *args, struct stasis_http_response *response)
+#include "asterisk/astobj2.h"
+#include "asterisk/stasis.h"
+#include "asterisk/stasis_endpoints.h"
+
+void stasis_http_get_endpoints(struct ast_variable *headers,
+	struct ast_get_endpoints_args *args,
+	struct stasis_http_response *response)
+{
+	RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
+	RAII_VAR(struct ao2_container *, snapshots, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+	struct ao2_iterator i;
+	void *obj;
+
+	caching_topic = ast_endpoint_topic_all_cached();
+	if (!caching_topic) {
+		stasis_http_response_error(
+			response, 500, "Internal Server Error",
+			"Message bus not initialized");
+		return;
+	}
+	ao2_ref(caching_topic, +1);
+
+	snapshots = stasis_cache_dump(caching_topic, ast_endpoint_snapshot_type());
+	if (!snapshots) {
+		stasis_http_response_alloc_failed(response);
+		return;
+	}
+
+	json = ast_json_array_create();
+	if (!json) {
+		stasis_http_response_alloc_failed(response);
+		return;
+	}
+
+	i = ao2_iterator_init(snapshots, 0);
+	while ((obj = ao2_iterator_next(&i))) {
+		RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup);
+		struct ast_endpoint_snapshot *snapshot = stasis_message_data(msg);
+		int r = ast_json_array_append(
+			json, ast_endpoint_snapshot_to_json(snapshot));
+		if (r != 0) {
+			stasis_http_response_alloc_failed(response);
+			return;
+		}
+	}
+	ao2_iterator_destroy(&i);
+
+	stasis_http_response_ok(response, ast_json_ref(json));
+}
+void stasis_http_get_endpoints_by_tech(struct ast_variable *headers,
+	struct ast_get_endpoints_by_tech_args *args,
+	struct stasis_http_response *response)
 {
-	ast_log(LOG_ERROR, "TODO: stasis_http_get_endpoint\n");
+	RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
+	RAII_VAR(struct ao2_container *, snapshots, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+	struct ao2_iterator i;
+	void *obj;
+
+	/* TODO - if tech isn't a recognized type of endpoint, it should 404 */
+
+	caching_topic = ast_endpoint_topic_all_cached();
+	if (!caching_topic) {
+		stasis_http_response_error(
+			response, 500, "Internal Server Error",
+			"Message bus not initialized");
+		return;
+	}
+	ao2_ref(caching_topic, +1);
+
+	snapshots = stasis_cache_dump(caching_topic, ast_endpoint_snapshot_type());
+	if (!snapshots) {
+		stasis_http_response_alloc_failed(response);
+		return;
+	}
+
+	json = ast_json_array_create();
+	if (!json) {
+		stasis_http_response_alloc_failed(response);
+		return;
+	}
+
+	i = ao2_iterator_init(snapshots, 0);
+	while ((obj = ao2_iterator_next(&i))) {
+		RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup);
+		struct ast_endpoint_snapshot *snapshot = stasis_message_data(msg);
+		int r;
+
+		if (strcmp(args->tech, snapshot->tech) != 0) {
+			continue;
+		}
+
+		r = ast_json_array_append(
+			json, ast_endpoint_snapshot_to_json(snapshot));
+		if (r != 0) {
+			stasis_http_response_alloc_failed(response);
+			return;
+		}
+	}
+	ao2_iterator_destroy(&i);
+
+	stasis_http_response_ok(response, ast_json_ref(json));
 }
-void stasis_http_get_endpoints(struct ast_variable *headers, struct ast_get_endpoints_args *args, struct stasis_http_response *response)
+void stasis_http_get_endpoint(struct ast_variable *headers,
+	struct ast_get_endpoint_args *args,
+	struct stasis_http_response *response)
 {
-	ast_log(LOG_ERROR, "TODO: stasis_http_get_endpoints\n");
+	RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+	RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
+
+	snapshot = ast_endpoint_latest_snapshot(args->tech, args->resource);
+	if (!snapshot) {
+		stasis_http_response_error(response, 404, "Not Found",
+			"Endpoint not found");
+		return;
+	}
+
+	json = ast_endpoint_snapshot_to_json(snapshot);
+	if (!json) {
+		stasis_http_response_alloc_failed(response);
+		return;
+	}
+
+	stasis_http_response_ok(response, ast_json_ref(json));
 }
diff --git a/res/stasis_http/resource_endpoints.h b/res/stasis_http/resource_endpoints.h
index 9f5a96e50..b534fb047 100644
--- a/res/stasis_http/resource_endpoints.h
+++ b/res/stasis_http/resource_endpoints.h
@@ -41,21 +41,34 @@
 
 /*! \brief Argument struct for stasis_http_get_endpoints() */
 struct ast_get_endpoints_args {
-	/*! \brief Filter endpoints by type (sip,iax2,dhadi,...) */
-	const char *with_type;
 };
 /*!
- * \brief List available endoints.
+ * \brief List all endoints.
  *
  * \param headers HTTP headers
  * \param args Swagger parameters
  * \param[out] response HTTP response
  */
 void stasis_http_get_endpoints(struct ast_variable *headers, struct ast_get_endpoints_args *args, struct stasis_http_response *response);
+/*! \brief Argument struct for stasis_http_get_endpoints_by_tech() */
+struct ast_get_endpoints_by_tech_args {
+	/*! \brief Technology of the endpoints (sip,iax2,...) */
+	const char *tech;
+};
+/*!
+ * \brief List available endoints for a given endpoint technology.
+ *
+ * \param headers HTTP headers
+ * \param args Swagger parameters
+ * \param[out] response HTTP response
+ */
+void stasis_http_get_endpoints_by_tech(struct ast_variable *headers, struct ast_get_endpoints_by_tech_args *args, struct stasis_http_response *response);
 /*! \brief Argument struct for stasis_http_get_endpoint() */
 struct ast_get_endpoint_args {
+	/*! \brief Technology of the endpoint */
+	const char *tech;
 	/*! \brief ID of the endpoint */
-	const char *endpoint_id;
+	const char *resource;
 };
 /*!
  * \brief Details for an endpoint.