diff options
| -rw-r--r-- | include/asterisk/stasis_app.h | 50 | ||||
| -rw-r--r-- | include/asterisk/stasis_bridging.h | 12 | ||||
| -rw-r--r-- | main/stasis_bridging.c | 22 | ||||
| -rw-r--r-- | res/res_stasis.c | 302 | ||||
| -rw-r--r-- | res/res_stasis_bridge_add.c | 73 | ||||
| -rw-r--r-- | res/res_stasis_bridge_add.exports.in | 6 | ||||
| -rw-r--r-- | res/res_stasis_json_events.c | 72 | ||||
| -rw-r--r-- | res/res_stasis_json_events.exports.in | 3 | ||||
| -rw-r--r-- | res/stasis/app.c | 38 | ||||
| -rw-r--r-- | res/stasis/app.h | 28 | ||||
| -rw-r--r-- | res/stasis/control.c | 9 | ||||
| -rw-r--r-- | res/stasis_http/resource_bridges.c | 191 | ||||
| -rw-r--r-- | res/stasis_http/resource_bridges.h | 2 | ||||
| -rw-r--r-- | res/stasis_json/resource_events.h | 42 | ||||
| -rw-r--r-- | rest-api/api-docs/bridges.json | 2 | ||||
| -rw-r--r-- | rest-api/api-docs/events.json | 15 |
16 files changed, 825 insertions, 42 deletions
diff --git a/include/asterisk/stasis_app.h b/include/asterisk/stasis_app.h index 27b3502f3..3eed47e35 100644 --- a/include/asterisk/stasis_app.h +++ b/include/asterisk/stasis_app.h @@ -187,6 +187,56 @@ int stasis_app_control_queue_control(struct stasis_app_control *control, enum ast_control_frame_type frame_type); /*! + * \brief Create a bridge of the specified type. + * + * \param type The type of bridge to be created + * + * \return New bridge. + * \return \c NULL on error. + */ +struct ast_bridge *stasis_app_bridge_create(const char *type); + +/*! + * \brief Returns the bridge with the given id. + * \param bridge_id Uniqueid of the bridge. + * \return NULL bridge not created by a Stasis application, or bridge does not exist. + * \return Pointer to bridge. + */ +struct ast_bridge *stasis_app_bridge_find_by_id( + const char *bridge_id); + +/*! + * \brief Add a channel to the bridge. + * + * \param control Control whose channel should be added to the bridge + * \param bridge Pointer to the bridge + */ +void stasis_app_control_add_channel_to_bridge( + struct stasis_app_control *control, struct ast_bridge *bridge); + +/*! + * \brief Remove a channel from the 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( + struct stasis_app_control *control, struct ast_bridge *bridge); + +/*! + * \brief Destroy the bridge. + * + * \param bridge_id Uniqueid of bridge to be destroyed + * + * \retval non-zero on failure + * \retval zero on success + */ +void stasis_app_bridge_destroy(const char *bridge_id); + +/*! * \brief Increment the res_stasis reference count. * * This ensures graceful shutdown happens in the proper order. diff --git a/include/asterisk/stasis_bridging.h b/include/asterisk/stasis_bridging.h index 5b8ff6ffc..42cb73ba0 100644 --- a/include/asterisk/stasis_bridging.h +++ b/include/asterisk/stasis_bridging.h @@ -207,6 +207,18 @@ void ast_bridge_publish_leave(struct ast_bridge *bridge, struct ast_channel *cha struct ast_json *ast_bridge_snapshot_to_json(const struct ast_bridge_snapshot *snapshot); /*! + * \brief Returns the most recent snapshot for the bridge. + * + * The returned pointer is AO2 managed, so ao2_cleanup() when you're done. + * + * \param bridge_id Uniqueid of the bridge for which to get the snapshot. + * \return Most recent snapshot. ao2_cleanup() when done. + * \return \c NULL if channel isn't in cache. + */ +struct ast_bridge_snapshot *ast_bridge_snapshot_get_latest( + const char *bridge_id); + +/*! * \brief Initialize the stasis bridging topic and message types * \retval 0 on success * \retval -1 on failure diff --git a/main/stasis_bridging.c b/main/stasis_bridging.c index d181c400c..28771a7b8 100644 --- a/main/stasis_bridging.c +++ b/main/stasis_bridging.c @@ -310,6 +310,28 @@ struct ast_json *ast_bridge_snapshot_to_json(const struct ast_bridge_snapshot *s return ast_json_ref(json_chan); } +struct ast_bridge_snapshot *ast_bridge_snapshot_get_latest(const char *uniqueid) +{ + RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup); + struct ast_bridge_snapshot *snapshot; + + ast_assert(!ast_strlen_zero(uniqueid)); + + message = stasis_cache_get(ast_bridge_topic_all_cached(), + ast_bridge_snapshot_type(), + uniqueid); + if (!message) { + return NULL; + } + + snapshot = stasis_message_data(message); + if (!snapshot) { + return NULL; + } + ao2_ref(snapshot, +1); + return snapshot; +} + static void stasis_bridging_cleanup(void) { ao2_cleanup(bridge_topic_all); diff --git a/res/res_stasis.c b/res/res_stasis.c index 23d044971..5227654f3 100644 --- a/res/res_stasis.c +++ b/res/res_stasis.c @@ -61,6 +61,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/module.h" #include "asterisk/stasis_app_impl.h" #include "asterisk/stasis_channels.h" +#include "asterisk/stasis_bridging.h" #include "asterisk/stasis_message_router.h" #include "asterisk/strings.h" #include "stasis/app.h" @@ -89,9 +90,14 @@ struct ao2_container *apps_registry; struct ao2_container *app_controls; +struct ao2_container *app_bridges; + /*! \brief Message router for the channel caching topic */ struct stasis_message_router *channel_router; +/*! \brief Message router for the bridge caching topic */ +struct stasis_message_router *bridge_router; + /*! AO2 hash function for \ref app */ static int app_hash(const void *obj, const int flags) { @@ -159,9 +165,42 @@ struct stasis_app_control *stasis_app_control_find_by_channel_id( return ao2_find(app_controls, channel_id, OBJ_KEY); } +/*! AO2 hash function for bridges container */ +static int bridges_hash(const void *obj, const int flags) +{ + const struct ast_bridge *bridge = obj; + const char *id = flags & OBJ_KEY ? + obj : bridge->uniqueid; + + return ast_str_hash(id); +} + +/*! AO2 comparison function for bridges container */ +static int bridges_compare(void *lhs, void *rhs, int flags) +{ + const struct ast_bridge *lhs_bridge = lhs; + const struct ast_bridge *rhs_bridge = rhs; + const char *lhs_id = lhs_bridge->uniqueid; + const char *rhs_id = flags & OBJ_KEY ? + rhs : rhs_bridge->uniqueid; + + if (strcmp(lhs_id, rhs_id) == 0) { + return CMP_MATCH | CMP_STOP; + } else { + return 0; + } +} + +struct ast_bridge *stasis_app_bridge_find_by_id( + const char *bridge_id) +{ + return ao2_find(app_bridges, bridge_id, OBJ_KEY); +} + /*! \brief Typedef for blob handler callbacks */ typedef struct ast_json *(*channel_blob_handler_cb)(struct ast_channel_blob *); +/*! \brief Callback to check whether an app is watching a given channel */ static int app_watching_channel_cb(void *obj, void *arg, int flags) { struct app *app = obj; @@ -170,7 +209,8 @@ static int app_watching_channel_cb(void *obj, void *arg, int flags) return app_is_watching_channel(app, uniqueid) ? CMP_MATCH : 0; } -static struct ao2_container *get_watching_apps(const char *uniqueid) +/*! \brief Get a container full of apps that are interested in the specified channel */ +static struct ao2_container *get_apps_watching_channel(const char *uniqueid) { struct ao2_container *watching_apps; char *uniqueid_dup; @@ -302,7 +342,7 @@ static int app_send_cb(void *obj, void *arg, int flags) return 0; } -static void sub_snapshot_handler(void *data, +static void sub_channel_snapshot_handler(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message) @@ -313,7 +353,7 @@ static void sub_snapshot_handler(void *data, struct ast_channel_snapshot *old_snapshot = stasis_message_data(update->old_snapshot); int i; - watching_apps = get_watching_apps(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid); + watching_apps = get_apps_watching_channel(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid); if (!watching_apps) { return; } @@ -342,7 +382,7 @@ static void generic_blob_handler(struct ast_channel_blob *obj, channel_blob_hand return; } - watching_apps = get_watching_apps(obj->snapshot->uniqueid); + watching_apps = get_apps_watching_channel(obj->snapshot->uniqueid); if (!watching_apps) { return; } @@ -357,7 +397,7 @@ static void generic_blob_handler(struct ast_channel_blob *obj, channel_blob_hand /*! * \brief In addition to running ao2_cleanup(), this function also removes the - * object from the app_controls() container. + * object from the app_controls container. */ static void control_unlink(struct stasis_app_control *control) { @@ -370,6 +410,38 @@ static void control_unlink(struct stasis_app_control *control) ao2_cleanup(control); } +struct ast_bridge *stasis_app_bridge_create(const char *type) +{ + struct ast_bridge *bridge; + int capabilities, flags = 0; + if (ast_strlen_zero(type) || !strcmp(type, "mixing")) { + capabilities = AST_BRIDGE_CAPABILITY_1TO1MIX | + AST_BRIDGE_CAPABILITY_MULTIMIX | + AST_BRIDGE_CAPABILITY_NATIVE; + flags = AST_BRIDGE_FLAG_SMART; + } else if (!strcmp(type, "holding")) { + capabilities = AST_BRIDGE_CAPABILITY_HOLDING; + } else { + return NULL; + } + + bridge = ast_bridge_base_new(capabilities, flags); + if (bridge) { + ao2_link(app_bridges, bridge); + } + return bridge; +} + +void stasis_app_bridge_destroy(const char *bridge_id) +{ + struct ast_bridge *bridge = stasis_app_bridge_find_by_id(bridge_id); + if (!bridge) { + return; + } + ao2_unlink(app_bridges, bridge); + ast_bridge_destroy(bridge); +} + int app_send_start_msg(struct app *app, struct ast_channel *chan, int argc, char *argv[]) { @@ -679,6 +751,199 @@ void stasis_app_unref(void) ast_module_unref(ast_module_info->self); } +/*! \brief Callback to check whether an app is watching a given bridge */ +static int app_watching_bridge_cb(void *obj, void *arg, int flags) +{ + struct app *app = obj; + char *uniqueid = arg; + + return app_is_watching_bridge(app, uniqueid) ? CMP_MATCH : 0; +} + +/*! \brief Get a container full of apps that are interested in the specified bridge */ +static struct ao2_container *get_apps_watching_bridge(const char *uniqueid) +{ + struct ao2_container *watching_apps; + char *uniqueid_dup; + RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy); + ast_assert(uniqueid != NULL); + + uniqueid_dup = ast_strdupa(uniqueid); + + watching_apps_iter = ao2_callback(apps_registry, OBJ_MULTIPLE, app_watching_bridge_cb, uniqueid_dup); + watching_apps = watching_apps_iter->c; + + if (!ao2_container_count(watching_apps)) { + return NULL; + } + + ao2_ref(watching_apps, +1); + return watching_apps_iter->c; +} + +/*! Callback used to remove an app's interest in a bridge */ +static int remove_bridge_cb(void *obj, void *arg, int flags) +{ + app_remove_bridge(obj, arg); + return 0; +} + +static void sub_bridge_snapshot_handler(void *data, + struct stasis_subscription *sub, + struct stasis_topic *topic, + struct stasis_message *message) +{ + RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup); + struct stasis_cache_update *update = stasis_message_data(message); + struct ast_bridge_snapshot *new_snapshot = stasis_message_data(update->new_snapshot); + struct ast_bridge_snapshot *old_snapshot = stasis_message_data(update->old_snapshot); + RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref); + + watching_apps = get_apps_watching_bridge(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid); + if (!watching_apps || !ao2_container_count(watching_apps)) { + return; + } + + if (!new_snapshot) { + RAII_VAR(char *, bridge_id, ast_strdup(old_snapshot->uniqueid), ast_free); + + /* The bridge has gone away. Create the message, make sure no apps are + * watching this bridge anymore, and destroy the bridge's control + * structure */ + msg = stasis_json_event_bridge_destroyed_create(old_snapshot); + ao2_callback(watching_apps, OBJ_NODATA, remove_bridge_cb, bridge_id); + stasis_app_bridge_destroy(old_snapshot->uniqueid); + } else if (!old_snapshot) { + msg = stasis_json_event_bridge_created_create(old_snapshot); + } + + if (!msg) { + return; + } + + distribute_message(watching_apps, msg); +} + +/*! \brief Callback used to merge two containers of applications */ +static int list_merge_cb(void *obj, void *arg, int flags) +{ + /* remove any current entries for this app */ + ao2_find(arg, obj, OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE); + /* relink as the only entry */ + ao2_link(arg, obj); + return 0; +} + +/*! \brief Merge container src into container dst without modifying src */ +static void update_apps_list(struct ao2_container *dst, struct ao2_container *src) +{ + ao2_callback(src, OBJ_NODATA, list_merge_cb, dst); +} + +/*! \brief Callback for adding to an app's bridges of interest */ +static int app_add_bridge_cb(void *obj, void *arg, int flags) +{ + app_add_bridge(obj, arg); + return 0; +} + +/*! \brief Add interest in the given bridge to all apps in the container */ +static void update_bridge_interest(struct ao2_container *apps, const char *bridge_id) +{ + RAII_VAR(char *, bridge_id_dup, ast_strdup(bridge_id), ast_free); + ao2_callback(apps, OBJ_NODATA, app_add_bridge_cb, bridge_id_dup); +} + +static void sub_bridge_merge_handler(void *data, + struct stasis_subscription *sub, + struct stasis_topic *topic, + struct stasis_message *message) +{ + RAII_VAR(struct ao2_container *, watching_apps_to, NULL, ao2_cleanup); + RAII_VAR(struct ao2_container *, watching_apps_from, NULL, ao2_cleanup); + RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup); + struct ast_bridge_merge_message *merge = stasis_message_data(message); + RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref); + RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref); + + watching_apps_to = get_apps_watching_bridge(merge->to->uniqueid); + if (watching_apps_to) { + update_apps_list(watching_apps_all, watching_apps_to); + } + + watching_apps_from = get_apps_watching_bridge(merge->from->uniqueid); + if (watching_apps_from) { + update_bridge_interest(watching_apps_from, merge->to->uniqueid); + update_apps_list(watching_apps_all, watching_apps_from); + } + + if (!ao2_container_count(watching_apps_all)) { + return; + } + + /* The secondary bridge has to be packed into JSON by hand because the auto-generated + * JSON event generator can only handle one instance of a given snapshot type in an + * elegant way */ + blob = ast_json_pack("{s: o}", "bridge_from", ast_bridge_snapshot_to_json(merge->from)); + if (!blob) { + return; + } + + msg = stasis_json_event_bridge_merged_create(merge->to, blob); + + distribute_message(watching_apps_all, msg); +} + +static void sub_bridge_enter_handler(void *data, + struct stasis_subscription *sub, + struct stasis_topic *topic, + struct stasis_message *message) +{ + RAII_VAR(struct ao2_container *, watching_apps_channel, NULL, ao2_cleanup); + RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup); + RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup); + struct ast_bridge_blob *obj = stasis_message_data(message); + RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref); + + watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid); + if (watching_apps_bridge) { + update_apps_list(watching_apps_all, watching_apps_bridge); + } + + watching_apps_channel = get_apps_watching_channel(obj->channel->uniqueid); + if (watching_apps_channel) { + update_bridge_interest(watching_apps_channel, obj->bridge->uniqueid); + update_apps_list(watching_apps_all, watching_apps_channel); + } + + if (!ao2_container_count(watching_apps_all)) { + return; + } + + msg = stasis_json_event_channel_entered_bridge_create(obj->bridge, obj->channel); + + distribute_message(watching_apps_all, msg); +} + +static void sub_bridge_leave_handler(void *data, + struct stasis_subscription *sub, + struct stasis_topic *topic, + struct stasis_message *message) +{ + RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup); + struct ast_bridge_blob *obj = stasis_message_data(message); + RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref); + + watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid); + if (!watching_apps_bridge) { + return; + } + + msg = stasis_json_event_channel_left_bridge_create(obj->bridge, obj->channel); + + distribute_message(watching_apps_bridge, msg); +} + static int load_module(void) { int r = 0; @@ -695,12 +960,18 @@ static int load_module(void) return AST_MODULE_LOAD_FAILURE; } + app_bridges = ao2_container_alloc(CONTROLS_NUM_BUCKETS, + bridges_hash, bridges_compare); + if (app_bridges == NULL) { + return AST_MODULE_LOAD_FAILURE; + } + channel_router = stasis_message_router_create(stasis_caching_get_topic(ast_channel_topic_all_cached())); if (!channel_router) { return AST_MODULE_LOAD_FAILURE; } - r |= stasis_message_router_add(channel_router, stasis_cache_update_type(), sub_snapshot_handler, NULL); + r |= stasis_message_router_add(channel_router, stasis_cache_update_type(), sub_channel_snapshot_handler, NULL); r |= stasis_message_router_add(channel_router, ast_channel_user_event_type(), sub_userevent_handler, NULL); r |= stasis_message_router_add(channel_router, ast_channel_varset_type(), sub_varset_handler, NULL); r |= stasis_message_router_add(channel_router, ast_channel_dtmf_begin_type(), sub_dtmf_handler, NULL); @@ -709,6 +980,19 @@ static int load_module(void) return AST_MODULE_LOAD_FAILURE; } + bridge_router = stasis_message_router_create(stasis_caching_get_topic(ast_bridge_topic_all_cached())); + if (!bridge_router) { + return AST_MODULE_LOAD_FAILURE; + } + + r |= stasis_message_router_add(bridge_router, stasis_cache_update_type(), sub_bridge_snapshot_handler, NULL); + r |= stasis_message_router_add(bridge_router, ast_bridge_merge_message_type(), sub_bridge_merge_handler, NULL); + r |= stasis_message_router_add(bridge_router, ast_channel_entered_bridge_type(), sub_bridge_enter_handler, NULL); + r |= stasis_message_router_add(bridge_router, ast_channel_left_bridge_type(), sub_bridge_leave_handler, NULL); + if (r) { + return AST_MODULE_LOAD_FAILURE; + } + return AST_MODULE_LOAD_SUCCESS; } @@ -719,12 +1003,18 @@ static int unload_module(void) stasis_message_router_unsubscribe_and_join(channel_router); channel_router = NULL; + stasis_message_router_unsubscribe_and_join(bridge_router); + bridge_router = NULL; + ao2_cleanup(apps_registry); apps_registry = NULL; ao2_cleanup(app_controls); app_controls = NULL; + ao2_cleanup(app_bridges); + app_bridges = NULL; + return r; } diff --git a/res/res_stasis_bridge_add.c b/res/res_stasis_bridge_add.c new file mode 100644 index 000000000..73340d1bb --- /dev/null +++ b/res/res_stasis_bridge_add.c @@ -0,0 +1,73 @@ +/* + * 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/bridging.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 new file mode 100644 index 000000000..0ad493c49 --- /dev/null +++ b/res/res_stasis_bridge_add.exports.in @@ -0,0 +1,6 @@ +{ + global: + LINKER_SYMBOL_PREFIXstasis_app_*; + local: + *; +}; diff --git a/res/res_stasis_json_events.c b/res/res_stasis_json_events.c index e96d84e4f..3116545be 100644 --- a/res/res_stasis_json_events.c +++ b/res/res_stasis_json_events.c @@ -259,6 +259,56 @@ struct ast_json *stasis_json_event_playback_started_create( return ast_json_ref(message); } +struct ast_json *stasis_json_event_channel_varset_create( + struct ast_channel_snapshot *channel_snapshot, + struct ast_json *blob + ) +{ + RAII_VAR(struct ast_json *, message, NULL, ast_json_unref); + RAII_VAR(struct ast_json *, event, NULL, ast_json_unref); + struct ast_json *validator; + int ret; + + ast_assert(channel_snapshot != NULL); + ast_assert(blob != NULL); + ast_assert(ast_json_object_get(blob, "channel") == NULL); + ast_assert(ast_json_object_get(blob, "type") == NULL); + + validator = ast_json_object_get(blob, "variable"); + if (validator) { + /* do validation? XXX */ + } else { + /* fail message generation if the required parameter doesn't exist */ + return NULL; + } + + validator = ast_json_object_get(blob, "value"); + if (validator) { + /* do validation? XXX */ + } else { + /* fail message generation if the required parameter doesn't exist */ + return NULL; + } + + event = ast_json_deep_copy(blob); + if (!event) { + return NULL; + } + + ret = ast_json_object_set(event, + "channel", ast_channel_snapshot_to_json(channel_snapshot)); + if (ret) { + return NULL; + } + + message = ast_json_pack("{s: o}", "channel_varset", ast_json_ref(event)); + if (!message) { + return NULL; + } + + return ast_json_ref(message); +} + struct ast_json *stasis_json_event_bridge_destroyed_create( struct ast_bridge_snapshot *bridge_snapshot ) @@ -370,8 +420,8 @@ struct ast_json *stasis_json_event_channel_destroyed_create( return ast_json_ref(message); } -struct ast_json *stasis_json_event_channel_varset_create( - struct ast_channel_snapshot *channel_snapshot, +struct ast_json *stasis_json_event_bridge_merged_create( + struct ast_bridge_snapshot *bridge_snapshot, struct ast_json *blob ) { @@ -380,20 +430,12 @@ struct ast_json *stasis_json_event_channel_varset_create( struct ast_json *validator; int ret; - ast_assert(channel_snapshot != NULL); + ast_assert(bridge_snapshot != NULL); ast_assert(blob != NULL); - ast_assert(ast_json_object_get(blob, "channel") == NULL); + ast_assert(ast_json_object_get(blob, "bridge") == NULL); ast_assert(ast_json_object_get(blob, "type") == NULL); - validator = ast_json_object_get(blob, "variable"); - if (validator) { - /* do validation? XXX */ - } else { - /* fail message generation if the required parameter doesn't exist */ - return NULL; - } - - validator = ast_json_object_get(blob, "value"); + validator = ast_json_object_get(blob, "bridge_from"); if (validator) { /* do validation? XXX */ } else { @@ -407,12 +449,12 @@ struct ast_json *stasis_json_event_channel_varset_create( } ret = ast_json_object_set(event, - "channel", ast_channel_snapshot_to_json(channel_snapshot)); + "bridge", ast_bridge_snapshot_to_json(bridge_snapshot)); if (ret) { return NULL; } - message = ast_json_pack("{s: o}", "channel_varset", ast_json_ref(event)); + message = ast_json_pack("{s: o}", "bridge_merged", ast_json_ref(event)); if (!message) { return NULL; } diff --git a/res/res_stasis_json_events.exports.in b/res/res_stasis_json_events.exports.in index 8be4c849b..5865c026b 100644 --- a/res/res_stasis_json_events.exports.in +++ b/res/res_stasis_json_events.exports.in @@ -6,10 +6,11 @@ LINKER_SYMBOL_PREFIXstasis_json_event_channel_snapshot_create; LINKER_SYMBOL_PREFIXstasis_json_event_channel_caller_id_create; LINKER_SYMBOL_PREFIXstasis_json_event_playback_started_create; + LINKER_SYMBOL_PREFIXstasis_json_event_channel_varset_create; LINKER_SYMBOL_PREFIXstasis_json_event_bridge_destroyed_create; LINKER_SYMBOL_PREFIXstasis_json_event_application_replaced_create; LINKER_SYMBOL_PREFIXstasis_json_event_channel_destroyed_create; - LINKER_SYMBOL_PREFIXstasis_json_event_channel_varset_create; + LINKER_SYMBOL_PREFIXstasis_json_event_bridge_merged_create; LINKER_SYMBOL_PREFIXstasis_json_event_channel_left_bridge_create; LINKER_SYMBOL_PREFIXstasis_json_event_channel_created_create; LINKER_SYMBOL_PREFIXstasis_json_event_stasis_start_create; diff --git a/res/stasis/app.c b/res/stasis/app.c index 229f4bb20..31e15c221 100644 --- a/res/stasis/app.c +++ b/res/stasis/app.c @@ -38,6 +38,12 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") */ #define APP_CHANNELS_BUCKETS 7 +/*! + * \brief Number of buckets for the bridges container for app instances. Remember + * to keep it a prime number! + */ +#define APP_BRIDGES_BUCKETS 7 + struct app { /*! Callback function for this application. */ stasis_app_cb handler; @@ -45,6 +51,8 @@ struct app { void *data; /*! List of channel identifiers this app instance is interested in */ struct ao2_container *channels; + /*! List of bridge identifiers this app instance owns */ + struct ao2_container *bridges; /*! Name of the Stasis application */ char name[]; }; @@ -57,6 +65,8 @@ static void app_dtor(void *obj) app->data = NULL; ao2_cleanup(app->channels); app->channels = NULL; + ao2_cleanup(app->bridges); + app->bridges = NULL; } struct app *app_create(const char *name, stasis_app_cb handler, void *data) @@ -84,6 +94,11 @@ struct app *app_create(const char *name, stasis_app_cb handler, void *data) return NULL; } + app->bridges = ast_str_container_alloc(APP_BRIDGES_BUCKETS); + if (!app->bridges) { + return NULL; + } + ao2_ref(app, +1); return app; } @@ -106,6 +121,22 @@ void app_remove_channel(struct app* app, const struct ast_channel *chan) ao2_find(app->channels, ast_channel_uniqueid(chan), OBJ_KEY | OBJ_NODATA | OBJ_UNLINK); } +int app_add_bridge(struct app *app, const char *uniqueid) +{ + ast_assert(uniqueid != NULL); + ast_assert(app != NULL); + + return ast_str_container_add(app->bridges, uniqueid) ? -1 : 0; +} + +void app_remove_bridge(struct app* app, const char *uniqueid) +{ + ast_assert(uniqueid != NULL); + ast_assert(app != NULL); + + ao2_find(app->bridges, uniqueid, OBJ_KEY | OBJ_NODATA | OBJ_UNLINK | OBJ_MULTIPLE); +} + /*! * \brief Send a message to the given application. * \param app App to send the message to. @@ -137,3 +168,10 @@ int app_is_watching_channel(struct app *app, const char *uniqueid) found = ao2_find(app->channels, uniqueid, OBJ_KEY); return found != NULL; } + +int app_is_watching_bridge(struct app *app, const char *uniqueid) +{ + RAII_VAR(char *, found, NULL, ao2_cleanup); + found = ao2_find(app->bridges, uniqueid, OBJ_KEY); + return found != NULL; +} diff --git a/res/stasis/app.h b/res/stasis/app.h index c9e63502d..7a5405a89 100644 --- a/res/stasis/app.h +++ b/res/stasis/app.h @@ -135,4 +135,32 @@ int app_add_channel(struct app *app, const struct ast_channel *chan); */ void app_remove_channel(struct app *app, const struct ast_channel *chan); +/*! + * \brief Add a bridge to an application's watch list by uniqueid. + * + * \param app Application. + * \param bridge Bridge to watch. + * \return 0 on success. + * \return Non-zero on error. + */ +int app_add_bridge(struct app *app, const char *uniqueid); + +/*! + * \brief Remove a bridge from an application's watch list by uniqueid. + * + * \param app Application. + * \param bridge Bridge to remove. + */ +void app_remove_bridge(struct app* app, const char *uniqueid); + +/*! + * \brief Checks if an application is watching a given bridge. + * + * \param app Application. + * \param uniqueid Uniqueid of the bridge to check. + * \return True (non-zero) if \a app is watching bridge with given \a uniqueid + * \return False (zero) if \a app isn't. + */ +int app_is_watching_bridge(struct app *app, const char *uniqueid); + #endif /* _ASTERISK_RES_STASIS_APP_H */ diff --git a/res/stasis/control.c b/res/stasis/control.c index 1663dd8a6..06f36728f 100644 --- a/res/stasis/control.c +++ b/res/stasis/control.c @@ -31,6 +31,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "command.h" #include "control.h" +#include "asterisk/bridging.h" +#include "asterisk/bridging_features.h" struct stasis_app_control { /*! Queue of commands to dispatch on the channel */ @@ -200,3 +202,10 @@ int control_dispatch_all(struct stasis_app_control *control, ao2_iterator_destroy(&i); 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) +{ + return ast_bridge_remove(bridge, control->channel); +} diff --git a/res/stasis_http/resource_bridges.c b/res/stasis_http/resource_bridges.c index ca48ee725..6dad91116 100644 --- a/res/stasis_http/resource_bridges.c +++ b/res/stasis_http/resource_bridges.c @@ -32,32 +32,211 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "resource_bridges.h" +#include "asterisk/stasis.h" +#include "asterisk/stasis_bridging.h" +#include "asterisk/stasis_app.h" +#include "asterisk/channel.h" +#include "asterisk/bridging.h" + +/*! + * \brief Finds a bridge, filling the response with an error, if appropriate. + * + * \param[out] response Response to fill with an error if control is not found. + * \param bridge_id ID of the bridge to lookup. + * + * \return Bridget. + * \return \c NULL if bridge does not exist. + */ +static struct ast_bridge *find_bridge( + struct stasis_http_response *response, + const char *bridge_id) +{ + RAII_VAR(struct ast_bridge *, bridge, NULL, ao2_cleanup); + + ast_assert(response != NULL); + + bridge = stasis_app_bridge_find_by_id(bridge_id); + if (bridge == NULL) { + RAII_VAR(struct ast_bridge_snapshot *, snapshot, + ast_bridge_snapshot_get_latest(bridge_id), ao2_cleanup); + if (!snapshot) { + stasis_http_response_error(response, 404, "Not found", + "Bridge not found"); + return NULL; + } + + stasis_http_response_error(response, 409, "Conflict", + "Bridge not in Stasis application"); + return NULL; + } + + ao2_ref(bridge, +1); + return bridge; +} + +/*! + * \brief Finds the control object for a channel, filling the response with an + * error, if appropriate. + * \param[out] response Response to fill with an error if control is not found. + * \param channel_id ID of the channel to lookup. + * \return Channel control object. + * \return \c NULL if control object does not exist. + */ +static struct stasis_app_control *find_channel_control( + struct stasis_http_response *response, + const char *channel_id) +{ + RAII_VAR(struct stasis_app_control *, control, NULL, ao2_cleanup); + + ast_assert(response != NULL); + + control = stasis_app_control_find_by_channel_id(channel_id); + if (control == NULL) { + stasis_http_response_error(response, 422, "Unprocessable Entity", + "Channel not in Stasis application"); + return NULL; + } + + ao2_ref(control, +1); + return control; +} void stasis_http_add_channel_to_bridge(struct ast_variable *headers, struct ast_add_channel_to_bridge_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_add_channel_to_bridge\n"); + RAII_VAR(struct ast_bridge *, bridge, find_bridge(response, args->bridge_id), ao2_cleanup); + RAII_VAR(struct stasis_app_control *, control, NULL, ao2_cleanup); + if (!bridge) { + return; + } + + control = find_channel_control(response, args->channel); + if (!control) { + return; + } + + stasis_app_control_add_channel_to_bridge(control, bridge); + stasis_http_response_no_content(response); } + void stasis_http_remove_channel_from_bridge(struct ast_variable *headers, struct ast_remove_channel_from_bridge_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_remove_channel_from_bridge\n"); + RAII_VAR(struct ast_bridge *, bridge, find_bridge(response, args->bridge_id), ao2_cleanup); + RAII_VAR(struct stasis_app_control *, control, NULL, ao2_cleanup); + if (!bridge) { + return; + } + + control = find_channel_control(response, args->channel); + if (!control) { + return; + } + + /* BUGBUG this should make sure the bridge requested for removal is actually + * the bridge the channel is in. This will be possible once the bridge uniqueid + * is added to the channel snapshot. A 409 response should be issued if the bridge + * uniqueids don't match */ + if (stasis_app_control_remove_channel_from_bridge(control, bridge)) { + stasis_http_response_error(response, 500, "Internal Error", + "Could not remove channel from bridge"); + return; + } + + stasis_http_response_no_content(response); } + void stasis_http_record_bridge(struct ast_variable *headers, struct ast_record_bridge_args *args, struct stasis_http_response *response) { ast_log(LOG_ERROR, "TODO: stasis_http_record_bridge\n"); } + void stasis_http_get_bridge(struct ast_variable *headers, struct ast_get_bridge_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_get_bridge\n"); + RAII_VAR(struct ast_bridge_snapshot *, snapshot, ast_bridge_snapshot_get_latest(args->bridge_id), ao2_cleanup); + if (!snapshot) { + stasis_http_response_error( + response, 404, "Not Found", + "Bridge not found"); + return; + } + + stasis_http_response_ok(response, + ast_bridge_snapshot_to_json(snapshot)); } + void stasis_http_delete_bridge(struct ast_variable *headers, struct ast_delete_bridge_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_delete_bridge\n"); + RAII_VAR(struct ast_bridge *, bridge, find_bridge(response, args->bridge_id), ao2_cleanup); + if (!bridge) { + return; + } + + stasis_app_bridge_destroy(args->bridge_id); + stasis_http_response_no_content(response); } + void stasis_http_get_bridges(struct ast_variable *headers, struct ast_get_bridges_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_get_bridges\n"); + RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup); + RAII_VAR(struct ao2_container *, snapshots, NULL, ao2_cleanup); + RAII_VAR(struct ast_json *, json, NULL, ast_json_unref); + struct ao2_iterator i; + void *obj; + + caching_topic = ast_bridge_topic_all_cached(); + if (!caching_topic) { + stasis_http_response_error( + response, 500, "Internal Server Error", + "Message bus not initialized"); + return; + } + ao2_ref(caching_topic, +1); + + snapshots = stasis_cache_dump(caching_topic, ast_bridge_snapshot_type()); + if (!snapshots) { + stasis_http_response_alloc_failed(response); + return; + } + + json = ast_json_array_create(); + if (!json) { + stasis_http_response_alloc_failed(response); + return; + } + + i = ao2_iterator_init(snapshots, 0); + while ((obj = ao2_iterator_next(&i))) { + RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup); + struct ast_bridge_snapshot *snapshot = stasis_message_data(msg); + if (ast_json_array_append(json, ast_bridge_snapshot_to_json(snapshot))) { + stasis_http_response_alloc_failed(response); + return; + } + } + ao2_iterator_destroy(&i); + + stasis_http_response_ok(response, ast_json_ref(json)); } + void stasis_http_new_bridge(struct ast_variable *headers, struct ast_new_bridge_args *args, struct stasis_http_response *response) { - ast_log(LOG_ERROR, "TODO: stasis_http_new_bridge\n"); + RAII_VAR(struct ast_bridge *, bridge, stasis_app_bridge_create(args->type), ao2_cleanup); + RAII_VAR(struct ast_bridge_snapshot *, snapshot, NULL, ao2_cleanup); + + if (!bridge) { + stasis_http_response_error( + response, 500, "Internal Error", + "Unable to create bridge"); + return; + } + + snapshot = ast_bridge_snapshot_create(bridge); + if (!snapshot) { + stasis_http_response_error( + response, 500, "Internal Error", + "Unable to create snapshot for new bridge"); + return; + } + + stasis_http_response_ok(response, + ast_bridge_snapshot_to_json(snapshot)); } diff --git a/res/stasis_http/resource_bridges.h b/res/stasis_http/resource_bridges.h index db6db205b..ec1992f26 100644 --- a/res/stasis_http/resource_bridges.h +++ b/res/stasis_http/resource_bridges.h @@ -84,7 +84,7 @@ struct ast_delete_bridge_args { const char *bridge_id; }; /*! - * \brief Shut down a bridge bridge. + * \brief Shut down a bridge. * * If any channels are in this bridge, they will be removed and resume whatever they were doing beforehand. * diff --git a/res/stasis_json/resource_events.h b/res/stasis_json/resource_events.h index d63181788..a2af30daa 100644 --- a/res/stasis_json/resource_events.h +++ b/res/stasis_json/resource_events.h @@ -122,6 +122,22 @@ struct ast_json *stasis_json_event_playback_started_create( ); /*! + * \brief Channel variable changed. + * + * \param channel The channel on which the variable was set. + * \param blob JSON blob containing the following parameters: + * - variable: string - The variable that changed. (required) + * - value: string - The new value of the variable. (required) + * + * \retval NULL on error + * \retval JSON (ast_json) describing the event + */ +struct ast_json *stasis_json_event_channel_varset_create( + struct ast_channel_snapshot *channel_snapshot, + struct ast_json *blob + ); + +/*! * \brief Notification that a bridge has been destroyed. * * \param bridge The bridge to be used to generate this event @@ -163,18 +179,17 @@ struct ast_json *stasis_json_event_channel_destroyed_create( ); /*! - * \brief Channel variable changed. + * \brief Notification that one bridge has merged into another. * - * \param channel The channel on which the variable was set. + * \param bridge The bridge to be used to generate this event * \param blob JSON blob containing the following parameters: - * - variable: string - The variable that changed. (required) - * - value: string - The new value of the variable. (required) + * - bridge_from: Bridge (required) * * \retval NULL on error * \retval JSON (ast_json) describing the event */ -struct ast_json *stasis_json_event_channel_varset_create( - struct ast_channel_snapshot *channel_snapshot, +struct ast_json *stasis_json_event_bridge_merged_create( + struct ast_bridge_snapshot *bridge_snapshot, struct ast_json *blob ); @@ -318,15 +333,17 @@ struct ast_json *stasis_json_event_stasis_end_create( * - caller_presentation: integer (required) * PlaybackStarted * - playback: Playback (required) + * ChannelVarset + * - variable: string (required) + * - value: string (required) * BridgeDestroyed * ApplicationReplaced * - application: string (required) * ChannelDestroyed * - cause: integer (required) * - cause_txt: string (required) - * ChannelVarset - * - variable: string (required) - * - value: string (required) + * BridgeMerged + * - bridge_from: Bridge (required) * ChannelLeftBridge * ChannelCreated * StasisStart @@ -342,24 +359,25 @@ struct ast_json *stasis_json_event_stasis_end_create( * ChannelDtmfReceived * - digit: string (required) * Event - * - stasis_start: StasisStart + * - channel_varset: ChannelVarset * - channel_created: ChannelCreated * - channel_destroyed: ChannelDestroyed * - channel_entered_bridge: ChannelEnteredBridge * - channel_left_bridge: ChannelLeftBridge + * - bridge_merged: BridgeMerged * - channel_dialplan: ChannelDialplan - * - channel_varset: ChannelVarset * - application_replaced: ApplicationReplaced * - channel_state_change: ChannelStateChange * - bridge_created: BridgeCreated * - application: string (required) * - channel_hangup_request: ChannelHangupRequest * - channel_userevent: ChannelUserevent - * - playback_started: PlaybackStarted + * - stasis_start: StasisStart * - channel_snapshot: ChannelSnapshot * - channel_dtmf_received: ChannelDtmfReceived * - channel_caller_id: ChannelCallerId * - bridge_destroyed: BridgeDestroyed + * - playback_started: PlaybackStarted * - playback_finished: PlaybackFinished * - stasis_end: StasisEnd * StasisEnd diff --git a/rest-api/api-docs/bridges.json b/rest-api/api-docs/bridges.json index fd0971a4d..e4a33f8c3 100644 --- a/rest-api/api-docs/bridges.json +++ b/rest-api/api-docs/bridges.json @@ -66,7 +66,7 @@ }, { "httpMethod": "DELETE", - "summary": "Shut down a bridge bridge.", + "summary": "Shut down a bridge.", "notes": "If any channels are in this bridge, they will be removed and resume whatever they were doing beforehand.", "nickname": "deleteBridge", "responseClass": "void", diff --git a/rest-api/api-docs/events.json b/rest-api/api-docs/events.json index 0e0a822cf..c5f38862f 100644 --- a/rest-api/api-docs/events.json +++ b/rest-api/api-docs/events.json @@ -56,6 +56,7 @@ "application_replaced": { "type": "ApplicationReplaced" }, "bridge_created": { "type": "BridgeCreated" }, "bridge_destroyed": { "type": "BridgeDestroyed" }, + "bridge_merged": { "type": "BridgeMerged" }, "channel_created": { "type": "ChannelCreated" }, "channel_destroyed": { "type": "ChannelDestroyed" }, "channel_snapshot": { "type": "ChannelSnapshot" }, @@ -127,6 +128,20 @@ } } }, + "BridgeMerged": { + "id": "BridgeMerged", + "description": "Notification that one bridge has merged into another.", + "properties": { + "bridge": { + "required": true, + "type": "Bridge" + }, + "bridge_from": { + "required": true, + "type": "Bridge" + } + } + }, "ChannelCreated": { "id": "ChannelCreated", "description": "Notification that a channel has been created.", |