Migrate a large number of AMI events over to Stasis-Core

This patch moves a number of AMI events over to the Stasis-Core message bus.
This includes:
 * ChanSpyStart/Stop
 * MonitorStart/Stop
 * MusicOnHoldStart/Stop
 * FullyBooted/Reload
 * All Voicemail/MWI related events

In addition, it adds some Stasis-Core and AMI support for generic AMI messages,
refactors the message router in AMI to use a single router with topic
forwarding for the topics that AMI cares about, and refactors MWI message
types and topics to be more name compliant.

Review: https://reviewboard.asterisk.org/r/2532

(closes issue ASTERISK-21462)



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

7 files changed, 205 insertions, 190 deletions

diff --git a/res/parking/parking_manager.c b/res/parking/parking_manager.c
index d6a05573f..697468eb3 100644
--- a/res/parking/parking_manager.c
+++ b/res/parking/parking_manager.c
@@ -277,14 +277,14 @@ static struct ast_str *manager_build_parked_call_string(const struct ast_parked_
 		return NULL;
 	}
 
-	parkee_string = ast_manager_build_channel_state_string_suffix(payload->parkee, "Parkee");
+	parkee_string = ast_manager_build_channel_state_string_prefix(payload->parkee, "Parkee");
 
 	if (payload->parker) {
-		parker_string = ast_manager_build_channel_state_string_suffix(payload->parker, "Parker");
+		parker_string = ast_manager_build_channel_state_string_prefix(payload->parker, "Parker");
 	}
 
 	if (payload->retriever) {
-		retriever_string = ast_manager_build_channel_state_string_suffix(payload->retriever, "Retriever");
+		retriever_string = ast_manager_build_channel_state_string_prefix(payload->retriever, "Retriever");
 	}
 
 	ast_str_set(&out, 0,
diff --git a/res/res_fax.c b/res/res_fax.c
index 9afad4a19..4b8c9d7ca 100644
--- a/res/res_fax.c
+++ b/res/res_fax.c
@@ -83,11 +83,12 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/file.h"
 #include "asterisk/channel.h"
 #include "asterisk/pbx.h"
-#include "asterisk/manager.h"
 #include "asterisk/dsp.h"
 #include "asterisk/indications.h"
 #include "asterisk/ast_version.h"
 #include "asterisk/translate.h"
+#include "asterisk/stasis.h"
+#include "asterisk/stasis_channels.h"
 
 /*** DOCUMENTATION
 	<application name="ReceiveFAX" language="en_US" module="res_fax">
@@ -384,12 +385,6 @@ AST_APP_OPTIONS(fax_exec_options, BEGIN_OPTIONS
 	AST_APP_OPTION('z', OPT_REQUEST_T38),
 END_OPTIONS);
 
-struct manager_event_info {
-	char context[AST_MAX_CONTEXT];
-	char exten[AST_MAX_EXTENSION];
-	char cid[128];
-};
-
 static void debug_check_frame_for_silence(struct ast_fax_session *s, unsigned int c2s, struct ast_frame *frame)
 {
 	struct debug_info_history *history = c2s ? &s->debug_info->c2s : &s->debug_info->s2c;
@@ -1091,13 +1086,39 @@ static struct ast_fax_session *fax_session_new(struct ast_fax_session_details *d
 	return s;
 }
 
-static void get_manager_event_info(struct ast_channel *chan, struct manager_event_info *info)
+/*!
+ * \internal
+ * \brief Convert the filenames in a fax session into a JSON array
+ * \retval NULL on error
+ * \retval A \ref ast_json array on success
+ */
+static struct ast_json *generate_filenames_json(struct ast_fax_session_details *details)
 {
-	pbx_substitute_variables_helper(chan, "${CONTEXT}", info->context, sizeof(info->context));
-	pbx_substitute_variables_helper(chan, "${EXTEN}", info->exten, sizeof(info->exten));
-	pbx_substitute_variables_helper(chan, "${CALLERID(num)}", info->cid, sizeof(info->cid));
-}
+	RAII_VAR(struct ast_json *, json_array, ast_json_array_create(), ast_json_unref);
+	struct ast_fax_document *doc;
+
+	if (!details || !json_array) {
+		return NULL;
+	}
+
+	/* don't process empty lists */
+	if (AST_LIST_EMPTY(&details->documents)) {
+		return NULL;
+	}
+
+	AST_LIST_TRAVERSE(&details->documents, doc, next) {
+		struct ast_json *entry = ast_json_string_create(doc->filename);
+		if (!entry) {
+			return NULL;
+		}
+		if (ast_json_array_append(json_array, entry)) {
+			return NULL;
+		}
+	}
 
+	ast_json_ref(json_array);
+	return json_array;
+}
 
 /* \brief Generate a string of filenames using the given prefix and separator.
  * \param details the fax session details
@@ -1149,39 +1170,39 @@ static char *generate_filenames_string(struct ast_fax_session_details *details,
 /*! \brief send a FAX status manager event */
 static int report_fax_status(struct ast_channel *chan, struct ast_fax_session_details *details, const char *status)
 {
-	char *filenames = generate_filenames_string(details, "FileName: ", "\r\n");
+	RAII_VAR(struct ast_json *, json_object, NULL, ast_json_unref);
+	RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	struct ast_json *json_filenames = NULL;
 
-	ast_channel_lock(chan);
-	if (details->option.statusevents) {
-		struct manager_event_info info;
-
-		get_manager_event_info(chan, &info);
-		manager_event(EVENT_FLAG_CALL,
-			      "FAXStatus",
-			      "Operation: %s\r\n"
-			      "Status: %s\r\n"
-			      "Channel: %s\r\n"
-			      "Context: %s\r\n"
-			      "Exten: %s\r\n"
-			      "CallerID: %s\r\n"
-			      "LocalStationID: %s\r\n"
-			      "%s%s",
-			      (details->caps & AST_FAX_TECH_GATEWAY) ? "gateway" : (details->caps & AST_FAX_TECH_RECEIVE) ? "receive" : "send",
-			      status,
-			      ast_channel_name(chan),
-			      info.context,
-			      info.exten,
-			      info.cid,
-			      details->localstationid,
-			      S_OR(filenames, ""),
-			      filenames ? "\r\n" : "");
+	if (!details->option.statusevents) {
+		return 0;
 	}
-	ast_channel_unlock(chan);
 
-	if (filenames) {
-		ast_free(filenames);
+	json_filenames = generate_filenames_json(details);
+	if (!json_filenames) {
+		return -1;
+	}
+
+	json_object = ast_json_pack("{s: s, s: s, s: s, s: s, s: s, s: o}",
+			"type", "status",
+			"operation", (details->caps & AST_FAX_TECH_GATEWAY) ? "gateway" : (details->caps & AST_FAX_TECH_RECEIVE) ? "receive" : "send",
+			"status", status,
+			"local_station_id", details->localstationid,
+			"filenames", json_filenames);
+	if (!json_object) {
+		return -1;
 	}
 
+	{
+		SCOPED_CHANNELLOCK(lock, chan);
+
+		message = ast_channel_cached_blob_create(chan, ast_channel_fax_type(), json_object);
+		if (!message) {
+			return -1;
+		}
+		stasis_publish(ast_channel_topic(chan), message);
+	}
 	return 0;
 }
 
@@ -1738,13 +1759,53 @@ static int receivefax_t38_init(struct ast_channel *chan, struct ast_fax_session_
 	return 0;
 }
 
+/*! \brief Report on the final state of a receive fax operation
+ * \note This will lock the \ref ast_channel
+ */
+static int report_receive_fax_status(struct ast_channel *chan, const char *filename)
+{
+	RAII_VAR(struct ast_json *, json_object, NULL, ast_json_unref);
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_json *, json_array, ast_json_array_create(), ast_json_unref);
+	struct ast_json *json_filename = ast_json_string_create(filename);
+
+	if (!json_array || !json_filename) {
+		return -1;
+	}
+	ast_json_array_append(json_array, json_filename);
+
+	{
+		SCOPED_CHANNELLOCK(lock, chan);
+
+		json_object = ast_json_pack("s: s, s: s, s: s, s: s, s: s, s: s, s: s, s: o",
+				"type", "receive"
+				"remote_station_id", S_OR(pbx_builtin_getvar_helper(chan, "REMOTESTATIONID"), ""),
+				"local_station_id", S_OR(pbx_builtin_getvar_helper(chan, "LOCALSTATIONID"), ""),
+				"fax_pages", S_OR(pbx_builtin_getvar_helper(chan, "FAXPAGES"), ""),
+				"fax_resolution", S_OR(pbx_builtin_getvar_helper(chan, "FAXRESOLUTION"), ""),
+				"fax_bitrate", S_OR(pbx_builtin_getvar_helper(chan, "FAXBITRATE"), ""),
+				"filenames", json_array);
+		if (!json_object) {
+			return -1;
+		}
+
+		message = ast_channel_cached_blob_create(chan, ast_channel_fax_type(), json_object);
+		if (!message) {
+			return -1;
+		}
+
+		stasis_publish(ast_channel_topic(chan), message);
+	}
+	return 0;
+}
+
 /*! \brief initiate a receive FAX session */
 static int receivefax_exec(struct ast_channel *chan, const char *data)
 {
 	char *parse, modems[128] = "";
 	int channel_alive;
-	struct ast_fax_session_details *details;
-	struct ast_fax_session *s;
+	RAII_VAR(struct ast_fax_session_details *, details, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_fax_session *, s, NULL, ao2_cleanup);
 	struct ast_fax_tech_token *token = NULL;
 	struct ast_fax_document *doc;
 	AST_DECLARE_APP_ARGS(args,
@@ -1752,7 +1813,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		AST_APP_ARG(options);
 	);
 	struct ast_flags opts = { 0, };
-	struct manager_event_info info;
 	enum ast_t38_state t38state;
 
 	/* initialize output channel variables */
@@ -1780,7 +1840,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "can't receive a fax on a channel with a T.38 gateway");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "executing ReceiveFAX on a channel with a T.38 Gateway is not supported\n");
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1789,7 +1848,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "maxrate is less than minrate");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "maxrate %d is less than minrate %d\n", details->maxrate, details->minrate);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1799,7 +1857,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "incompatible 'modems' and 'minrate' settings");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1809,7 +1866,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "incompatible 'modems' and 'maxrate' settings");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1818,7 +1874,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s requires an argument (filename[,options])\n", app_receivefax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 	parse = ast_strdupa(data);
@@ -1829,7 +1884,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 	if (ast_strlen_zero(args.filename)) {
@@ -1837,7 +1891,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s requires an argument (filename[,options])\n", app_receivefax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1847,7 +1900,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s does not support polling\n", app_receivefax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1861,7 +1913,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "error allocating memory");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "System cannot provide memory for session requirements.\n");
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1894,7 +1945,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "error reserving fax session");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "Unable to reserve FAX session.\n");
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -1905,8 +1955,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 			set_channel_variables(chan, details);
 			ast_log(LOG_WARNING, "Channel '%s' failed answer attempt.\n", ast_channel_name(chan));
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			return -1;
 		}
 	}
