Stasis: Convey transfer information to applications

This fixes a class of issues where Stasis applications were not made
aware that their channels were being manipulated or replaced by
external entitiessuch as transfers, AMI commands, or dialplan
applications such as Bridge(). Inconsistent information such as
StasisEnd events with unknown channels as a result of masquerades has
also been corrected. To accomplish these fixes, several new fields
were added to blind and attended transfer messages as well as
StasisStart and BridgeAttendedTransfer Stasis events.

ASTERISK-23941 #close
Review: https://reviewboard.asterisk.org/r/3865/
Review: https://reviewboard.asterisk.org/r/3857/
Review: https://reviewboard.asterisk.org/r/3852/
Review: https://reviewboard.asterisk.org/r/3816/
Review: https://reviewboard.asterisk.org/r/3731/
Review: https://reviewboard.asterisk.org/r/3729/
Review: https://reviewboard.asterisk.org/r/3728/
........

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


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

7 files changed, 363 insertions, 12 deletions

diff --git a/res/stasis/app.c b/res/stasis/app.c
index 7e7911b9c..745969615 100644
--- a/res/stasis/app.c
+++ b/res/stasis/app.c
@@ -28,6 +28,7 @@
 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 
 #include "app.h"
+#include "control.h"
 #include "messaging.h"
 
 #include "asterisk/callerid.h"
