[asterisk-commits] qwell: branch qwell/agi_events r390380 - /team/qwell/agi_events/res/res_agi.c

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Mon Jun 3 13:50:12 CDT 2013


Author: qwell
Date: Mon Jun  3 13:50:10 2013
New Revision: 390380

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=390380
Log:
Move AGI events to Stasis.

Modified:
    team/qwell/agi_events/res/res_agi.c

Modified: team/qwell/agi_events/res/res_agi.c
URL: http://svnview.digium.com/svn/asterisk/team/qwell/agi_events/res/res_agi.c?view=diff&rev=390380&r1=390379&r2=390380
==============================================================================
--- team/qwell/agi_events/res/res_agi.c (original)
+++ team/qwell/agi_events/res/res_agi.c Mon Jun  3 13:50:10 2013
@@ -66,6 +66,8 @@
 #include "asterisk/srv.h"
 #include "asterisk/test.h"
 #include "asterisk/netsock2.h"
+#include "asterisk/stasis_channels.h"
+#include "asterisk/stasis_message_router.h"
 
 #define AST_API_MODULE
 #include "asterisk/agi.h"
@@ -1019,6 +1021,55 @@
 	AGI_RESULT_HANGUP,
 };
 
+struct stasis_message_type *agi_exec_start_type(void);
+struct stasis_message_type *agi_exec_end_type(void);
+struct stasis_message_type *agi_async_start_type(void);
+struct stasis_message_type *agi_async_exec_type(void);
+struct stasis_message_type *agi_async_end_type(void);
+
+STASIS_MESSAGE_TYPE_DEFN(agi_exec_start_type);
+STASIS_MESSAGE_TYPE_DEFN(agi_exec_end_type);
+STASIS_MESSAGE_TYPE_DEFN(agi_async_start_type);
+STASIS_MESSAGE_TYPE_DEFN(agi_async_exec_type);
+STASIS_MESSAGE_TYPE_DEFN(agi_async_end_type);
+
+static void agi_channel_manager_event(void *data,
+	struct stasis_subscription *sub, struct stasis_topic *topic,
+	struct stasis_message *message)
+{
+	const char *type = data;
+	struct ast_channel_blob *obj = stasis_message_data(message);
+	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
+	RAII_VAR(struct ast_str *, event_string, NULL, ast_free);
+
+	channel_event_string = ast_manager_build_channel_state_string(obj->snapshot);
+	if (!channel_event_string) {
+		return;
+	}
+
+	event_string = ast_manager_str_from_json_object(obj->blob, NULL);
+	if (!event_string) {
+		return;
+	}
+
+	manager_event(EVENT_FLAG_AGI, type,
+		"%s"
+		"%s",
+		ast_str_buffer(channel_event_string),
+		ast_str_buffer(event_string));
+}
+
+static void agi_publish_channel_blob(struct ast_channel *chan, struct stasis_message_type *type, struct ast_json *blob)
+{
+	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+	msg = ast_channel_blob_create(chan, type, blob);
+	if (!msg) {
+		return;
+	}
+	stasis_publish(ast_channel_topic(chan), msg);
+}
+
 static agi_command *find_command(const char * const cmds[], int exact);
 
 AST_THREADSTORAGE(agi_buf);
@@ -1356,6 +1407,7 @@
 	char ami_buffer[AMI_BUF_SIZE];
 	enum agi_result returnstatus = AGI_RESULT_SUCCESS;
 	AGI async_agi;