@@ -1917,8 +1965,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error negotiating T.38");
 			set_channel_variables(chan, details);
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			return -1;
 		}
 	} else {
@@ -1931,8 +1977,6 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error negotiating T.38");
 			set_channel_variables(chan, details);
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			ast_log(LOG_ERROR, "error initializing channel '%s' in T.38 mode\n", ast_channel_name(chan));
 			return -1;
 		}
@@ -1948,36 +1992,9 @@ static int receivefax_exec(struct ast_channel *chan, const char *data)
 		}
 	}
 
-	/* send out the AMI completion event */
-	ast_channel_lock(chan);
-
-	get_manager_event_info(chan, &info);
-	manager_event(EVENT_FLAG_CALL,
-		      "ReceiveFAX",
-		      "Channel: %s\r\n"
-		      "Context: %s\r\n"
-		      "Exten: %s\r\n"
-		      "CallerID: %s\r\n"
-		      "RemoteStationID: %s\r\n"
-		      "LocalStationID: %s\r\n"
-		      "PagesTransferred: %s\r\n"
-		      "Resolution: %s\r\n"
-		      "TransferRate: %s\r\n"
-		      "FileName: %s\r\n",
-		      ast_channel_name(chan),
-		      info.context,
-		      info.exten,
-		      info.cid,
-		      S_OR(pbx_builtin_getvar_helper(chan, "REMOTESTATIONID"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "LOCALSTATIONID"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXPAGES"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXRESOLUTION"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXBITRATE"), ""),
-		      args.filename);
-	ast_channel_unlock(chan);
-
-	ao2_ref(s, -1);
-	ao2_ref(details, -1);
+	if (report_receive_fax_status(chan, args.filename)) {
+		ast_log(AST_LOG_ERROR, "Error publishing ReceiveFax status message\n");
+	}
 
 	/* If the channel hungup return -1; otherwise, return 0 to continue in the dialplan */
 	return (!channel_alive) ? -1 : 0;
@@ -2223,14 +2240,53 @@ static int sendfax_t38_init(struct ast_channel *chan, struct ast_fax_session_det
 	return 0;
 }
 
+/*!
+ * \brief Report on the status of a completed fax send attempt
+ * \note This will lock the \ref ast_channel
+ */
+static int report_send_fax_status(struct ast_channel *chan, struct ast_fax_session_details *details)
+{
+	RAII_VAR(struct ast_json *, json_obj, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	struct ast_json *json_filenames;
+
+	json_filenames = generate_filenames_json(details);
+	if (!json_filenames) {
+		return -1;
+	}
+
+	{
+		SCOPED_CHANNELLOCK(lock, chan);
+		json_obj = ast_json_pack("{s: s, s: s, s: s, s: s, s: s, s: s, s: s, s: o}",
+				"type", "send"
+				"remote_station_id", S_OR(pbx_builtin_getvar_helper(chan, "REMOTESTATIONID"), ""),
+				"local_station_id", S_OR(pbx_builtin_getvar_helper(chan, "LOCALSTATIONID"), ""),
+				"fax_pages", S_OR(pbx_builtin_getvar_helper(chan, "FAXPAGES"), ""),
+				"fax_resolution", S_OR(pbx_builtin_getvar_helper(chan, "FAXRESOLUTION"), ""),
+				"fax_bitrate", S_OR(pbx_builtin_getvar_helper(chan, "FAXBITRATE"), ""),
+				"filenames", json_filenames);
+		if (!json_obj) {
+			return -1;
+		}
+
+		message = ast_channel_cached_blob_create(chan, ast_channel_fax_type(), json_obj);
+		if (!message) {
+			return -1;
+		}
+		stasis_publish(ast_channel_topic(chan), message);
+	}
+	return 0;
+}
+
+
 
 /*! \brief initiate a send FAX session */
 static int sendfax_exec(struct ast_channel *chan, const char *data)
 {
 	char *parse, *filenames, *c, modems[128] = "";
 	int channel_alive, file_count;
-	struct ast_fax_session_details *details;
-	struct ast_fax_session *s;
+	RAII_VAR(struct ast_fax_session_details *, details, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_fax_session *, s, NULL, ao2_cleanup);
 	struct ast_fax_tech_token *token = NULL;
 	struct ast_fax_document *doc;
 	AST_DECLARE_APP_ARGS(args,
@@ -2238,7 +2294,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		AST_APP_ARG(options);
 	);
 	struct ast_flags opts = { 0, };
-	struct manager_event_info info;
 	enum ast_t38_state t38state;
 
 	/* initialize output channel variables */
@@ -2266,7 +2321,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "can't send a fax on a channel with a T.38 gateway");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "executing SendFAX on a channel with a T.38 Gateway is not supported\n");
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2275,7 +2329,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "maxrate is less than minrate");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "maxrate %d is less than minrate %d\n", details->maxrate, details->minrate);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2285,7 +2338,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "incompatible 'modems' and 'minrate' settings");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2295,7 +2347,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "incompatible 'modems' and 'maxrate' settings");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2304,7 +2355,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s requires an argument (filename[&filename[&filename]][,options])\n", app_sendfax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 	parse = ast_strdupa(data);
@@ -2316,7 +2366,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, error, "INVALID_ARGUMENTS");
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
-		ao2_ref(details, -1);
 		return -1;
 	}
 	if (ast_strlen_zero(args.filenames)) {
@@ -2324,7 +2373,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s requires an argument (filename[&filename[&filename]],options])\n", app_sendfax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2334,7 +2382,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "invalid arguments");
 		set_channel_variables(chan, details);
 		ast_log(LOG_WARNING, "%s does not support polling\n", app_sendfax);
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2348,7 +2395,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error reading file");
 			set_channel_variables(chan, details);
 			ast_log(LOG_ERROR, "access failure.  Verify '%s' exists and check permissions.\n", args.filenames);
