Stasis: Fix StasisEnd message ordering

This change corrects message ordering in cases where a channel-related
message can be received after a Stasis/ARI application has received the
StasisEnd message. The StasisEnd message was being passed to
applications directly without waiting for the channel topic to empty.

As a result of this fix, other bugs were also identified and fixed:
* StasisStart messages were also being sent directly to apps and are
  now routed through the stasis message bus properly
* Masquerade monitor datastores were being removed at the incorrect
  time in some cases and were causing StasisEnd messages to not be sent
* General refactoring where necessary for the above
* Unsubscription on StasisEnd timing changes to prevent additional
  messages from following the StasisEnd when they shouldn't

A channel sanitization function pointer was added to reduce processing
and AO2 lookups.

Review: https://reviewboard.asterisk.org/r/4163/
ASTERISK-24501 #close
Reported by: Matt Jordan
........

Merged revisions 427788 from http://svn.asterisk.org/svn/asterisk/branches/12
........

Merged revisions 427789 from http://svn.asterisk.org/svn/asterisk/branches/13


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

3 files changed, 28 insertions, 6 deletions

diff --git a/res/stasis/app.c b/res/stasis/app.c
index 9440cf1bd..cda1c045d 100644
--- a/res/stasis/app.c
+++ b/res/stasis/app.c
@@ -302,6 +302,10 @@ static void sub_default_handler(void *data, struct stasis_subscription *sub,
 		call_forwarded_handler(app, message);
 	}
 
+	if (stasis_message_type(message) == app_end_message_type()) {
+		app_end_message_handler(message);
+	}
+
 	/* By default, send any message that has a JSON representation */
 	json = stasis_message_to_json(message, stasis_app_get_sanitizer());
 	if (!json) {
diff --git a/res/stasis/app.h b/res/stasis/app.h
index 1ab6097a7..63143e026 100644
--- a/res/stasis/app.h
+++ b/res/stasis/app.h
@@ -270,4 +270,27 @@ char *app_get_replace_channel_app(struct ast_channel *chan);
  */
 int app_replace_channel_forwards(struct stasis_app *app, const char *old_id, struct ast_channel *new_chan);
 
+/*!
+ * \brief Send StasisEnd message to the listening app
+ *
+ * \param app The app that owns the channel
+ * \param chan The channel for which the message is being sent
+ *
+ * \retval zero on success
+ * \return non-zero on failure
+ */
+int app_send_end_msg(struct stasis_app *app, struct ast_channel *chan);
+
+/*!
+ * \brief Handle cleanup related to StasisEnd messages
+ *
+ * \param message The message for which to clean up
+ */
+void app_end_message_handler(struct stasis_message *message);
+
+/*!
+ * \brief Accessor for the StasisEnd message type
+ */
+struct stasis_message_type *app_end_message_type(void);
+
 #endif /* _ASTERISK_RES_STASIS_APP_H */
diff --git a/res/stasis/stasis_bridge.c b/res/stasis/stasis_bridge.c
index 7229a87d5..fd984ab86 100644
--- a/res/stasis/stasis_bridge.c
+++ b/res/stasis/stasis_bridge.c
@@ -164,7 +164,6 @@ static int bridge_stasis_moving(struct ast_bridge_channel *bridge_channel, void
 {
 	if (src->v_table == &bridge_stasis_v_table &&
 			dst->v_table != &bridge_stasis_v_table) {
-		RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 		RAII_VAR(struct stasis_app_control *, control, NULL, ao2_cleanup);
 		struct ast_channel *chan;
 
@@ -176,11 +175,7 @@ static int bridge_stasis_moving(struct ast_bridge_channel *bridge_channel, void
 			return -1;
 		}
 
-		blob = ast_json_pack("{s: s}", "app", app_name(control_app(control)));
-
-		stasis_app_channel_set_stasis_end_published(chan);
-
-		ast_channel_publish_blob(chan, ast_stasis_end_message_type(), blob);
+		app_send_end_msg(control_app(control), chan);
 	}
 
 	return -1;