+	RAII_VAR(struct ast_json *, startblob, NULL, ast_json_unref);
 
 	if (efd) {
 		ast_log(LOG_WARNING, "Async AGI does not support Enhanced AGI yet\n");
@@ -1405,14 +1457,9 @@
 	   care of AGI commands on this channel can decide which AGI commands
 	   to execute based on the setup info */
 	ast_uri_encode(agi_buffer, ami_buffer, AMI_BUF_SIZE, ast_uri_http);
-	manager_event(EVENT_FLAG_AGI, "AsyncAGI",
-		"SubEvent: Start\r\n"
-		"Channel: %s\r\n"
-		"Uniqueid: %s\r\n"
-		"Env: %s\r\n",
-		ast_channel_name(chan),
-		ast_channel_uniqueid(chan),
-		ami_buffer);
+	startblob = ast_json_pack("{s: s}", "Env", ami_buffer);
+	agi_publish_channel_blob(chan, agi_async_start_type(), startblob);
+
 	hungup = ast_check_hangup(chan);
 	for (;;) {
 		/*
@@ -1420,6 +1467,7 @@
 		 * the manager or the cli threads.
 		 */
 		while (!hungup) {
+			RAII_VAR(struct ast_json *, execblob, NULL, ast_json_unref);
 			res = get_agi_cmd(chan, &cmd);
 
 			if (res) {
@@ -1451,27 +1499,13 @@
 			 */
 			agi_buffer[res] = '\0';
 			ast_uri_encode(agi_buffer, ami_buffer, AMI_BUF_SIZE, ast_uri_http);
-			if (ast_strlen_zero(cmd->cmd_id)) {
-				manager_event(EVENT_FLAG_AGI, "AsyncAGI",
-					"SubEvent: Exec\r\n"
-					"Channel: %s\r\n"
-					"Uniqueid: %s\r\n"
-					"Result: %s\r\n",
-					ast_channel_name(chan),
-					ast_channel_uniqueid(chan),
-					ami_buffer);
-			} else {
-				manager_event(EVENT_FLAG_AGI, "AsyncAGI",
-					"SubEvent: Exec\r\n"
-					"Channel: %s\r\n"
-					"Uniqueid: %s\r\n"
-					"CommandID: %s\r\n"
-					"Result: %s\r\n",
-					ast_channel_name(chan),
-					ast_channel_uniqueid(chan),
-					cmd->cmd_id,
-					ami_buffer);
+
+			execblob = ast_json_pack("{s: s}", "Result", ami_buffer);
+			if (!ast_strlen_zero(cmd->cmd_id)) {
+				ast_json_object_set(execblob, "CommandId", ast_json_string_create(cmd->cmd_id));
 			}
+			agi_publish_channel_blob(chan, agi_async_exec_type(), execblob);
+
 			free_agi_cmd(cmd);
 
 			/*
@@ -1530,12 +1564,7 @@
 		ast_speech_destroy(async_agi.speech);
 	}
 	/* notify manager users this channel cannot be controlled anymore by Async AGI */
-	manager_event(EVENT_FLAG_AGI, "AsyncAGI",
-		"SubEvent: End\r\n"
-		"Channel: %s\r\n"
-		"Uniqueid: %s\r\n",
-		ast_channel_name(chan),
-		ast_channel_uniqueid(chan));
+	agi_publish_channel_blob(chan, agi_async_end_type(), NULL);
 
 async_agi_abort:
 	/* close the pipe */
@@ -3572,21 +3601,17 @@
 	int argc = MAX_ARGS;
 	int res;
 	agi_command *c;
+	char *ami_cmd = ast_strdupa(buf);
 	const char *ami_res;
-	char *ami_cmd = ast_strdupa(buf);
 	int command_id = ast_random();
-	int resultcode;
-
-	manager_event(EVENT_FLAG_AGI, "AGIExec",
-		"SubEvent: Start\r\n"
-		"Channel: %s\r\n"
-		"Uniqueid: %s\r\n"
-		"CommandId: %d\r\n"
-		"Command: %s\r\n",
-		ast_channel_name(chan),
-		ast_channel_uniqueid(chan),
-		command_id,
-		ami_cmd);
+	int resultcode = 0;
+	RAII_VAR(struct ast_json *, startblob, NULL, ast_json_unref);
+
+	startblob = ast_json_pack("{s: i, s: s}",
+			     "CommandId", command_id,
+			     "Command", ami_cmd);
+	agi_publish_channel_blob(chan, agi_exec_start_type(), startblob);
+
 	parse_args(buf, &argc, argv);
 	c = find_command(argv, 0);
 	if (c && (!dead || (dead && c->dead))) {
@@ -3621,20 +3646,7 @@
 			resultcode = 200;
 			break;
 		}
-		manager_event(EVENT_FLAG_AGI, "AGIExec",
-			"SubEvent: End\r\n"
-			"Channel: %s\r\n"
-			"Uniqueid: %s\r\n"
-			"CommandId: %d\r\n"
-			"Command: %s\r\n"
-			"ResultCode: %d\r\n"
-			"Result: %s\r\n",
-			ast_channel_name(chan),
-			ast_channel_uniqueid(chan),
-			command_id,
-			ami_cmd,
-			resultcode,
-			ami_res);
+
 		switch (res) {
 		case RESULT_SHOWUSAGE:
 			if (ast_strlen_zero(c->usage)) {
@@ -3654,33 +3666,25 @@
 			break;
 		}
 	} else if (c) {
-		ast_agi_send(agi->fd, chan, "511 Command Not Permitted on a dead channel\n");
-		manager_event(EVENT_FLAG_AGI, "AGIExec",
-			"SubEvent: End\r\n"
-			"Channel: %s\r\n"
-			"Uniqueid: %s\r\n"
-			"CommandId: %d\r\n"
-			"Command: %s\r\n"
-			"ResultCode: 511\r\n"
-			"Result: Command not permitted on a dead channel\r\n",
-			ast_channel_name(chan),
-			ast_channel_uniqueid(chan),
-			command_id,
-			ami_cmd);
+		ami_res = "Command Not Permitted on a dead channel";
+		resultcode = 511;
+
+		ast_agi_send(agi->fd, chan, "%d %s\n", resultcode, ami_res);
 	} else {
-		ast_agi_send(agi->fd, chan, "510 Invalid or unknown command\n");
-		manager_event(EVENT_FLAG_AGI, "AGIExec",
-			"SubEvent: End\r\n"
-			"Channel: %s\r\n"
-			"Uniqueid: %s\r\n"
-			"CommandId: %d\r\n"
-			"Command: %s\r\n"
-			"ResultCode: 510\r\n"
-			"Result: Invalid or unknown command\r\n",
-			ast_channel_name(chan),
-			ast_channel_uniqueid(chan),
-			command_id,
-			ami_cmd);
+		ami_res = "Invalid or unknown command";
+		resultcode = 510;
+
+		ast_agi_send(agi->fd, chan, "%d %s\n", resultcode, ami_res);
+	}
+
+	if (resultcode != 0) {
+		RAII_VAR(struct ast_json *, endblob, NULL, ast_json_unref);
+		endblob = ast_json_pack("{s: i, s: s, s: i, s: s}",
+				     "CommandId", command_id,
+				     "Command", ami_cmd,
+				     "ResultCode", resultcode,
+				     "Result", ami_res);
+		agi_publish_channel_blob(chan, agi_exec_end_type(), endblob);
 	}
 	return AGI_RESULT_SUCCESS;
 }
@@ -4223,6 +4227,12 @@
 
 static int unload_module(void)
 {
+	STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_start_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_end_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(agi_async_start_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(agi_async_exec_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(agi_async_end_type);
+
 	ast_cli_unregister_multiple(cli_agi, ARRAY_LEN(cli_agi));
 	/* we can safely ignore the result of ast_agi_unregister_multiple() here, since it cannot fail, as
 	   we know that these commands were registered by this module and are still registered
@@ -4237,6 +4247,44 @@
 
 static int load_module(void)
 {
+	struct stasis_message_router *message_router;
+
+	message_router = ast_manager_get_message_router();
+	if (!message_router) {
+		return AST_MODULE_LOAD_DECLINE;
+	}
+
+	STASIS_MESSAGE_TYPE_INIT(agi_exec_start_type);
+	STASIS_MESSAGE_TYPE_INIT(agi_exec_end_type);
+	STASIS_MESSAGE_TYPE_INIT(agi_async_start_type);
+	STASIS_MESSAGE_TYPE_INIT(agi_async_exec_type);
+	STASIS_MESSAGE_TYPE_INIT(agi_async_end_type);
+
+	stasis_message_router_add(message_router,
+				  agi_exec_start_type(),
+				  agi_channel_manager_event,
+				  "AGIExecStart");
+
+	stasis_message_router_add(message_router,
+				  agi_exec_end_type(),
+				  agi_channel_manager_event,
+				  "AGIExecEnd");
+
+	stasis_message_router_add(message_router,
+				  agi_async_start_type(),
+				  agi_channel_manager_event,
+				  "AsyncAGIStart");
+
+	stasis_message_router_add(message_router,
+				  agi_async_exec_type(),
+				  agi_channel_manager_event,
+				  "AsyncAGIExec");
+
+	stasis_message_router_add(message_router,
+				  agi_async_end_type(),
+				  agi_channel_manager_event,
+				  "AsyncAGIEnd");
+
 	ast_cli_register_multiple(cli_agi, ARRAY_LEN(cli_agi));
 	/* we can safely ignore the result of ast_agi_register_multiple() here, since it cannot fail, as
 	   no other commands have been registered yet




More information about the asterisk-commits mailing list