@@ -699,14 +700,32 @@ static void bridge_blind_transfer_handler(void *data, struct stasis_subscription
 	struct stasis_message *message)
 {
 	struct stasis_app *app = data;
-	struct ast_bridge_blob *blob = stasis_message_data(message);
+	struct ast_blind_transfer_message *transfer_msg = stasis_message_data(message);
+	struct ast_bridge_snapshot *bridge = transfer_msg->to_transferee.bridge_snapshot;
 
-	if (bridge_app_subscribed(app, blob->channel->uniqueid) ||
-		(blob->bridge && bridge_app_subscribed_involved(app, blob->bridge))) {
+	if (bridge_app_subscribed(app, transfer_msg->to_transferee.channel_snapshot->uniqueid) ||
+		(bridge && bridge_app_subscribed_involved(app, bridge))) {
 		stasis_publish(app->topic, message);
 	}
 }
 
+static void set_replacement_channel(struct ast_channel_snapshot *to_be_replaced,
+		struct ast_channel_snapshot *replacing)
+{
+	struct stasis_app_control *control = stasis_app_control_find_by_channel_id(
+		to_be_replaced->uniqueid);
+	struct ast_channel *chan = ast_channel_get_by_name(replacing->uniqueid);
+
+	if (control && chan) {
+		ast_channel_lock(chan);
+		app_set_replace_channel_app(chan, app_name(control_app(control)));
+		app_set_replace_channel_snapshot(chan, to_be_replaced);
+		ast_channel_unlock(chan);
+	}
+	ast_channel_cleanup(chan);
+	ao2_cleanup(control);
+}
+
 static void bridge_attended_transfer_handler(void *data, struct stasis_subscription *sub,
 	struct stasis_message *message)
 {
@@ -751,6 +770,18 @@ static void bridge_attended_transfer_handler(void *data, struct stasis_subscript
 	if (subscribed) {
 		stasis_publish(app->topic, message);
 	}
+
+	if (transfer_msg->replace_channel) {
+		set_replacement_channel(transfer_msg->to_transferee.channel_snapshot,
+				transfer_msg->replace_channel);
+	}
+
+	if (transfer_msg->dest_type == AST_ATTENDED_TRANSFER_DEST_LINK) {
+		set_replacement_channel(transfer_msg->to_transferee.channel_snapshot,
+				transfer_msg->dest.links[0]);
+		set_replacement_channel(transfer_msg->to_transfer_target.channel_snapshot,
+				transfer_msg->dest.links[1]);
+	}
 }
 
 static void bridge_default_handler(void *data, struct stasis_subscription *sub,
@@ -1091,6 +1122,30 @@ int app_is_subscribed_channel_id(struct stasis_app *app, const char *channel_id)
 	return forwards != NULL;
 }
 
+int app_replace_channel_forwards(struct stasis_app *app, const char *old_id, struct ast_channel *new_chan)
+{
+	RAII_VAR(struct app_forwards *, old_forwards, NULL, ao2_cleanup);
+	struct app_forwards *new_forwards;
+
+	old_forwards = ao2_find(app->forwards, old_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
+	if (!old_forwards) {
+		return -1;
+	}
+
+	new_forwards = forwards_create_channel(app, new_chan);
+	if (!new_forwards) {
+		return -1;
+	}
+
+	new_forwards->interested = old_forwards->interested;
+	ao2_link_flags(app->forwards, new_forwards, 0);
+	ao2_cleanup(new_forwards);
+
+	/* Clean up old forwards */
+	forwards_unsubscribe(old_forwards);
+	return 0;
+}
+
 static void *channel_find(const struct stasis_app *app, const char *id)
 {
 	return ast_channel_get_by_name(id);
diff --git a/res/stasis/app.h b/res/stasis/app.h
index 419ec54a8..1ab6097a7 100644
--- a/res/stasis/app.h
+++ b/res/stasis/app.h
@@ -226,4 +226,48 @@ int app_unsubscribe_endpoint_id(struct stasis_app *app, const char *endpoint_id)
  */
 int app_is_subscribed_endpoint_id(struct stasis_app *app, const char *endpoint_id);
 
+/*!
+ * \brief Set the snapshot of the channel that this channel will replace
+ *
+ * \param channel The channel on which this will be set
+ * \param replace_snapshot The snapshot of the channel that is being replaced
+ *
+ * \retval zero success
+ * \retval non-zero failure
+ */
+int app_set_replace_channel_snapshot(struct ast_channel *chan, struct ast_channel_snapshot *replace_snapshot);
+
+/*!
+ * \brief Set the app that the replacement channel will be controlled by
+ *
+ * \param channel The channel on which this will be set
+ * \param replace_app The app that will be controlling this channel
+ *
+ * \retval zero success
+ * \retval non-zero failure
+ */
+int app_set_replace_channel_app(struct ast_channel *chan, const char *replace_app);
+
+/*!
+ * \brief Get the app that the replacement channel will be controlled by
+ *
+ * \param channel The channel on which this will be set
+ *
+ * \retval NULL on error
+ * \return the name of the controlling app (must be ast_free()d)
+ */
+char *app_get_replace_channel_app(struct ast_channel *chan);
+
+/*!
+ * \brief Replace channel topic forwards for the old channel with forwards for the new channel
+ *
+ * \param app The app that owns the channel
+ * \param old_id The unique ID of the channel to be replaced
+ * \param new_chan The channel that is replacing the old one
+ *
+ * \retval zero on success
+ * \return non-zero on failure
+ */
+int app_replace_channel_forwards(struct stasis_app *app, const char *old_id, struct ast_channel *new_chan);
+
 #endif /* _ASTERISK_RES_STASIS_APP_H */
diff --git a/res/stasis/command.c b/res/stasis/command.c
index a9e53af12..867de180a 100644
--- a/res/stasis/command.c
+++ b/res/stasis/command.c
@@ -93,3 +93,61 @@ void command_invoke(struct stasis_app_command *command,
 	command_complete(command, retval);
 }
 
+static void command_queue_prestart_destroy(void *obj)
+{
+	/* Clean up the container */
+	ao2_cleanup(obj);
+}
+
+static const struct ast_datastore_info command_queue_prestart = {
+        .type = "stasis-command-prestart-queue",
+        .destroy = command_queue_prestart_destroy,
+};
+
+int command_prestart_queue_command(struct ast_channel *chan,
+	stasis_app_command_cb command_fn, void *data)
+{
+	struct ast_datastore *datastore;
+	struct ao2_container *command_queue;
+	RAII_VAR(struct stasis_app_command *, command,
+		command_create(command_fn, data), ao2_cleanup);
+
+	if (!command) {
+		return -1;
+	}
+
+        datastore = ast_channel_datastore_find(chan, &command_queue_prestart, NULL);
+        if (datastore) {
+		command_queue = datastore->data;
+		ao2_link(command_queue, command);
+                return 0;
+        }
+
+	command_queue = ao2_container_alloc_list(
+                AO2_ALLOC_OPT_LOCK_MUTEX, 0, NULL, NULL);
+	if (!command_queue) {
+		return -1;
+	}
+
+	datastore = ast_datastore_alloc(&command_queue_prestart, NULL);
+	if (!datastore) {
+		ao2_cleanup(command_queue);
+		return -1;
+	}
+	ast_channel_datastore_add(chan, datastore);
+
+	datastore->data = command_queue;
+	ao2_link(command_queue, command);
+
+	return 0;
+}
+
+struct ao2_container *command_prestart_get_container(struct ast_channel *chan)
+{
+        struct ast_datastore *datastore = ast_channel_datastore_find(chan, &command_queue_prestart, NULL);
+	if (!datastore) {
+		return NULL;
+	}
+
+	return ao2_bump(datastore->data);
+}
diff --git a/res/stasis/command.h b/res/stasis/command.h
index a99d40d0a..7f12ab36f 100644
--- a/res/stasis/command.h
+++ b/res/stasis/command.h
@@ -41,4 +41,31 @@ void command_invoke(struct stasis_app_command *command,
 
 int command_join(struct stasis_app_command *command);
 
+/*!
+ * \brief Queue a Stasis() prestart command for a channel
+ *
+ * \pre chan must be locked
+ *
+ * \param chan The channel on which to queue the prestart command
+ * \param command_fn The callback to call for the command
+ * \param data The data to pass to the command callback
+ *
+ * \retval zero on success
+ * \retval non-zero on failure
+ */
+int command_prestart_queue_command(struct ast_channel *chan,
+	stasis_app_command_cb command_fn, void *data);
+
+/*!
+ * \brief Get the Stasis() prestart commands for a channel
+ *
+ * \pre chan must be locked
+ *
+ * \param chan The channel from which to get prestart commands
+ *
+ * \return The command prestart container for chan (must be ao2_cleanup()'d)
+ */
+struct ao2_container *command_prestart_get_container(struct ast_channel *chan);
+
+
 #endif /* _ASTERISK_RES_STASIS_CONTROL_H */
diff --git a/res/stasis/control.c b/res/stasis/control.c
index 8802e8128..0a9669d3b 100644
--- a/res/stasis/control.c
+++ b/res/stasis/control.c
@@ -276,10 +276,6 @@ struct stasis_app_control_dial_data {
 	int timeout;
 };
 
-static int app_control_add_channel_to_bridge(
-        struct stasis_app_control *control,
-        struct ast_channel *chan, void *data);
-
 static int app_control_dial(struct stasis_app_control *control,
 	struct ast_channel *chan, void *data)
 {
@@ -322,7 +318,7 @@ static int app_control_dial(struct stasis_app_control *control,
 		AST_BRIDGE_IMPART_CHAN_INDEPENDENT)) {
 		ast_hangup(new_chan);
 	} else {
-		app_control_add_channel_to_bridge(control, chan, bridge);
+		control_add_channel_to_bridge(control, chan, bridge);
 	}
 
 	return 0;
@@ -855,7 +851,7 @@ static void bridge_after_cb_failed(enum ast_bridge_after_cb_reason reason,
 		ast_bridge_after_cb_reason_string(reason));
 }
 
-static int app_control_add_channel_to_bridge(
+int control_add_channel_to_bridge(
 	struct stasis_app_control *control,
 	struct ast_channel *chan, void *data)
 {
@@ -935,7 +931,7 @@ int stasis_app_control_add_channel_to_bridge(
 			stasis_app_control_get_channel_id(control));
 
 	return app_send_command_on_condition(
-		control, app_control_add_channel_to_bridge, bridge,
+		control, control_add_channel_to_bridge, bridge,
 		app_control_can_add_channel_to_bridge);
 }
 
@@ -1036,3 +1032,36 @@ void control_wait(struct stasis_app_control *control)
 	}
 	ao2_unlock(control->command_queue);
 }
+
+int control_prestart_dispatch_all(struct stasis_app_control *control,
+	struct ast_channel *chan)
+{
+	struct ao2_container *command_queue;
+	int count = 0;
+	struct ao2_iterator iter;
+	struct stasis_app_command *command;
+
+	ast_channel_lock(chan);
+	command_queue = command_prestart_get_container(chan);
+	ast_channel_unlock(chan);
+	if (!command_queue) {
+		return 0;
+	}
+
+	iter = ao2_iterator_init(command_queue, AO2_ITERATOR_UNLINK);
+
+	while ((command = ao2_iterator_next(&iter))) {
+		command_invoke(command, control, chan);
+		ao2_cleanup(command);
+		++count;
+	}
+
+	ao2_iterator_destroy(&iter);
+	ao2_cleanup(command_queue);
+	return count;
+}
+
+struct stasis_app *control_app(struct stasis_app_control *control)
+{
+	return control->app;
+}
diff --git a/res/stasis/control.h b/res/stasis/control.h
index 0febd8438..a139f82e4 100644
--- a/res/stasis/control.h
+++ b/res/stasis/control.h
@@ -77,5 +77,36 @@ int control_is_done(struct stasis_app_control *control);
 
 void control_mark_done(struct stasis_app_control *control);
 
+/*!
+ * \brief Dispatch all queued prestart commands
+ *
+ * \param control The control for chan
+ * \param channel The channel on which commands should be executed
+ *
+ * \return The number of commands executed
+ */
+int control_prestart_dispatch_all(struct stasis_app_control *control,
+	struct ast_channel *chan);
+
+/*!
+ * \brief Returns the pointer (non-reffed) to the app associated with this control
+ *
+ * \param control Control to query.
+ *
+ * \returns A pointer to the associated stasis_app
+ */
+struct stasis_app *control_app(struct stasis_app_control *control);
+
+/*!
+ * \brief Command callback for adding a channel to a bridge
+ *
+ * \param control The control for chan
+ * \param channel The channel on which commands should be executed
+ * \param bridge Data to be passed to the callback
+ */
+int control_add_channel_to_bridge(
+	struct stasis_app_control *control,
+	struct ast_channel *chan, void *obj);
+
 
 #endif /* _ASTERISK_RES_STASIS_CONTROL_H */
diff --git a/res/stasis/stasis_bridge.c b/res/stasis/stasis_bridge.c
index c3a266a11..be7836d35 100644
--- a/res/stasis/stasis_bridge.c
+++ b/res/stasis/stasis_bridge.c
@@ -32,11 +32,73 @@
 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 
 #include "asterisk/bridge.h"
+#include "asterisk/bridge_after.h"
 #include "asterisk/bridge_internal.h"
+#include "asterisk/bridge_features.h"
+#include "asterisk/stasis_app.h"
+#include "asterisk/stasis_channels.h"
 #include "stasis_bridge.h"
+#include "control.h"
+#include "command.h"
+#include "app.h"
+#include "asterisk/stasis_app.h"
+#include "asterisk/pbx.h"
 
 /* ------------------------------------------------------------------- */
 
+static struct ast_bridge_methods bridge_stasis_v_table;
+
+static void bridge_stasis_run_cb(struct ast_channel *chan, void *data)
+{
+	RAII_VAR(char *, app_name, NULL, ast_free);
+	struct ast_app *app_stasis;
+
+	/* Take ownership of the swap_app memory from the datastore */
+	app_name = app_get_replace_channel_app(chan);
+	if (!app_name) {
+		ast_log(LOG_ERROR, "Failed to get app name for %s (%p)\n", ast_channel_name(chan), chan);
+		return;
+	}
+
+	/* find Stasis() */
+	app_stasis = pbx_findapp("Stasis");
+	if (!app_stasis) {
+		ast_log(LOG_WARNING, "Could not find application (Stasis)\n");
+		return;
+	}
+
+	if (ast_check_hangup_locked(chan)) {
+		/* channel hungup, don't run Stasis() */
+		return;
+	}
+
+	/* run Stasis() */
+	pbx_exec(chan, app_stasis, app_name);
+}
+
+static int add_channel_to_bridge(
+	struct stasis_app_control *control,
+	struct ast_channel *chan, void *obj)
+{
+	struct ast_bridge *bridge = obj;
+	int res;
+
+	res = control_add_channel_to_bridge(control,
+		chan, bridge);
+	ao2_cleanup(bridge);
+	return res;
+}
+
+static void bridge_stasis_queue_join_action(struct ast_bridge *self,
+	struct ast_bridge_channel *bridge_channel)
+{
+	ast_channel_lock(bridge_channel->chan);
+	if (command_prestart_queue_command(bridge_channel->chan, add_channel_to_bridge, ao2_bump(self))) {
+		ao2_cleanup(self);
+	}
+	ast_channel_unlock(bridge_channel->chan);
+}
+
 /*!
  * \internal
  * \brief Push this channel into the Stasis bridge.
@@ -53,6 +115,24 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
  */
 static int bridge_stasis_push(struct ast_bridge *self, struct ast_bridge_channel *bridge_channel, struct ast_bridge_channel *swap)
 {
+	struct stasis_app_control *control = stasis_app_control_find_by_channel(bridge_channel->chan);
+
+	if (!control) {
+		/* channel not in Stasis(), get it there */
+		/* Attach after-bridge callback and pass ownership of swap_app to it */
+		if (ast_bridge_set_after_callback(bridge_channel->chan,
+			bridge_stasis_run_cb, NULL, NULL)) {
+			ast_log(LOG_ERROR, "Failed to set after bridge callback\n");
+			return -1;
+		}
+
+		bridge_stasis_queue_join_action(self, bridge_channel);
+
+		/* Return -1 so the push fails and the after-bridge callback gets called */
+		return -1;
+	}
+
+	ao2_cleanup(control);
 	if (self->allowed_capabilities & STASIS_BRIDGE_MIXING_CAPABILITIES) {
 		ast_bridge_channel_update_linkedids(bridge_channel, swap);
 		if (ast_test_flag(&self->feature_flags, AST_BRIDGE_FLAG_SMART)) {
@@ -63,6 +143,33 @@ static int bridge_stasis_push(struct ast_bridge *self, struct ast_bridge_channel
 	return ast_bridge_base_v_table.push(self, bridge_channel, swap);
 }
 
+static int bridge_stasis_moving(struct ast_bridge_channel *bridge_channel, void *hook_pvt,
+		struct ast_bridge *src, struct ast_bridge *dst)
+{
+	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;
+
+		chan = bridge_channel->chan;
+		ast_assert(chan != NULL);
+
+		control = stasis_app_control_find_by_channel(chan);
+		if (!control) {
+			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);
+	}
+
+	return -1;
+}
+
 /*!
  * \internal
  * \brief Pull this channel from the Stasis bridge.
@@ -82,11 +189,11 @@ static void bridge_stasis_pull(struct ast_bridge *self, struct ast_bridge_channe
 		ast_bridge_channel_update_accountcodes(NULL, bridge_channel);
 	}
 
+	ast_bridge_move_hook(bridge_channel->features, bridge_stasis_moving, NULL, NULL, 0);
+
 	ast_bridge_base_v_table.pull(self, bridge_channel);
 }
 
-static struct ast_bridge_methods bridge_stasis_v_table;
-
 struct ast_bridge *bridge_stasis_new(uint32_t capabilities, unsigned int flags, const char *name, const char *id)
 {
 	void *bridge;