ARI: allow other operations to happen while bridged

This patch changes ARI bridging to allow other channel operations to
happen while the channel is bridged.

ARI channel operations are designed to queue up and execute
sequentially. This meant, though, that while a channel was bridged,
any other channel operations would queue up and execute only after the
channel left the bridge.

This patch changes ARI bridging so that channel commands can execute
while the channel is bridged. For most operations, things simply work
as expected. The one thing that ended up being a bit odd is recording.

The current recording implementation will fail when one attempts to
record a channel that's in a bridge. Note that the bridge itself may
be recording; it's recording a specific channel in the bridge that
fails. While this is an annoying limitation, channel recording is
still very useful for use cases such as voice mail, and bridge
recording makes up much of the difference for other use cases.

(closes issue ASTERISK-22084)
Review: https://reviewboard.asterisk.org/r/2726/


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

12 files changed, 423 insertions, 139 deletions

diff --git a/include/asterisk/bridge_features.h b/include/asterisk/bridge_features.h
index 7590d8af8..ba3cee4e9 100644
--- a/include/asterisk/bridge_features.h
+++ b/include/asterisk/bridge_features.h
@@ -25,6 +25,8 @@
 #ifndef _ASTERISK_BRIDGING_FEATURES_H
 #define _ASTERISK_BRIDGING_FEATURES_H
 