-			ao2_ref(details, -1);
 			return -1;
 		}
 
@@ -2357,7 +2403,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error allocating memory");
 			set_channel_variables(chan, details);
 			ast_log(LOG_ERROR, "System cannot provide memory for session requirements.\n");
-			ao2_ref(details, -1);
 			return -1;
 		}
 
@@ -2402,7 +2447,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 		ast_string_field_set(details, resultstr, "error reserving fax session");
 		set_channel_variables(chan, details);
 		ast_log(LOG_ERROR, "Unable to reserve FAX session.\n");
-		ao2_ref(details, -1);
 		return -1;
 	}
 
@@ -2413,8 +2457,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 			set_channel_variables(chan, details);
 			ast_log(LOG_WARNING, "Channel '%s' failed answer attempt.\n", ast_channel_name(chan));
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			return -1;
 		}
 	}
@@ -2425,8 +2467,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error negotiating T.38");
 			set_channel_variables(chan, details);
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			return -1;
 		}
 	} else {
@@ -2439,8 +2479,6 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 			ast_string_field_set(details, resultstr, "error negotiating T.38");
 			set_channel_variables(chan, details);
 			fax_session_release(s, token);
-			ao2_ref(s, -1);
-			ao2_ref(details, -1);
 			ast_log(LOG_ERROR, "error initializing channel '%s' in T.38 mode\n", ast_channel_name(chan));
 			return -1;
 		}
