diff options
Diffstat (limited to 'res')
| -rw-r--r-- | res/ari/resource_bridges.c | 6 | ||||
| -rw-r--r-- | res/res_stasis.c | 7 | ||||
| -rw-r--r-- | res/res_stasis_bridge_add.c | 72 | ||||
| -rw-r--r-- | res/res_stasis_bridge_add.exports.in | 6 | ||||
| -rw-r--r-- | res/res_stasis_playback.c | 177 | ||||
| -rw-r--r-- | res/res_stasis_recording.c | 14 | ||||
| -rw-r--r-- | res/stasis/control.c | 239 | ||||
| -rw-r--r-- | res/stasis/control.h | 23 |
8 files changed, 409 insertions, 135 deletions
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 */ |