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, 200 insertions, 0 deletions

diff --git a/main/manager.c b/main/manager.c
index fc0ec2631..10a3a3397 100644
--- a/main/manager.c
+++ b/main/manager.c
@@ -91,6 +91,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/aoc.h"
 #include "asterisk/stringfields.h"
 #include "asterisk/presencestate.h"
+#include "asterisk/stasis.h"
 
 /*** DOCUMENTATION
 	<manager name="Ping" language="en_US">
@@ -963,6 +964,73 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
                         manager.conf will be present upon starting a new session.</para>
 		</description>
 	</manager>
+	<managerEvent language="en_US" name="Newchannel">
+		<managerEventInstance class="EVENT_FLAG_CALL">
+			<synopsis>Raised when a new channel is created.</synopsis>
+			<syntax>
+				<parameter name="Channel">
+				</parameter>
+				<parameter name="ChannelState">
+					<para>A numeric code for the channel's current state, related to ChannelStateDesc</para>
+				</parameter>
+				<parameter name="ChannelStateDesc">
+					<enumlist>
+						<enum name="Down"/>
+						<enum name="Rsrvd"/>
+						<enum name="OffHook"/>
+						<enum name="Dialing"/>
+						<enum name="Ring"/>
+						<enum name="Ringing"/>
+						<enum name="Up"/>
+						<enum name="Busy"/>
+						<enum name="Dialing Offhook"/>
+						<enum name="Pre-ring"/>
+						<enum name="Unknown"/>
+					</enumlist>
+				</parameter>
+				<parameter name="CallerIDNum">
+				</parameter>
+				<parameter name="CallerIDName">
+				</parameter>
+				<parameter name="ConnectedLineNum">
+				</parameter>
+				<parameter name="ConnectedLineName">
+				</parameter>
+				<parameter name="AccountCode">
+				</parameter>
+				<parameter name="Context">
+				</parameter>
+				<parameter name="Exten">
+				</parameter>
+				<parameter name="Priority">
+				</parameter>
+				<parameter name="Uniqueid">
+				</parameter>
+				<parameter name="Cause">
+					<para>A numeric cause code for why the channel was hung up.</para>
+				</parameter>
+				<parameter name="Cause-txt">
+					<para>A description of why the channel was hung up.</para>
+				</parameter>
+			</syntax>
+		</managerEventInstance>
+	</managerEvent>
+	<managerEvent language="en_US" name="Newstate">
+		<managerEventInstance class="EVENT_FLAG_CALL">
+			<synopsis>Raised when a channel's state changes.</synopsis>
+			<syntax>
+				<xi:include xpointer="xpointer(/docs/managerEvent[@name='Newchannel']/managerEventInstance/syntax/parameter)" />
+			</syntax>
+		</managerEventInstance>
+	</managerEvent>
+	<managerEvent language="en_US" name="Hangup">
+		<managerEventInstance class="EVENT_FLAG_CALL">
+			<synopsis>Raised when a channel is hung up.</synopsis>
+			<syntax>
+				<xi:include xpointer="xpointer(/docs/managerEvent[@name='Newchannel']/managerEventInstance/syntax/parameter)" />
+			</syntax>
+		</managerEventInstance>
+	</managerEvent>
  ***/
 
 /*! \addtogroup Group_AMI AMI functions
@@ -1060,6 +1128,8 @@ static const struct {
 	{{ "restart", "gracefully", NULL }},
 };
 
+static struct stasis_subscription *channel_state_sub;
+
 static void acl_change_event_cb(const struct ast_event *event, void *userdata);
 
 static void acl_change_event_subscribe(void)
@@ -7376,6 +7446,127 @@ static void load_channelvars(struct ast_variable *var)
 	AST_RWLIST_UNLOCK(&channelvars);
 }
 
+/*!
+ * \brief Generate the AMI message body from a channel snapshot
+ * \internal
+ *
+ * \param snapshot the channel snapshot for which to generate an AMI message body
+ *
+ * \retval NULL on error
+ * \retval ast_str* on success (must be ast_freed by caller)
+ */
+static struct ast_str *manager_build_channel_state_string(const struct ast_channel_snapshot *snapshot)
+{
+	struct ast_str *out = ast_str_create(1024);
+	int res = 0;
+	if (!out) {
+		return NULL;
+	}
+	res = ast_str_set(&out, 0,
+		"Channel: %s\r\n"
+		"ChannelState: %d\r\n"
+		"ChannelStateDesc: %s\r\n"
+		"CallerIDNum: %s\r\n"
+		"CallerIDName: %s\r\n"
+		"ConnectedLineNum: %s\r\n"
+		"ConnectedLineName: %s\r\n"
+		"AccountCode: %s\r\n"
+		"Context: %s\r\n"
+		"Exten: %s\r\n"
+		"Priority: %d\r\n"
+		"Uniqueid: %s\r\n"
+		"Cause: %d\r\n"
+		"Cause-txt: %s\r\n",
+		snapshot->name,
+		snapshot->state,
+		ast_state2str(snapshot->state),
+		snapshot->caller_number,
+		snapshot->caller_name,
+		snapshot->connected_number,
+		snapshot->connected_name,
+		snapshot->accountcode,
+		snapshot->context,
+		snapshot->exten,
+		snapshot->priority,
+		snapshot->uniqueid,
+		snapshot->hangupcause,
+		ast_cause2str(snapshot->hangupcause));
+
+	if (!res) {
+		return NULL;
+	}
+
+	return out;
+}
+
+static void channel_snapshot_update(struct ast_channel_snapshot *old_snapshot, struct ast_channel_snapshot *new_snapshot)
+{
+	int is_hungup;
+	char *manager_event = NULL;
+
+	if (!new_snapshot) {
+		/* Ignore cache clearing events; we'll see the hangup first */
+		return;
+	}
+
+	is_hungup = ast_test_flag(&new_snapshot->flags, AST_FLAG_ZOMBIE) ? 1 : 0;
+
+	if (!old_snapshot) {
+		manager_event = "Newchannel";
+	}
+
+	if (old_snapshot && old_snapshot->state != new_snapshot->state) {
+		manager_event = "Newstate";
+	}
+
+	if (old_snapshot && is_hungup) {
+		manager_event = "Hangup";
+	}
+
+	if (manager_event) {
+		RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
+
+		channel_event_string = manager_build_channel_state_string(new_snapshot);
+		if (channel_event_string) {
+			manager_event(EVENT_FLAG_CALL, manager_event, "%s", ast_str_buffer(channel_event_string));
+		}
+	}
+}
+
+static void channel_varset(const char *channel_name, const char *uniqueid, const char *name, const char *value)
+{
+	/*** DOCUMENTATION
+		<managerEventInstance>
+			<synopsis>Raised when a variable is set to a particular value.</synopsis>
+		</managerEventInstance>
+	***/
+	manager_event(EVENT_FLAG_DIALPLAN, "VarSet",
+		      "Channel: %s\r\n"
+		      "Variable: %s\r\n"
+		      "Value: %s\r\n"
+		      "Uniqueid: %s\r\n",
+		      channel_name, name, value, uniqueid);
+}
+
+static void channel_event_cb(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
+{
+	if (stasis_message_type(message) == stasis_cache_update()) {
+		struct stasis_cache_update *update = stasis_message_data(message);
+		if (ast_channel_snapshot() == update->type) {
+			struct ast_channel_snapshot *old_snapshot =
+				stasis_message_data(update->old_snapshot);
+			struct ast_channel_snapshot *new_snapshot =
+				stasis_message_data(update->new_snapshot);
+			channel_snapshot_update(old_snapshot, new_snapshot);
+		}
+	} else if (stasis_message_type(message) == ast_channel_varset()) {
+		struct ast_channel_varset *varset = stasis_message_data(message);
+		const char *name = varset->snapshot ? varset->snapshot->name : "none";
+		const char *uniqueid = varset->snapshot ? varset->snapshot->uniqueid : "none";
+		channel_varset(name, uniqueid, varset->variable, varset->value);
+	}
+}
+
 /*! \internal \brief Free a user record.  Should already be removed from the list */
 static void manager_free_user(struct ast_manager_user *user)
 {
@@ -7399,6 +7590,9 @@ static void manager_shutdown(void)
 {
 	struct ast_manager_user *user;
 
+	stasis_unsubscribe(channel_state_sub);
+	channel_state_sub = NULL;
+
 	if (registered) {
 		ast_manager_unregister("Ping");
 		ast_manager_unregister("Events");
@@ -7490,6 +7684,12 @@ static int __init_manager(int reload, int by_external_config)
 
 	manager_enabled = 0;
 
+	if (!channel_state_sub) {
+		channel_state_sub = stasis_subscribe(
+			stasis_caching_get_topic(ast_channel_topic_all_cached()),
+			channel_event_cb, NULL);
+	}
+
 	if (!registered) {
 		/* Register default actions */
 		ast_manager_register_xml_core("Ping", 0, action_ping);