@@ -2460,42 +2498,13 @@ static int sendfax_exec(struct ast_channel *chan, const char *data)
 
 	if (!(filenames = generate_filenames_string(details, "FileName: ", "\r\n"))) {
 		ast_log(LOG_ERROR, "Error generating SendFAX manager event\n");
-		ao2_ref(s, -1);
-		ao2_ref(details, -1);
 		return (!channel_alive) ? -1 : 0;
 	}
 
 	/* send out the AMI completion event */
-	ast_channel_lock(chan);
-	get_manager_event_info(chan, &info);
-	manager_event(EVENT_FLAG_CALL,
-		      "SendFAX",
-		      "Channel: %s\r\n"
-		      "Context: %s\r\n"
-		      "Exten: %s\r\n"
-		      "CallerID: %s\r\n"
-		      "RemoteStationID: %s\r\n"
-		      "LocalStationID: %s\r\n"
-		      "PagesTransferred: %s\r\n"
-		      "Resolution: %s\r\n"
-		      "TransferRate: %s\r\n"
-		      "%s\r\n",
-		      ast_channel_name(chan),
-		      info.context,
-		      info.exten,
-		      info.cid,
-		      S_OR(pbx_builtin_getvar_helper(chan, "REMOTESTATIONID"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "LOCALSTATIONID"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXPAGES"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXRESOLUTION"), ""),
-		      S_OR(pbx_builtin_getvar_helper(chan, "FAXBITRATE"), ""),
-		      filenames);
-	ast_channel_unlock(chan);
-
-	ast_free(filenames);
-
-	ao2_ref(s, -1);
-	ao2_ref(details, -1);
+	if (report_send_fax_status(chan, details)) {
+		ast_log(AST_LOG_ERROR, "Error publishing SendFAX status message\n");
+	}
 
 	/* If the channel hungup return -1; otherwise, return 0 to continue in the dialplan */
 	return (!channel_alive) ? -1 : 0;
