[asterisk-commits] kmoore: branch kmoore/stasis-bridging-channel_events r385813 - /team/kmoore/s...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Mon Apr 15 14:35:32 CDT 2013
Author: kmoore
Date: Mon Apr 15 14:35:28 2013
New Revision: 385813
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=385813
Log:
Re-apply changes from app_stasis to res_stasis with enhancements in dealing with channel blob messages
Modified:
team/kmoore/stasis-bridging-channel_events/res/res_stasis.c
Modified: team/kmoore/stasis-bridging-channel_events/res/res_stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/kmoore/stasis-bridging-channel_events/res/res_stasis.c?view=diff&rev=385813&r1=385812&r2=385813
==============================================================================
--- team/kmoore/stasis-bridging-channel_events/res/res_stasis.c (original)
+++ team/kmoore/stasis-bridging-channel_events/res/res_stasis.c Mon Apr 15 14:35:28 2013
@@ -39,6 +39,8 @@
#include "asterisk/stasis_app.h"
#include "asterisk/stasis_channels.h"
#include "asterisk/strings.h"
+#include "asterisk/stasis_message_router.h"
+#include "asterisk/callerid.h"
/*!
* \brief Number of buckets for the Stasis application hash table. Remember to
@@ -53,12 +55,30 @@
#define CONTROLS_NUM_BUCKETS 127
/*!
+ * \brief Number of buckets for the channels container for app instances. Remember
+ * to keep it a prime number!
+ */
+#define APP_CHANNELS_BUCKETS 7
+
+/*!
+ * \brief Number of buckets for the blob_handlers container. Remember to keep
+ * it a prime number!
+ */
+#define BLOB_HANDLER_BUCKETS 7
+
+/*!
* \brief Stasis application container. Please call apps_registry() instead of
* directly accessing.
*/
struct ao2_container *__apps_registry;
struct ao2_container *__app_controls;
+
+/*! \brief Container for handlers for channel blob messages */
+struct ao2_container *blob_handlers;
+
+/*! \brief Message router for the channel caching topic */
+struct stasis_message_router *app_channel_router;
/*! Ref-counting accessor for the stasis applications container */
static struct ao2_container *apps_registry(void)
@@ -78,6 +98,8 @@
stasis_app_cb handler;
/*! Opaque data to hand to callback function. */
void *data;
+ /*! List of channel identifiers this app instance is interested in */
+ struct ao2_container *channels;
/*! Name of the Stasis application */
char name[];
};
@@ -88,12 +110,14 @@
ao2_cleanup(app->data);
app->data = NULL;
+ ao2_cleanup(app->channels);
+ app->channels = NULL;
}
/*! Constructor for \ref app. */
static struct app *app_create(const char *name, stasis_app_cb handler, void *data)
{
- struct app *app;
+ RAII_VAR(struct app *, app, NULL, ao2_cleanup);
size_t size;
ast_assert(name != NULL);
@@ -111,6 +135,12 @@
ao2_ref(data, +1);
app->data = data;
+ app->channels = ast_str_container_alloc(APP_CHANNELS_BUCKETS);
+ if (!app->channels) {
+ return NULL;
+ }
+
+ ao2_ref(app, +1);
return app;
}
@@ -135,6 +165,27 @@
} else {
return 0;
}
+}
+
+static int app_add_channel(struct app* app, const struct ast_channel *chan)
+{
+ const char *uniqueid;
+ ast_assert(chan != NULL);
+ ast_assert(app != NULL);
+
+ uniqueid = ast_channel_uniqueid(chan);
+ if (!ast_str_container_add(app->channels, uniqueid)) {
+ return -1;
+ }
+ return 0;
+}
+
+static void app_remove_channel(struct app* app, const struct ast_channel *chan)
+{
+ ast_assert(chan != NULL);
+ ast_assert(app != NULL);
+
+ ao2_find(app->channels, ast_channel_uniqueid(chan), OBJ_KEY | OBJ_NODATA | OBJ_UNLINK);
}
/*!
@@ -231,6 +282,23 @@
{
SCOPED_AO2LOCK(lock, control);
control->continue_to_dialplan = 1;
+}
+
+/*! \brief Typedef for blob handler callbacks */
+typedef struct ast_json *(*blob_handler_cb)(struct ast_channel_blob *);
+
+/*! \brief AO2 refcounted object linking channel blob json type to its handler callback */
+struct blob_handler {
+ char *type;
+ blob_handler_cb handler;
+};
+
+static void blob_handler_dtor(void *obj)
+{
+ struct blob_handler *handler = obj;
+
+ ast_free(handler->type);
+ handler->type = NULL;
}
static struct ast_json *app_event_create(
@@ -331,65 +399,264 @@
return 0;
}
-static void dtmf_handler(struct app *app, struct ast_channel_blob *obj)
-{
- RAII_VAR(struct ast_json *, extra, NULL, ast_json_unref);
+static int app_watching_channel_cb(void *obj, void *arg, int flags)
+{
+ RAII_VAR(char *, uniqueid, NULL, ao2_cleanup);
+ struct app *app = obj;
+ struct ast_channel_snapshot *snapshot = arg;
+
+ uniqueid = ao2_find(app->channels, snapshot->uniqueid, OBJ_KEY);
+ return uniqueid ? CMP_MATCH : 0;
+}
+
+static struct ao2_container *get_watching_apps(struct ast_channel_snapshot *snapshot)
+{
+ RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
+ struct ao2_container *watching_apps;
+ RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy);
+ ast_assert(snapshot != NULL);
+ ast_assert(apps != NULL);
+
+ watching_apps_iter = ao2_callback(apps, OBJ_MULTIPLE, app_watching_channel_cb, snapshot);
+ watching_apps = watching_apps_iter->c;
+
+ if (!ao2_container_count(watching_apps)) {
+ return NULL;
+ }
+
+ ao2_ref(watching_apps, +1);
+ return watching_apps_iter->c;
+}
+
+/*! \brief Typedef for callbacks that get called on channel snapshot updates */
+typedef struct ast_json *(*snapshot_monitor)(
+ struct ast_channel_snapshot *old_snapshot,
+ struct ast_channel_snapshot *new_snapshot);
+
+/*! \brief Handle channel state changes */
+static struct ast_json *channel_state(
+ struct ast_channel_snapshot *old_snapshot,
+ struct ast_channel_snapshot *new_snapshot)
+{
+ RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+ char *event_name = NULL;
+
+ if (!old_snapshot) {
+ event_name = "channel-event-create";
+ } else if (!new_snapshot) {
+ event_name = "channel-event-destroy";
+ json = ast_json_pack("{s: i, s: s}",
+ "Cause", old_snapshot->hangupcause,
+ "Cause-txt", ast_cause2str(old_snapshot->hangupcause));
+ if (!json) {
+ return NULL;
+ }
+ } else if (old_snapshot->state != new_snapshot->state) {
+ event_name = "channel-event-state";
+ }
+
+ if (!event_name) {
+ return NULL;
+ }
+
+ return app_event_create(event_name, new_snapshot ? new_snapshot : old_snapshot, json);
+}
+
+/*!
+ * \brief Compares the context, exten and priority of two snapshots.
+ * \param old_snapshot Old snapshot
+ * \param new_snapshot New snapshot
+ * \return True (non-zero) if context, exten or priority are identical.
+ * \return False (zero) if context, exten and priority changed.
+ */
+static inline int cep_equal(
+ const struct ast_channel_snapshot *old_snapshot,
+ const struct ast_channel_snapshot *new_snapshot)
+{
+ ast_assert(old_snapshot != NULL);
+ ast_assert(new_snapshot != NULL);
+
+ /* We actually get some snapshots with CEP set, but before the
+ * application is set. Since empty application is invalid, we treat
+ * setting the application from nothing as a CEP change.
+ */
+ if (ast_strlen_zero(old_snapshot->appl) &&
+ !ast_strlen_zero(new_snapshot->appl)) {
+ return 0;
+ }
+
+ return old_snapshot->priority == new_snapshot->priority &&
+ strcmp(old_snapshot->context, new_snapshot->context) == 0 &&
+ strcmp(old_snapshot->exten, new_snapshot->exten) == 0;
+}
+
+static struct ast_json *channel_dialplan(
+ struct ast_channel_snapshot *old_snapshot,
+ struct ast_channel_snapshot *new_snapshot)
+{
+ RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+
+ /* No Newexten event on cache clear */
+ if (!new_snapshot) {
+ return NULL;
+ }
+
+ /* Empty application is not valid for a Newexten event */
+ if (ast_strlen_zero(new_snapshot->appl)) {
+ return NULL;
+ }
+
+ if (old_snapshot && cep_equal(old_snapshot, new_snapshot)) {
+ return NULL;
+ }
+
+ json = ast_json_pack("{s: s, s: s}",
+ "Application", new_snapshot->appl,
+ "ApplicationData", new_snapshot->data);
+ if (!json) {
+ return NULL;
+ }
+
+ return app_event_create("channel-event-dialplan", new_snapshot, json);
+}
+
+/*!
+ * \brief Compares the callerid info of two snapshots.
+ * \param old_snapshot Old snapshot
+ * \param new_snapshot New snapshot
+ * \return True (non-zero) if callerid are identical.
+ * \return False (zero) if callerid changed.
+ */
+static inline int caller_id_equal(
+ const struct ast_channel_snapshot *old_snapshot,
+ const struct ast_channel_snapshot *new_snapshot)
+{
+ ast_assert(old_snapshot != NULL);
+ ast_assert(new_snapshot != NULL);
+ return strcmp(old_snapshot->caller_number, new_snapshot->caller_number) == 0 &&
+ strcmp(old_snapshot->caller_name, new_snapshot->caller_name) == 0;
+}
+
+static struct ast_json *channel_callerid(
+ struct ast_channel_snapshot *old_snapshot,
+ struct ast_channel_snapshot *new_snapshot)
+{
+ RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
+
+ /* No NewCallerid event on cache clear or first event */
+ if (!old_snapshot || !new_snapshot) {
+ return NULL;
+ }
+
+ if (caller_id_equal(old_snapshot, new_snapshot)) {
+ return NULL;
+ }
+
+ json = ast_json_pack("{s: i, s: s}",
+ "CallerPresentation", new_snapshot->caller_pres,
+ "CallerPresentation-txt", ast_describe_caller_presentation(new_snapshot->caller_pres));
+ if (!json) {
+ return NULL;
+ }
+
+ return app_event_create("channel-event-callerid", new_snapshot, json);
+}
+
+static struct ast_json *channel_snapshot(
+ struct ast_channel_snapshot *old_snapshot,
+ struct ast_channel_snapshot *new_snapshot)
+{
+ if (!new_snapshot) {
+ return NULL;
+ }
+
+ return app_event_create("channel-snapshot", new_snapshot, NULL);
+}
+
+snapshot_monitor monitors[] = {
+ channel_snapshot,
+ channel_state,
+ channel_dialplan,
+ channel_callerid
+};
+
+static int app_send_cb(void *obj, void *arg, int flags)
+{
+ struct app *app = obj;
+ struct ast_json *msg = arg;
+
+ app_send(app, msg);
+ return 0;
+}
+
+static void sub_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_channel_snapshot *new_snapshot = stasis_message_data(update->new_snapshot);
+ struct ast_channel_snapshot *old_snapshot = stasis_message_data(update->old_snapshot);
+ int i;
+
+ watching_apps = get_watching_apps(new_snapshot ? new_snapshot : old_snapshot);
+ if (!watching_apps) {
+ return;
+ }
+
+ for (i = 0; i < ARRAY_LEN(monitors); ++i) {
+ RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
+
+ msg = monitors[i](old_snapshot, new_snapshot);
+ if (msg) {
+ ao2_callback(watching_apps, OBJ_NODATA, app_send_cb, msg);
+ }
+ }
+}
+
+static void distribute_message(struct ao2_container *apps, struct ast_json *msg)
+{
+ ao2_callback(apps, OBJ_NODATA, app_send_cb, msg);
+}
+
+static struct ast_json *handle_blob(struct ast_channel_blob *obj)
+{
+ RAII_VAR(struct blob_handler *, handler, NULL, ao2_cleanup);
+ const char *handler_type = ast_channel_blob_json_type(obj);
+
+ handler = ao2_find(blob_handlers, handler_type, OBJ_KEY);
+ if (!handler) {
+ return NULL;
+ }
+
+ return handler->handler(obj);
+}
+
+static void sub_blob_handler(void *data,
+ struct stasis_subscription *sub,
+ struct stasis_topic *topic,
+ struct stasis_message *message)
+{
RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
- const char *direction;
-
- /* To simplify events, we'll only generate on receive */
- direction = ast_json_string_get(
- ast_json_object_get(obj->blob, "direction"));
-
- if (strcmp("Received", direction) != 0) {
+ RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
+ struct ast_channel_blob *obj = stasis_message_data(message);
+
+ if (!obj->snapshot) {
return;
}
- extra = ast_json_pack(
- "{s: o}",
- "digit", ast_json_ref(ast_json_object_get(obj->blob, "digit")));
- if (!extra) {
+ watching_apps = get_watching_apps(obj->snapshot);
+ if (!watching_apps) {
return;
}
- msg = app_event_create("dtmf-received", obj->snapshot, extra);
+ msg = handle_blob(obj);
if (!msg) {
return;
}
- app_send(app, msg);
-}
-
-static void blob_handler(struct app *app, struct ast_channel_blob *blob)
-{
- /* To simplify events, we'll only generate on DTMF end */
- if (strcmp(ast_channel_blob_json_type(blob), "dtmf_end") == 0) {
- dtmf_handler(app, blob);
- }
-}
-
-static void sub_handler(void *data, struct stasis_subscription *sub,
- struct stasis_topic *topic,
- struct stasis_message *message)
-{
- struct app *app = data;
- if (ast_channel_snapshot_type() == stasis_message_type(message)) {
- RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
- struct ast_channel_snapshot *snapshot =
- stasis_message_data(message);
-
- msg = app_event_create("channel-state-change", snapshot, NULL);
- if (!msg) {
- return;
- }
- app_send(app, msg);
- } else if (ast_channel_blob_type() == stasis_message_type(message)) {
- struct ast_channel_blob *blob = stasis_message_data(message);
- blob_handler(app, blob);
- }
- if (stasis_subscription_final_message(sub, message)) {
- ao2_cleanup(data);
- }
+ distribute_message(watching_apps, msg);
}
/*!
@@ -417,8 +684,6 @@
RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
RAII_VAR(struct app *, app, NULL, ao2_cleanup);
RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
- RAII_VAR(struct stasis_subscription *, subscription, NULL,
- stasis_unsubscribe);
int res = 0;
int hungup = 0;
@@ -443,19 +708,15 @@
ao2_link(controls, control);
}
- subscription =
- stasis_subscribe(ast_channel_topic(chan), sub_handler, app);
- if (subscription == NULL) {
- ast_log(LOG_ERROR, "Error subscribing app %s to channel %s\n",
- app_name, ast_channel_name(chan));
- return -1;
- }
- ao2_ref(app, +1); /* subscription now has a reference */
-
res = send_start_msg(app, chan, argc, argv);
if (res != 0) {
ast_log(LOG_ERROR, "Error sending start message to %s\n", app_name);
return res;
+ }
+
+ if (app_add_channel(app, chan)) {
+ ast_log(LOG_ERROR, "Error adding listener for channel %s to app %s\n", ast_channel_name(chan), app_name);
+ return -1;
}
while (!hungup && !control_continue_test_and_reset(control) && ast_waitfor(chan, -1) > -1) {
@@ -479,6 +740,7 @@
}
}
+ app_remove_channel(app, chan);
res = send_end_msg(app, chan);
if (res != 0) {
ast_log(LOG_ERROR,
@@ -551,6 +813,70 @@
}
}
+/* To simplify events, we'll only generate on DTMF end (dtmf_end type) */
+static struct ast_json *handle_blob_dtmf(struct ast_channel_blob *obj)
+{
+ RAII_VAR(struct ast_json *, extra, NULL, ast_json_unref);
+ RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
+ const char *direction;
+
+ /* To simplify events, we'll only generate on receive */
+ direction = ast_json_string_get(
+ ast_json_object_get(obj->blob, "direction"));
+
+ if (strcmp("Received", direction) != 0) {
+ return NULL;
+ }
+
+ extra = ast_json_pack(
+ "{s: o}",
+ "digit", ast_json_ref(ast_json_object_get(obj->blob, "digit")));
+ if (!extra) {
+ return NULL;
+ }
+
+ return app_event_create("channel-event-dtmf-received", obj->snapshot, extra);
+}
+
+static struct ast_json *handle_blob_generic(struct ast_channel_blob *obj)
+{
+ RAII_VAR(struct ast_str *, event_name, ast_str_create(32), ast_free);
+
+ ast_str_set(&event_name, 0, "channel-event-%s", ast_channel_blob_json_type(obj));
+
+ return app_event_create(ast_str_buffer(event_name), obj->snapshot, obj->blob);
+}
+
+static void register_blob_handler(const char *blob_type, blob_handler_cb blob_type_handler_cb)
+{
+ RAII_VAR(struct blob_handler *, handler, ao2_alloc(sizeof(*handler), blob_handler_dtor), ao2_cleanup);
+
+ if (!handler) {
+ return;
+ }
+
+ handler->handler = blob_type_handler_cb;
+ handler->type = ast_strdup(blob_type);
+ if (!handler->type) {
+ return;
+ }
+
+ ao2_link(blob_handlers, handler);
+}
+
+static int blob_handler_hash(const void *obj, const int flags)
+{
+ const char *handler_type = (flags & OBJ_KEY) ? obj : ((struct blob_handler*) obj)->type;
+ return ast_str_case_hash(handler_type);
+}
+
+static int blob_handler_cmp(void *obj, void *arg, int flags)
+{
+ struct blob_handler *opt1 = obj, *opt2 = arg;
+ const char *handler_type = (flags & OBJ_KEY) ? arg : opt2->type;
+ return strcasecmp(opt1->type, handler_type) ? 0 : CMP_MATCH | CMP_STOP;
+}
+
static int load_module(void)
{
int r = 0;
@@ -567,12 +893,33 @@
return AST_MODULE_LOAD_FAILURE;
}
+ blob_handlers = ao2_container_alloc(BLOB_HANDLER_BUCKETS, blob_handler_hash, blob_handler_cmp);
+
+ register_blob_handler("userevent", handle_blob_generic);
+ register_blob_handler("hangup_request", handle_blob_generic);
+ register_blob_handler("varset", handle_blob_generic);
+ register_blob_handler("dtmf_end", handle_blob_dtmf);
+
+ app_channel_router = stasis_message_router_create(stasis_caching_get_topic(ast_channel_topic_all_cached()));
+ if (!app_channel_router) {
+ return AST_MODULE_LOAD_FAILURE;
+ }
+
+ r |= stasis_message_router_add(app_channel_router, stasis_cache_update_type(), sub_snapshot_handler, NULL);
+ r |= stasis_message_router_add(app_channel_router, ast_channel_blob_type(), sub_blob_handler, NULL);
+
return r;
}
static int unload_module(void)
{
int r = 0;
+
+ stasis_message_router_unsubscribe(app_channel_router);
+ app_channel_router = NULL;
+
+ ao2_cleanup(blob_handlers);
+ blob_handlers = NULL;
ao2_cleanup(__apps_registry);
__apps_registry = NULL;
More information about the asterisk-commits
mailing list