kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-06 12:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Stasis-HTTP: Flesh out bridge-related capabilities

Browse Source 
This adds support for Stasis applications to receive bridge-related
messages when the application shows interest in a given bridge.

To supplement this work and test it, this also adds support for the
following bridge-related Stasis-HTTP functionality:
* GET stasis/bridges
* GET stasis/bridges/{bridgeId}
* POST stasis/bridges
* DELETE stasis/bridges/{bridgeId}
* POST stasis/bridges/{bridgeId}/addChannel
* POST stasis/bridges/{bridgeId}/removeChannel

Review: https://reviewboard.asterisk.org/r/2572/
(closes issue ASTERISK-21711)
(closes issue ASTERISK-21621)
(closes issue ASTERISK-21622)
(closes issue ASTERISK-21623)
(closes issue ASTERISK-21624)
(closes issue ASTERISK-21625)
(closes issue ASTERISK-21626)


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@391199 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Kinsey Moore
2013-06-10 13:07:11 +00:00
parent 0cec7dcdcd
commit a5bbc790e7
16 changed files with 825 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
include/asterisk/stasis_app.h
View File

@@ -186,6 +186,56 @@ void stasis_app_control_publish(
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.
*

12
include/asterisk/stasis_bridging.h
View File

@@ -206,6 +206,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

22
main/stasis_bridging.c
View File

@@ -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);

302
res/res_stasis.c
View File

@@ -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;
}

73
res/res_stasis_bridge_add.c Normal file
View File

@@ -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");

6
res/res_stasis_bridge_add.exports.in Normal file
View File

@@ -0,0 +1,6 @@
{
global:
LINKER_SYMBOL_PREFIXstasis_app_*;
local:
*;
};

72
res/res_stasis_json_events.c
View File

@@ -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;
}

3
res/res_stasis_json_events.exports.in
View File

@@ -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;

38
res/stasis/app.c
View File

@@ -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;
}

28
res/stasis/app.h
View File

@@ -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 */

9
res/stasis/control.c
View File

@@ -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);
}

191
res/stasis_http/resource_bridges.c
View File

@@ -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));
}

2
res/stasis_http/resource_bridges.h
View File

@@ -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.
*

42
res/stasis_json/resource_events.h
View File

@@ -121,6 +121,22 @@ struct ast_json *stasis_json_event_playback_started_create(
struct ast_json *blob
);
/*!
* \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.
*
@@ -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

2
rest-api/api-docs/bridges.json
View File

@@ -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",

15
rest-api/api-docs/events.json
View File

@@ -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.",