+#include "asterisk/channel.h"
+
 #if defined(__cplusplus) || defined(c_plusplus)
 extern "C" {
 #endif
diff --git a/include/asterisk/bridge_internal.h b/include/asterisk/bridge_internal.h
index 18ef56e96..4ff2fd2aa 100644
--- a/include/asterisk/bridge_internal.h
+++ b/include/asterisk/bridge_internal.h
@@ -36,6 +36,7 @@
 
 struct ast_bridge;
 struct ast_bridge_channel;
+struct ast_bridge_methods;
 
 /*!
  * \brief Register the new bridge with the system.
diff --git a/include/asterisk/stasis_app.h b/include/asterisk/stasis_app.h
index 87b4bdb41..d0ea01080 100644
--- a/include/asterisk/stasis_app.h
+++ b/include/asterisk/stasis_app.h
@@ -348,11 +348,8 @@ void stasis_app_control_add_channel_to_bridge(
  *
  * \param control Control whose channel should be removed from the bridge
  * \param bridge Pointer to the bridge
- *
- * \retval non-zero on failure
- * \retval zero on success
  */
-int stasis_app_control_remove_channel_from_bridge(
+void stasis_app_control_remove_channel_from_bridge(
 	struct stasis_app_control *control, struct ast_bridge *bridge);
 
 /*!
diff --git a/include/asterisk/stasis_app_impl.h b/include/asterisk/stasis_app_impl.h
index d4b467756..e9b93a84a 100644
--- a/include/asterisk/stasis_app_impl.h
+++ b/include/asterisk/stasis_app_impl.h
@@ -85,4 +85,14 @@ void *stasis_app_send_command(struct stasis_app_control *control,
 int stasis_app_send_command_async(struct stasis_app_control *control,
 	stasis_app_command_cb command, void *data);
 
+/*!
+ * \since 12
+ * \brief Gets the bridge currently associated with a control object.
+ *
+ * \param control Control object for the channel to query.
+ * \return Associated \ref ast_bridge.
+ * \return \c NULL if not associated with a bridge.
+ */
+struct ast_bridge *stasis_app_get_bridge(struct stasis_app_control *control);
+
 #endif /* _ASTERISK_RES_STASIS_H */
diff --git a/res/ari/resource_bridges.c b/res/ari/resource_bridges.c
index cc92017fc..348cf972b 100644
--- a/res/ari/resource_bridges.c
+++ b/res/ari/resource_bridges.c
@@ -214,10 +214,8 @@ void ast_ari_remove_channel_from_bridge(struct ast_variable *headers, struct ast
 	 * is added to the channel snapshot. A 409 response should be issued if the bridge
 	 * uniqueids don't match */
 	for (i = 0; i < list->count; ++i) {
-		if (stasis_app_control_remove_channel_from_bridge(list->controls[i], bridge)) {
-			ast_ari_response_error(response, 500, "Internal Error",
-				"Could not remove channel from bridge");
-		}
+		stasis_app_control_remove_channel_from_bridge(list->controls[i],
+			bridge);
 	}
 
 	if (response->response_code) {
diff --git a/res/res_stasis.c b/res/res_stasis.c
index e4ad97eae..b64f40ca9 100644
--- a/res/res_stasis.c
+++ b/res/res_stasis.c
@@ -594,6 +594,13 @@ int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
 		int r;
 		int command_count;
 
+		if (stasis_app_get_bridge(control)) {
+			/* Bridge is handling channel frames */
+			control_wait(control);
+			control_dispatch_all(control, chan);
+			continue;
+		}
+
 		r = ast_waitfor(chan, MAX_WAIT_MS);
 
 		if (r < 0) {
diff --git a/res/res_stasis_bridge_add.c b/res/res_stasis_bridge_add.c
deleted file mode 100644
index ce1315560..000000000
--- a/res/res_stasis_bridge_add.c
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Asterisk -- An open source telephony toolkit.
- *
- * Copyright (C) 2013, Digium, Inc.
- *
- * Kinsey Moore <kmoore@digium.com>
- *
- * See http://www.asterisk.org for more information about
- * the Asterisk project. Please do not directly contact
- * any of the maintainers of this project for assistance;
- * the project provides a web site, mailing lists and IRC
- * channels for your use.
- *
- * This program is free software, distributed under the terms of
- * the GNU General Public License Version 2. See the LICENSE file
- * at the top of the source tree.
- */
-
-/*! \file
- *
- * \brief res_stasis bridge add channel support.
- *
- * \author Kinsey Moore <kmoore@digium.com>
- */
-
-/*** MODULEINFO
-	<depend type="module">res_stasis</depend>
-	<support_level>core</support_level>
- ***/
-
-#include "asterisk.h"
-
-ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
-
-#include "asterisk/module.h"
-#include "asterisk/stasis_app_impl.h"
-#include "asterisk/bridge.h"
-
-static void *app_control_join_bridge(struct stasis_app_control *control,
-	struct ast_channel *chan, void *data)
-{
-	struct ast_bridge_features features;
-	struct ast_bridge *bridge = data;
-	ast_bridge_features_init(&features);
-	ast_bridge_join(bridge,	chan, NULL, &features, NULL, 0);
-	ast_bridge_features_cleanup(&features);
-
-	return NULL;
-}
-
-void stasis_app_control_add_channel_to_bridge(struct stasis_app_control *control, struct ast_bridge *bridge)
-{
-	ast_debug(3, "%s: Sending channel add_to_bridge command\n",
-			stasis_app_control_get_channel_id(control));
-
-	stasis_app_send_command_async(control, app_control_join_bridge, bridge);
-}
-
-static int load_module(void)
-{
-	return AST_MODULE_LOAD_SUCCESS;
-}
-
-static int unload_module(void)
-{
-	return 0;
-}
-
-AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application bridge add channel support",
-	.load = load_module,
-	.unload = unload_module,
-	.nonoptreq = "res_stasis");
diff --git a/res/res_stasis_bridge_add.exports.in b/res/res_stasis_bridge_add.exports.in
deleted file mode 100644
index 0ad493c49..000000000
--- a/res/res_stasis_bridge_add.exports.in
+++ /dev/null
@@ -1,6 +0,0 @@
-{
-	global:
-		LINKER_SYMBOL_PREFIXstasis_app_*;
-	local:
-		*;
-};
diff --git a/res/res_stasis_playback.c b/res/res_stasis_playback.c
index 483aff8c2..83730491c 100644
--- a/res/res_stasis_playback.c
+++ b/res/res_stasis_playback.c
@@ -34,6 +34,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 
 #include "asterisk/app.h"
 #include "asterisk/astobj2.h"
+#include "asterisk/bridge.h"
+#include "asterisk/bridge_internal.h"
 #include "asterisk/file.h"
 #include "asterisk/logger.h"
 #include "asterisk/module.h"
@@ -73,12 +75,56 @@ struct stasis_app_playback {
 	/*! Number of milliseconds to skip for forward/reverse operations */
 	int skipms;
 
+	/*! Set when playback has been completed */
+	int done;
+	/*! Condition for waiting on done to be set */
+	ast_cond_t done_cond;
 	/*! Number of milliseconds of media that has been played */
 	long playedms;
 	/*! Current playback state */
 	enum stasis_app_playback_state state;
 };
 
+static void playback_dtor(void *obj)
+{
+	struct stasis_app_playback *playback = obj;
+
+	ast_string_field_free_memory(playback);
+	ast_cond_destroy(&playback->done_cond);
+}
+
+static struct stasis_app_playback *playback_create(
+	struct stasis_app_control *control)
+{
+	RAII_VAR(struct stasis_app_playback *, playback, NULL, ao2_cleanup);
+	char id[AST_UUID_STR_LEN];
+	int res;
+
+	if (!control) {
+		return NULL;
+	}
+
+	playback = ao2_alloc(sizeof(*playback), playback_dtor);
+	if (!playback || ast_string_field_init(playback, 128)) {
+		return NULL;
+	}
+
+	res = ast_cond_init(&playback->done_cond, NULL);
+	if (res != 0) {
+		ast_log(LOG_ERROR, "Error creating done condition: %s\n",
+			strerror(errno));
+		return NULL;
+	}
+
+	ast_uuid_generate_str(id, sizeof(id));
+	ast_string_field_set(playback, id, id);
+
+	playback->control = control;
+
+	ao2_ref(playback, +1);
+	return playback;
+}
+
 static int playback_hash(const void *obj, int flags)
 {
 	const struct stasis_app_playback *playback = obj;
@@ -144,12 +190,6 @@ static void playback_publish(struct stasis_app_playback *playback)
 	stasis_app_control_publish(playback->control, message);
 }
 
-static void playback_cleanup(struct stasis_app_playback *playback)
-{
-	ao2_unlink_flags(playbacks, playback,
-		OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA);
-}
-
 static int playback_first_update(struct stasis_app_playback *playback,
 	const char *uniqueid)
 {
@@ -191,11 +231,21 @@ static void playback_final_update(struct stasis_app_playback *playback,
 	playback_publish(playback);
 }
 
-static void *play_uri(struct stasis_app_control *control,
-	struct ast_channel *chan, void *data)
+/*!
+ * \brief RAII_VAR function to mark a playback as done when leaving scope.
+ */
+static void mark_as_done(struct stasis_app_playback *playback)
 {
-	RAII_VAR(struct stasis_app_playback *, playback, NULL,
-		playback_cleanup);
+	SCOPED_AO2LOCK(lock, playback);
+	playback->done = 1;
+	ast_cond_broadcast(&playback->done_cond);
+}
+
+static void play_on_channel(struct stasis_app_playback *playback,
+	struct ast_channel *chan)
+{
+	RAII_VAR(struct stasis_app_playback *, mark_when_done, playback,
+		mark_as_done);
 	RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
 	RAII_VAR(char *, file, NULL, ast_free);
 	int res;
@@ -210,7 +260,6 @@ static void *play_uri(struct stasis_app_control *control,
 	const char *pause = NULL;
 	const char *restart = NULL;
 
-	playback = data;
 	ast_assert(playback != NULL);
 
 	offsetms = playback->offsetms;
@@ -218,7 +267,7 @@ static void *play_uri(struct stasis_app_control *control,
 	res = playback_first_update(playback, ast_channel_uniqueid(chan));
 
 	if (res != 0) {
-		return NULL;
+		return;
 	}
 
 	if (ast_channel_state(chan) != AST_STATE_UP) {
@@ -241,11 +290,11 @@ static void *play_uri(struct stasis_app_control *control,
 	} else {
 		/* Play URL */
 		ast_log(LOG_ERROR, "Unimplemented\n");
-		return NULL;
+		return;
 	}
 
 	if (!file) {
-		return NULL;
+		return;
 	}
 
 	res = ast_control_streamfile_lang(chan, file, fwd, rev, stop, pause,
@@ -254,14 +303,87 @@ static void *play_uri(struct stasis_app_control *control,
 	playback_final_update(playback, offsetms, res,
 		ast_channel_uniqueid(chan));
 
-	return NULL;
+	return;
 }
 
-static void playback_dtor(void *obj)
+/*!
+ * \brief Special case code to play while a channel is in a bridge.
+ *
+ * \param bridge_channel The channel's bridge_channel.
+ * \param playback_id Id of the playback to start.
+ */
+static void play_on_channel_in_bridge(struct ast_bridge_channel *bridge_channel,
+	const char *playback_id)
 {
-	struct stasis_app_playback *playback = obj;
+	RAII_VAR(struct stasis_app_playback *, playback, NULL, ao2_cleanup);
 
-	ast_string_field_free_memory(playback);
+	playback = stasis_app_playback_find_by_id(playback_id);
+	if (!playback) {
+		ast_log(LOG_ERROR, "Couldn't find playback %s\n",
+			playback_id);
+		return;
+	}
+
+	play_on_channel(playback, bridge_channel->chan);
+}
+
+/*!
+ * \brief \ref RAII_VAR function to remove a playback from the global list when
+ * leaving scope.
+ */
+static void remove_from_playbacks(struct stasis_app_playback *playback)
+{
+	ao2_unlink_flags(playbacks, playback,
+		OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA);
+}
+
+static void *play_uri(struct stasis_app_control *control,
+	struct ast_channel *chan, void *data)
+{
+	RAII_VAR(struct stasis_app_playback *, playback, NULL,
+		remove_from_playbacks);
+	struct ast_bridge *bridge;
+	int res;
+
+	playback = data;
+
+	if (!control) {
+		return NULL;
+	}
+
+	bridge = stasis_app_get_bridge(control);
+	if (bridge) {
+		struct ast_bridge_channel *bridge_chan;
+
+		/* Queue up playback on the bridge */
+		ast_bridge_lock(bridge);
+		bridge_chan = bridge_find_channel(bridge, chan);
+		if (bridge_chan) {
+			ast_bridge_channel_queue_playfile(
+				bridge_chan,
+				play_on_channel_in_bridge,
+				playback->id,
+				NULL); /* moh_class */
+		}
+		ast_bridge_unlock(bridge);
+
+		/* Wait for playback to complete */
+		ao2_lock(playback);
+		while (!playback->done) {
+			res = ast_cond_wait(&playback->done_cond,
+				ao2_object_get_lockaddr(playback));
+			if (res != 0) {
+				ast_log(LOG_ERROR,
+					"Error waiting for playback to complete: %s\n",
+					strerror(errno));
+			}
+		}
+		ao2_unlock(playback);
+	} else {
+		play_on_channel(playback, chan);
+	}
+
+	return NULL;
 }
 
 static void set_target_uri(
@@ -291,7 +413,6 @@ struct stasis_app_playback *stasis_app_control_play_uri(
 	int skipms, long offsetms)
 {
 	RAII_VAR(struct stasis_app_playback *, playback, NULL, ao2_cleanup);
-	char id[AST_UUID_STR_LEN];
 
 	if (skipms < 0 || offsetms < 0) {
 		return NULL;
@@ -300,21 +421,15 @@ struct stasis_app_playback *stasis_app_control_play_uri(
 	ast_debug(3, "%s: Sending play(%s) command\n",
 		stasis_app_control_get_channel_id(control), uri);
 
-	playback = ao2_alloc(sizeof(*playback), playback_dtor);
-	if (!playback || ast_string_field_init(playback, 128)) {
-		return NULL;
-	}
+	playback = playback_create(control);
 
 	if (skipms == 0) {
 		skipms = PLAYBACK_DEFAULT_SKIPMS;
 	}
 
-	ast_uuid_generate_str(id, sizeof(id));
-	ast_string_field_set(playback, id, id);
 	ast_string_field_set(playback, media, uri);
 	ast_string_field_set(playback, language, language);
 	set_target_uri(playback, target_type, target_id);
-	playback->control = control;
 	playback->skipms = skipms;
 	playback->offsetms = offsetms;
 	ao2_link(playbacks, playback);
@@ -346,15 +461,7 @@ const char *stasis_app_playback_get_id(
 
 struct stasis_app_playback *stasis_app_playback_find_by_id(const char *id)
 {
-	RAII_VAR(struct stasis_app_playback *, playback, NULL, ao2_cleanup);
-
-	playback = ao2_find(playbacks, id, OBJ_KEY);
-	if (playback == NULL) {
-		return NULL;
-	}
-
-	ao2_ref(playback, +1);
-	return playback;
+	return ao2_find(playbacks, id, OBJ_KEY);
 }
 
 struct ast_json *stasis_app_playback_to_json(
diff --git a/res/res_stasis_recording.c b/res/res_stasis_recording.c
index 575ccae1e..f62716826 100644
--- a/res/res_stasis_recording.c
+++ b/res/res_stasis_recording.c
@@ -231,10 +231,11 @@ static void *record_file(struct stasis_app_control *control,
 	recording = data;
 	ast_assert(recording != NULL);
 
-	ao2_lock(recording);
-	recording->state = STASIS_APP_RECORDING_STATE_RECORDING;
-	recording_publish(recording);
-	ao2_unlock(recording);
+	if (stasis_app_get_bridge(control)) {
+		ast_log(LOG_ERROR, "Cannot record channel while in bridge\n");
+		recording_fail(recording);
+		return NULL;
+	}
 
 	switch (recording->options->terminate_on) {
 	case STASIS_APP_RECORDING_TERMINATE_NONE:
@@ -258,6 +259,11 @@ static void *record_file(struct stasis_app_control *control,
 		return NULL;
 	}
 
+	ao2_lock(recording);
+	recording->state = STASIS_APP_RECORDING_STATE_RECORDING;
+	recording_publish(recording);
+	ao2_unlock(recording);
+
 	ast_play_and_record_full(chan,
 		NULL, /* playfile */
 		recording->absolute_name,
diff --git a/res/stasis/control.c b/res/stasis/control.c
index dcc029701..df279165f 100644
--- a/res/stasis/control.c
+++ b/res/stasis/control.c
@@ -33,62 +33,99 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "control.h"
 #include "asterisk/dial.h"
 #include "asterisk/bridge.h"
+#include "asterisk/bridge_after.h"
 #include "asterisk/bridge_basic.h"
 #include "asterisk/frame.h"
 #include "asterisk/pbx.h"
 #include "asterisk/musiconhold.h"
 
 struct stasis_app_control {
+	ast_cond_t wait_cond;
 	/*! Queue of commands to dispatch on the channel */
 	struct ao2_container *command_queue;
 	/*!
-	 * When set, /c app_stasis should exit and continue in the dialplan.
-	 */
-	int is_done:1;
-	/*!
 	 * The associated channel.
 	 * Be very careful with the threading associated w/ manipulating
 	 * the channel.
 	 */
 	struct ast_channel *channel;
+	/*!
+	 * When a channel is in a bridge, the bridge that it is in.
+	 */
+	struct ast_bridge *bridge;
+	/*!
+	 * Holding place for channel's PBX while imparted to a bridge.
+	 */
+	struct ast_pbx *pbx;
+	/*!
+	 * When set, /c app_stasis should exit and continue in the dialplan.
+	 */
+	int is_done:1;
 };
 
+static void control_dtor(void *obj)
+{
+	struct stasis_app_control *control = obj;
+
+	ao2_cleanup(control->command_queue);
+	ast_cond_destroy(&control->wait_cond);
+}
+
 struct stasis_app_control *control_create(struct ast_channel *channel)
 {
-	struct stasis_app_control *control;
+	RAII_VAR(struct stasis_app_control *, control, NULL, ao2_cleanup);
+	int res;
 
-	control = ao2_alloc(sizeof(*control), NULL);
+	control = ao2_alloc(sizeof(*control), control_dtor);
 	if (!control) {
 		return NULL;
 	}
 
+	res = ast_cond_init(&control->wait_cond, NULL);
+	if (res != 0) {
+		ast_log(LOG_ERROR, "Error initializing ast_cond_t: %s\n",
+			strerror(errno));
+		return NULL;
+	}
+
 	control->command_queue = ao2_container_alloc_list(
 		AO2_ALLOC_OPT_LOCK_MUTEX, 0, NULL, NULL);
 
 	if (!control->command_queue) {
-		ao2_cleanup(control);
 		return NULL;
 	}
 
 	control->channel = channel;
 
+	ao2_ref(control, +1);
 	return control;
 }
 
+static void *noop_cb(struct stasis_app_control *control,
+	struct ast_channel *chan, void *data)
+{
+	return NULL;
+}
+
+
 static struct stasis_app_command *exec_command(
 	struct stasis_app_control *control, stasis_app_command_cb command_fn,
 	void *data)
 {
 	RAII_VAR(struct stasis_app_command *, command, NULL, ao2_cleanup);
 
+	command_fn = command_fn ? : noop_cb;
+
 	command = command_create(command_fn, data);
 
 	if (!command) {
 		return NULL;
 	}
 
-	/* command_queue is a thread safe list; no lock needed */
-	ao2_link(control->command_queue, command);
+	ao2_lock(control->command_queue);
+	ao2_link_flags(control->command_queue, command, OBJ_NOLOCK);
+	ast_cond_signal(&control->wait_cond);
+	ao2_unlock(control->command_queue);
 
 	ao2_ref(command, +1);
 	return command;
@@ -195,6 +232,14 @@ static void *app_control_continue(struct stasis_app_control *control,
 {
 	RAII_VAR(struct stasis_app_control_continue_data *, continue_data, data, ast_free);
 
+	ast_assert(control->channel != NULL);
+
+	/* If we're in a Stasis bridge, depart it before going back to the
+	 * dialplan */
+	if (stasis_app_get_bridge(control)) {
+		ast_bridge_depart(control->channel);
+	}
+
 	/* Called from stasis_app_exec thread; no lock needed */
 	ast_explicit_goto(control->channel, continue_data->context, continue_data->extension, continue_data->priority);
 
@@ -422,6 +467,161 @@ int stasis_app_send_command_async(struct stasis_app_control *control,
 	return 0;
 }
 
+struct ast_bridge *stasis_app_get_bridge(struct stasis_app_control *control)
+{
+	if (!control) {
+		return NULL;
+	} else {
+		SCOPED_AO2LOCK(lock, control);
+		return control->bridge;
+	}
+}
+
+
+static void bridge_after_cb(struct ast_channel *chan, void *data)
+{
+	struct stasis_app_control *control = data;
+	SCOPED_AO2LOCK(lock, control);
+
+	ast_debug(3, "%s, %s: Channel leaving bridge\n",
+		ast_channel_uniqueid(chan), control->bridge->uniqueid);
+
+	ast_assert(chan == control->channel);
+
+	/* Restore the channel's PBX */
+	ast_channel_pbx_set(control->channel, control->pbx);
+	control->pbx = NULL;
+
+	/* No longer in the bridge */
+	control->bridge = NULL;
+
+	/* Wakeup the command_queue loop */
+	exec_command(control, NULL, NULL);
+}
+
+static void bridge_after_cb_failed(enum ast_bridge_after_cb_reason reason,
+	void *data)
+{
+	struct stasis_app_control *control = data;
+
+	bridge_after_cb(control->channel, data);
+
+	ast_debug(3, "  reason: %s\n",
+		ast_bridge_after_cb_reason_string(reason));
+}
+
+static void *app_control_add_channel_to_bridge(
+	struct stasis_app_control *control,
+	struct ast_channel *chan, void *data)
+{
+	struct ast_bridge *bridge = data;
+	int res;
+
+	if (!control || !bridge) {
+		return NULL;
+	}
+
+	ast_debug(3, "%s: Adding to bridge %s\n",
+		stasis_app_control_get_channel_id(control),
+		bridge->uniqueid);
+
+	ast_assert(chan != NULL);
+
+	/* Depart whatever Stasis bridge we're currently in. */
+	if (stasis_app_get_bridge(control)) {
+		/* Note that it looks like there's a race condition here, since
+		 * we don't have control locked. But this happens from the
+		 * control callback thread, so there won't be any other
+		 * concurrent attempts to bridge.
+		 */
+		ast_bridge_depart(chan);
+	}
+
+
+	res = ast_bridge_set_after_callback(chan, bridge_after_cb,
+		bridge_after_cb_failed, control);
+	if (res != 0) {
+		ast_log(LOG_ERROR, "Error setting after-bridge callback\n");
+		return NULL;
+	}
+
+	{
+		/* pbx and bridge are modified by the bridging impart thread.
+		 * It shouldn't happen concurrently, but we still need to lock
+		 * for the memory fence.
+		 */
+		SCOPED_AO2LOCK(lock, control);
+
+		/* Save off the channel's PBX */
+		ast_assert(control->pbx == NULL);
+		if (!control->pbx) {
+			control->pbx = ast_channel_pbx(chan);
+			ast_channel_pbx_set(chan, NULL);
+		}
+
+		res = ast_bridge_impart(bridge,
+			chan,
+			NULL, /* swap channel */
+			NULL, /* features */
+			0); /* independent - false allows us to ast_bridge_depart() */
+
+		if (res != 0) {
+			ast_log(LOG_ERROR, "Error adding channel to bridge\n");
+			ast_channel_pbx_set(chan, control->pbx);
+			control->pbx = NULL;
+			return NULL;
+		}
+
+		ast_assert(stasis_app_get_bridge(control) == NULL);
+		control->bridge = bridge;
+	}
+	return NULL;
+}
+
+void stasis_app_control_add_channel_to_bridge(
+	struct stasis_app_control *control, struct ast_bridge *bridge)
+{
+	ast_debug(3, "%s: Sending channel add_to_bridge command\n",
+			stasis_app_control_get_channel_id(control));
+	stasis_app_send_command_async(control,
+		app_control_add_channel_to_bridge, bridge);
+}
+
+static void *app_control_remove_channel_from_bridge(
+	struct stasis_app_control *control,
+	struct ast_channel *chan, void *data)
+{
+	struct ast_bridge *bridge = data;
+
+	if (!control) {
+		return NULL;
+	}
+
+	/* We should only depart from our own bridge */
+	ast_debug(3, "%s: Departing bridge %s\n",
+		stasis_app_control_get_channel_id(control),
+		bridge->uniqueid);
+
+	if (bridge != stasis_app_get_bridge(control)) {
+		ast_log(LOG_WARNING, "%s: Not in bridge %s; not removing\n",
+			stasis_app_control_get_channel_id(control),
+			bridge->uniqueid);
+		return NULL;
+	}
+
+	ast_bridge_depart(chan);
+	return NULL;
+}
+
+void stasis_app_control_remove_channel_from_bridge(
+	struct stasis_app_control *control, struct ast_bridge *bridge)
+{
+	ast_debug(3, "%s: Sending channel remove_from_bridge command\n",
+			stasis_app_control_get_channel_id(control));
+	stasis_app_send_command_async(control,
+		app_control_remove_channel_from_bridge, bridge);
+}
+
 const char *stasis_app_control_get_channel_id(
 	const struct stasis_app_control *control)
 {
@@ -464,9 +664,22 @@ int control_dispatch_all(struct stasis_app_control *control,
 	return count;
 }
 
-/* Must be defined here since it must operate on the channel outside of the queue */
-int stasis_app_control_remove_channel_from_bridge(
-	struct stasis_app_control *control, struct ast_bridge *bridge)
+void control_wait(struct stasis_app_control *control)
 {
-	return ast_bridge_remove(bridge, control->channel);
+	if (!control) {
+		return;
+	}
+
+	ast_assert(control->command_queue != NULL);
+
+	ao2_lock(control->command_queue);
+	while (ao2_container_count(control->command_queue) == 0) {
+		int res = ast_cond_wait(&control->wait_cond,
+			ao2_object_get_lockaddr(control->command_queue));
+		if (res < 0) {
+			ast_log(LOG_ERROR, "Error waiting on command queue\n");
+			break;
+		}
+	}
+	ao2_unlock(control->command_queue);
 }
diff --git a/res/stasis/control.h b/res/stasis/control.h
index 9a4243be1..9e580b860 100644
--- a/res/stasis/control.h
+++ b/res/stasis/control.h
@@ -48,8 +48,29 @@ struct stasis_app_control *control_create(struct ast_channel *channel);
 int control_dispatch_all(struct stasis_app_control *control,
 	struct ast_channel *chan);
 
-int control_is_done(struct stasis_app_control *control);
+/*!
+ * \brief Blocks until \a control's command queue has a command available.
+ *
+ * \param control Control to block on.
+ */
+void control_wait(struct stasis_app_control *control);
 
+/*!
+ * \brief Signals that a control object should finish and exit back to the
+ * dialplan.
+ *
+ * \param control Control object to continue.
+ */
 void control_continue(struct stasis_app_control *control);
 
+/*!
+ * \brief Returns true if control_continue() has been called on this \a control.
+ *
+ * \param control Control to query.
+ * \return True (non-zero) if control_continue() has been called.
+ * \return False (zero) otherwise.
+ */
+int control_is_done(struct stasis_app_control *control);
+
+
 #endif /* _ASTERISK_RES_STASIS_CONTROL_H */