diff --git a/res/res_jabber.c b/res/res_jabber.c
index 2070c8024..7ca0bf81e 100644
--- a/res/res_jabber.c
+++ b/res/res_jabber.c
@@ -1536,10 +1536,6 @@ static void aji_log_hook(void *data, const char *xmpp, size_t size, int is_incom
 {
 	struct aji_client *client = ASTOBJ_REF((struct aji_client *) data);
 
-	if (!ast_strlen_zero(xmpp)) {
-		manager_event(EVENT_FLAG_USER, "JabberEvent", "Account: %s\r\nPacket: %s\r\n", client->name, xmpp);
-	}
-
 	if (client->debug) {
 		if (is_incoming) {
 			ast_verbose("\nJABBER: %s INCOMING: %s\n", client->name, xmpp);
@@ -3247,9 +3243,9 @@ static void aji_mwi_cb(void *data, struct stasis_subscription *sub, struct stasi
 	char oldmsgs[10];
 	char newmsgs[10];
 	struct aji_client *client = data;
-	struct stasis_mwi_state *mwi_state;
+	struct ast_mwi_state *mwi_state;
 
-	if (!stasis_subscription_is_subscribed(sub) || stasis_mwi_state_type() != stasis_message_type(msg)) {
+	if (!stasis_subscription_is_subscribed(sub) || ast_mwi_state_type() != stasis_message_type(msg)) {
 		return;
 	}
 
@@ -3308,7 +3304,7 @@ static int cached_devstate_cb(void *obj, void *arg, int flags)
 static void aji_init_event_distribution(struct aji_client *client)
 {
 	if (!mwi_sub) {
-		mwi_sub = stasis_subscribe(stasis_mwi_topic_all(), aji_mwi_cb, client);
+		mwi_sub = stasis_subscribe(ast_mwi_topic_all(), aji_mwi_cb, client);
 	}
 	if (!device_state_sub) {
 		RAII_VAR(struct ao2_container *, cached, NULL, ao2_cleanup);
@@ -3369,7 +3365,7 @@ static int aji_handle_pubsub_event(void *data, ikspak *pak)
 		sscanf(iks_find_cdata(item_content, "OLDMSGS"), "%10d", &oldmsgs);
 		sscanf(iks_find_cdata(item_content, "NEWMSGS"), "%10d", &newmsgs);
 
-		stasis_publish_mwi_state_full(item_id, context, newmsgs, oldmsgs, &pubsub_eid);
+		ast_publish_mwi_state_full(item_id, context, newmsgs, oldmsgs, NULL, &pubsub_eid);
 
 		return IKS_FILTER_EAT;
 	} else {
diff --git a/res/res_monitor.c b/res/res_monitor.c
index 9aca24a0d..72911b5b1 100644
--- a/res/res_monitor.c
+++ b/res/res_monitor.c
@@ -40,8 +40,10 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/file.h"
 #include "asterisk/pbx.h"
 #include "asterisk/module.h"
-#include "asterisk/manager.h"
 #include "asterisk/cli.h"
+#include "asterisk/manager.h"
+#include "asterisk/stasis.h"
+#include "asterisk/stasis_channels.h"
 #define AST_API_MODULE
 #include "asterisk/monitor.h"
 #include "asterisk/app.h"
@@ -291,6 +293,7 @@ int AST_OPTIONAL_API_NAME(ast_monitor_start)(struct ast_channel *chan, const cha
 					     const char *fname_base, int need_lock, int stream_action)
 {
 	int res = 0;
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 
 	LOCK_IF_NEEDED(chan, need_lock);
 
@@ -393,11 +396,12 @@ int AST_OPTIONAL_API_NAME(ast_monitor_start)(struct ast_channel *chan, const cha
 		/* so we know this call has been monitored in case we need to bill for it or something */
 		pbx_builtin_setvar_helper(chan, "__MONITORED","true");
 
-		ast_manager_event(chan, EVENT_FLAG_CALL, "MonitorStart",
-			                "Channel: %s\r\n"
-					        "Uniqueid: %s\r\n",
-	                        ast_channel_name(chan),
-			                ast_channel_uniqueid(chan));
+		message = ast_channel_cached_blob_create(chan,
+				ast_channel_monitor_start_type(),
+				NULL);
+		if (message) {
+			stasis_publish(ast_channel_topic(chan), message);
+		}
 	} else {
 		ast_debug(1,"Cannot start monitoring %s, already monitored\n", ast_channel_name(chan));
 		res = -1;
@@ -437,6 +441,7 @@ static const char *get_soxmix_format(const char *format)
 int AST_OPTIONAL_API_NAME(ast_monitor_stop)(struct ast_channel *chan, int need_lock)
 {
 	int delfiles = 0;
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 
 	LOCK_IF_NEEDED(chan, need_lock);
 
@@ -511,12 +516,12 @@ int AST_OPTIONAL_API_NAME(ast_monitor_stop)(struct ast_channel *chan, int need_l
 		ast_free(ast_channel_monitor(chan));
 		ast_channel_monitor_set(chan, NULL);
 
-		ast_manager_event(chan, EVENT_FLAG_CALL, "MonitorStop",
-			                "Channel: %s\r\n"
-	                        "Uniqueid: %s\r\n",
-	                        ast_channel_name(chan),
-	                        ast_channel_uniqueid(chan)
-	                        );
+		message = ast_channel_cached_blob_create(chan,
+				ast_channel_monitor_stop_type(),
+				NULL);
+		if (message) {
+			stasis_publish(ast_channel_topic(chan), message);
+		}
 		pbx_builtin_setvar_helper(chan, "MONITORED", NULL);
 	}
 	pbx_builtin_setvar_helper(chan, "AUTO_MONITOR", NULL);
diff --git a/res/res_musiconhold.c b/res/res_musiconhold.c
index 130618431..2ed7ea52b 100644
--- a/res/res_musiconhold.c
+++ b/res/res_musiconhold.c
@@ -67,7 +67,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/cli.h"
 #include "asterisk/stringfields.h"
 #include "asterisk/linkedlists.h"
-#include "asterisk/manager.h"
+#include "asterisk/stasis.h"
+#include "asterisk/stasis_channels.h"
 #include "asterisk/paths.h"
 #include "asterisk/astobj2.h"
 #include "asterisk/timing.h"
@@ -1373,6 +1374,8 @@ static int local_ast_moh_start(struct ast_channel *chan, const char *mclass, con
 	struct mohclass *mohclass = NULL;
 	struct moh_files_state *state = ast_channel_music_state(chan);
 	struct ast_variable *var = NULL;
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_json *, json_object, NULL, ast_json_unref);
 	int res;
 	int realtime_possible = ast_check_realtime("musiconhold");
 
@@ -1567,14 +1570,6 @@ static int local_ast_moh_start(struct ast_channel *chan, const char *mclass, con
 		}
 	}
 
-	ast_manager_event(chan, EVENT_FLAG_CALL, "MusicOnHold",
-		"State: Start\r\n"
-		"Channel: %s\r\n"
-		"UniqueID: %s\r\n"
-		"Class: %s\r\n",
-		ast_channel_name(chan), ast_channel_uniqueid(chan),
-		mohclass->name);
-
 	ast_set_flag(ast_channel_flags(chan), AST_FLAG_MOH);
 
 	if (mohclass->total_files) {
@@ -1583,6 +1578,20 @@ static int local_ast_moh_start(struct ast_channel *chan, const char *mclass, con
 		res = ast_activate_generator(chan, &mohgen, mohclass);
 	}
 
+	json_object = ast_json_pack("{s: s}",
+			"class", mohclass->name);
+	if (!json_object) {
+		mohclass = mohclass_unref(mohclass, "unreffing local reference to mohclass in local_ast_moh_start");
+		return -1;
+	}
+
+	message = ast_channel_cached_blob_create(chan,
+			ast_channel_moh_start_type(),
+			json_object);
+	if (message) {
+		stasis_publish(ast_channel_topic(chan), message);
+	}
+
 	mohclass = mohclass_unref(mohclass, "unreffing local reference to mohclass in local_ast_moh_start");
 
 	return res;
@@ -1590,6 +1599,7 @@ static int local_ast_moh_start(struct ast_channel *chan, const char *mclass, con
 
 static void local_ast_moh_stop(struct ast_channel *chan)
 {
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 	ast_clear_flag(ast_channel_flags(chan), AST_FLAG_MOH);
 	ast_deactivate_generator(chan);
 
@@ -1601,11 +1611,10 @@ static void local_ast_moh_stop(struct ast_channel *chan)
 		}
 	}
 
-	ast_manager_event(chan, EVENT_FLAG_CALL, "MusicOnHold",
-		"State: Stop\r\n"
-		"Channel: %s\r\n"
-		"UniqueID: %s\r\n",
-		ast_channel_name(chan), ast_channel_uniqueid(chan));
+	message = ast_channel_cached_blob_create(chan, ast_channel_moh_stop_type(), NULL);
+	if (message) {
+		stasis_publish(ast_channel_topic(chan), message);
+	}
 	ast_channel_unlock(chan);
 }
 
diff --git a/res/res_sip_mwi.c b/res/res_sip_mwi.c
index 588662f33..f18e56469 100644
--- a/res/res_sip_mwi.c
+++ b/res/res_sip_mwi.c
@@ -133,7 +133,7 @@ static struct mwi_stasis_subscription *mwi_stasis_subscription_alloc(const char
 		return NULL;
 	}
 
-	topic = stasis_mwi_topic(mailbox);
+	topic = ast_mwi_topic(mailbox);
 
 	/* Safe strcpy */
 	strcpy(mwi_stasis_sub->mailbox, mailbox);
@@ -237,9 +237,9 @@ static int get_message_count(void *obj, void *arg, int flags)
 	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
 	struct mwi_stasis_subscription *mwi_stasis = obj;
 	struct message_accumulator *counter = arg;
-	struct stasis_mwi_state *mwi_state;
+	struct ast_mwi_state *mwi_state;
 
-	msg = stasis_cache_get(stasis_mwi_topic_cached(), stasis_mwi_state_type(), mwi_stasis->mailbox);
+	msg = stasis_cache_get(ast_mwi_topic_cached(), ast_mwi_state_type(), mwi_stasis->mailbox);
 	if (!msg) {
 		return 0;
 	}
@@ -604,7 +604,7 @@ static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub,
 		return;
 	}
 
-	if (stasis_mwi_state_type() == stasis_message_type(msg)) {
+	if (ast_mwi_state_type() == stasis_message_type(msg)) {
 		struct ast_taskprocessor *serializer = mwi_sub->is_solicited ? ast_sip_subscription_get_serializer(mwi_sub->sip_sub) : NULL;
 		ao2_ref(mwi_sub, +1);
 		ast_sip_push_task(serializer, serialized_notify, mwi_sub);
diff --git a/res/res_xmpp.c b/res/res_xmpp.c
index dea837204..89eb45d18 100644
--- a/res/res_xmpp.c
+++ b/res/res_xmpp.c
@@ -1324,9 +1324,9 @@ static void xmpp_pubsub_mwi_cb(void *data, struct stasis_subscription *sub, stru
 	struct ast_xmpp_client *client = data;
 	const char *mailbox, *context;
 	char oldmsgs[10], newmsgs[10];
-	struct stasis_mwi_state *mwi_state;
+	struct ast_mwi_state *mwi_state;
 
-	if (!stasis_subscription_is_subscribed(sub) || stasis_mwi_state_type() != stasis_message_type(msg)) {
+	if (!stasis_subscription_is_subscribed(sub) || ast_mwi_state_type() != stasis_message_type(msg)) {
 		return;
 	}
 
@@ -1484,7 +1484,7 @@ static int xmpp_pubsub_handle_event(void *data, ikspak *pak)
 		sscanf(iks_find_cdata(item_content, "OLDMSGS"), "%10d", &oldmsgs);
 		sscanf(iks_find_cdata(item_content, "NEWMSGS"), "%10d", &newmsgs);
 
-		stasis_publish_mwi_state_full(item_id, context, newmsgs, oldmsgs, &pubsub_eid);
+		ast_publish_mwi_state_full(item_id, context, newmsgs, oldmsgs, NULL, &pubsub_eid);
 
 		return IKS_FILTER_EAT;
 	} else {
@@ -1596,7 +1596,7 @@ static void xmpp_init_event_distribution(struct ast_xmpp_client *client)
 	xmpp_pubsub_unsubscribe(client, "device_state");
 	xmpp_pubsub_unsubscribe(client, "message_waiting");
 
-	if (!(client->mwi_sub = stasis_subscribe(stasis_mwi_topic_all(), xmpp_pubsub_mwi_cb, client))) {
+	if (!(client->mwi_sub = stasis_subscribe(ast_mwi_topic_all(), xmpp_pubsub_mwi_cb, client))) {
 		return;
 	}
 
@@ -2546,10 +2546,6 @@ static void xmpp_log_hook(void *data, const char *xmpp, size_t size, int incomin
 	RAII_VAR(struct ast_xmpp_client_config *, clientcfg, NULL, ao2_cleanup);
 	struct ast_xmpp_client *client = data;
 
-	if (!ast_strlen_zero(xmpp)) {
-		manager_event(EVENT_FLAG_USER, "JabberEvent", "Account: %s\r\nPacket: %s\r\n", client->name, xmpp);
-	}
-
 	if (!debug && (!cfg || !cfg->clients || !(clientcfg = xmpp_config_find(cfg->clients, client->name)) || !ast_test_flag(&clientcfg->flags, XMPP_DEBUG))) {
 		return;
 	}