res/stasis/app.c

Go to the documentation of this file.
00001 /*
00002  * Asterisk -- An open source telephony toolkit.
00003  *
00004  * Copyright (C) 2013, Digium, Inc.
00005  *
00006  * David M. Lee, II <dlee@digium.com>
00007  *
00008  * See http://www.asterisk.org for more information about
00009  * the Asterisk project. Please do not directly contact
00010  * any of the maintainers of this project for assistance;
00011  * the project provides a web site, mailing lists and IRC
00012  * channels for your use.
00013  *
00014  * This program is free software, distributed under the terms of
00015  * the GNU General Public License Version 2. See the LICENSE file
00016  * at the top of the source tree.
00017  */
00018 
00019 /*! \file
00020  *
00021  * \brief Stasis application support.
00022  *
00023  * \author David M. Lee, II <dlee@digium.com>
00024  */
00025 
00026 #include "asterisk.h"
00027 
00028 ASTERISK_FILE_VERSION(__FILE__, "$Revision: 430940 $")
00029 
00030 #include "app.h"
00031 #include "control.h"
00032 #include "messaging.h"
00033 
00034 #include "asterisk/callerid.h"
00035 #include "asterisk/stasis_app.h"
00036 #include "asterisk/stasis_bridges.h"
00037 #include "asterisk/stasis_channels.h"
00038 #include "asterisk/stasis_endpoints.h"
00039 #include "asterisk/stasis_message_router.h"
00040 
00041 static int unsubscribe(struct stasis_app *app, const char *kind, const char *id, int terminate);
00042 
00043 struct stasis_app {
00044    /*! Aggregation topic for this application. */
00045    struct stasis_topic *topic;
00046    /*! Router for handling messages forwarded to \a topic. */
00047    struct stasis_message_router *router;
00048    /*! Router for handling messages to the bridge all \a topic. */
00049    struct stasis_message_router *bridge_router;
00050    /*! Container of the channel forwards to this app's topic. */
00051    struct ao2_container *forwards;
00052    /*! Callback function for this application. */
00053    stasis_app_cb handler;
00054    /*! Opaque data to hand to callback function. */
00055    void *data;
00056    /*! Name of the Stasis application */
00057    char name[];
00058 };
00059 
00060 enum forward_type {
00061    FORWARD_CHANNEL,
00062    FORWARD_BRIDGE,
00063    FORWARD_ENDPOINT,
00064 };
00065 
00066 /*! Subscription info for a particular channel/bridge. */
00067 struct app_forwards {
00068    /*! Count of number of times this channel/bridge has been subscribed */
00069    int interested;
00070 
00071    /*! Forward for the regular topic */
00072    struct stasis_forward *topic_forward;
00073    /*! Forward for the caching topic */
00074    struct stasis_forward *topic_cached_forward;
00075 
00076    /* Type of object being forwarded */
00077    enum forward_type forward_type;
00078    /*! Unique id of the object being forwarded */
00079    char id[];
00080 };
00081 
00082 static void forwards_dtor(void *obj)
00083 {
00084 #ifdef AST_DEVMODE
00085    struct app_forwards *forwards = obj;
00086 #endif /* AST_DEVMODE */
00087 
00088    ast_assert(forwards->topic_forward == NULL);
00089    ast_assert(forwards->topic_cached_forward == NULL);
00090 }
00091 
00092 static void forwards_unsubscribe(struct app_forwards *forwards)
00093 {
00094    stasis_forward_cancel(forwards->topic_forward);
00095    forwards->topic_forward = NULL;
00096    stasis_forward_cancel(forwards->topic_cached_forward);
00097    forwards->topic_cached_forward = NULL;
00098 }
00099 
00100 static struct app_forwards *forwards_create(struct stasis_app *app,
00101    const char *id)
00102 {
00103    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
00104 
00105    if (!app || ast_strlen_zero(id)) {
00106       return NULL;
00107    }
00108 
00109    forwards = ao2_alloc(sizeof(*forwards) + strlen(id) + 1, forwards_dtor);
00110    if (!forwards) {
00111       return NULL;
00112    }
00113 
00114    strcpy(forwards->id, id);
00115 
00116    ao2_ref(forwards, +1);
00117    return forwards;
00118 }
00119 
00120 /*! Forward a channel's topics to an app */
00121 static struct app_forwards *forwards_create_channel(struct stasis_app *app,
00122    struct ast_channel *chan)
00123 {
00124    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
00125 
00126    if (!app || !chan) {
00127       return NULL;
00128    }
00129 
00130    forwards = forwards_create(app, ast_channel_uniqueid(chan));
00131    if (!forwards) {
00132       return NULL;
00133    }
00134 
00135    forwards->forward_type = FORWARD_CHANNEL;
00136    forwards->topic_forward = stasis_forward_all(ast_channel_topic(chan),
00137       app->topic);
00138    if (!forwards->topic_forward) {
00139       return NULL;
00140    }
00141 
00142    forwards->topic_cached_forward = stasis_forward_all(
00143       ast_channel_topic_cached(chan), app->topic);
00144    if (!forwards->topic_cached_forward) {
00145       /* Half-subscribed is a bad thing */
00146       stasis_forward_cancel(forwards->topic_forward);
00147       forwards->topic_forward = NULL;
00148       return NULL;
00149    }
00150 
00151    ao2_ref(forwards, +1);
00152    return forwards;
00153 }
00154 
00155 /*! Forward a bridge's topics to an app */
00156 static struct app_forwards *forwards_create_bridge(struct stasis_app *app,
00157    struct ast_bridge *bridge)
00158 {
00159    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
00160 
00161    if (!app || !bridge) {
00162       return NULL;
00163    }
00164 
00165    forwards = forwards_create(app, bridge->uniqueid);
00166    if (!forwards) {
00167       return NULL;
00168    }
00169 
00170    forwards->forward_type = FORWARD_BRIDGE;
00171    forwards->topic_forward = stasis_forward_all(ast_bridge_topic(bridge),
00172       app->topic);
00173    if (!forwards->topic_forward) {
00174       return NULL;
00175    }
00176 
00177    forwards->topic_cached_forward = stasis_forward_all(
00178       ast_bridge_topic_cached(bridge), app->topic);
00179    if (!forwards->topic_cached_forward) {
00180       /* Half-subscribed is a bad thing */
00181       stasis_forward_cancel(forwards->topic_forward);
00182       forwards->topic_forward = NULL;
00183       return NULL;
00184    }
00185 
00186    ao2_ref(forwards, +1);
00187    return forwards;
00188 }
00189 
00190 /*! Forward a endpoint's topics to an app */
00191 static struct app_forwards *forwards_create_endpoint(struct stasis_app *app,
00192    struct ast_endpoint *endpoint)
00193 {
00194    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
00195 
00196    if (!app || !endpoint) {
00197       return NULL;
00198    }
00199 
00200    forwards = forwards_create(app, ast_endpoint_get_id(endpoint));
00201    if (!forwards) {
00202       return NULL;
00203    }
00204 
00205    forwards->forward_type = FORWARD_ENDPOINT;
00206    forwards->topic_forward = stasis_forward_all(ast_endpoint_topic(endpoint),
00207       app->topic);
00208    if (!forwards->topic_forward) {
00209       return NULL;
00210    }
00211 
00212    forwards->topic_cached_forward = stasis_forward_all(
00213       ast_endpoint_topic_cached(endpoint), app->topic);
00214    if (!forwards->topic_cached_forward) {
00215       /* Half-subscribed is a bad thing */
00216       stasis_forward_cancel(forwards->topic_forward);
00217       forwards->topic_forward = NULL;
00218       return NULL;
00219    }
00220 
00221    ao2_ref(forwards, +1);
00222    return forwards;
00223 }
00224 
00225 static int forwards_sort(const void *obj_left, const void *obj_right, int flags)
00226 {
00227    const struct app_forwards *object_left = obj_left;
00228    const struct app_forwards *object_right = obj_right;
00229    const char *right_key = obj_right;
00230    int cmp;
00231 
00232    switch (flags & (OBJ_POINTER | OBJ_KEY | OBJ_PARTIAL_KEY)) {
00233    case OBJ_POINTER:
00234       right_key = object_right->id;
00235       /* Fall through */
00236    case OBJ_KEY:
00237       cmp = strcmp(object_left->id, right_key);
00238       break;
00239    case OBJ_PARTIAL_KEY:
00240       /*
00241        * We could also use a partial key struct containing a length
00242        * so strlen() does not get called for every comparison instead.
00243        */
00244       cmp = strncmp(object_left->id, right_key, strlen(right_key));
00245       break;
00246    default:
00247       /* Sort can only work on something with a full or partial key. */
00248       ast_assert(0);
00249       cmp = 0;
00250       break;
00251    }
00252    return cmp;
00253 }
00254 
00255 static void app_dtor(void *obj)
00256 {
00257    struct stasis_app *app = obj;
00258 
00259    ast_verb(1, "Destroying Stasis app %s\n", app->name);
00260 
00261    ast_assert(app->router == NULL);
00262    ast_assert(app->bridge_router == NULL);
00263 
00264    ao2_cleanup(app->topic);
00265    app->topic = NULL;
00266    ao2_cleanup(app->forwards);
00267    app->forwards = NULL;
00268    ao2_cleanup(app->data);
00269    app->data = NULL;
00270 }
00271 
00272 static void call_forwarded_handler(struct stasis_app *app, struct stasis_message *message)
00273 {
00274    struct ast_multi_channel_blob *payload = stasis_message_data(message);
00275    struct ast_channel_snapshot *snapshot = ast_multi_channel_blob_get_channel(payload, "forwarded");
00276    struct ast_channel *chan;
00277 
00278    if (!snapshot) {
00279       return;
00280    }
00281 
00282    chan = ast_channel_get_by_name(snapshot->uniqueid);
00283    if (!chan) {
00284       return;
00285    }
00286 
00287    app_subscribe_channel(app, chan);
00288    ast_channel_unref(chan);
00289 }
00290 
00291 static void sub_default_handler(void *data, struct stasis_subscription *sub,
00292    struct stasis_message *message)
00293 {
00294    struct stasis_app *app = data;
00295    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
00296 
00297    if (stasis_subscription_final_message(sub, message)) {
00298       ao2_cleanup(app);
00299    }
00300 
00301    if (stasis_message_type(message) == ast_channel_dial_type()) {
00302       call_forwarded_handler(app, message);
00303    }
00304 
00305    /* By default, send any message that has a JSON representation */
00306    json = stasis_message_to_json(message, stasis_app_get_sanitizer());
00307    if (!json) {
00308       return;
00309    }
00310 
00311    app_send(app, json);
00312 }
00313 
00314 /*! \brief Typedef for callbacks that get called on channel snapshot updates */
00315 typedef struct ast_json *(*channel_snapshot_monitor)(
00316    struct ast_channel_snapshot *old_snapshot,
00317    struct ast_channel_snapshot *new_snapshot,
00318    const struct timeval *tv);
00319 
00320 static struct ast_json *simple_channel_event(
00321    const char *type,
00322    struct ast_channel_snapshot *snapshot,
00323    const struct timeval *tv)
00324 {
00325    struct ast_json *json_channel = ast_channel_snapshot_to_json(snapshot, stasis_app_get_sanitizer());
00326 
00327    if (!json_channel) {
00328       return NULL;
00329    }
00330 
00331    return ast_json_pack("{s: s, s: o, s: o}",
00332       "type", type,
00333       "timestamp", ast_json_timeval(*tv, NULL),
00334       "channel", json_channel);
00335 }
00336 
00337 static struct ast_json *channel_created_event(
00338    struct ast_channel_snapshot *snapshot,
00339    const struct timeval *tv)
00340 {
00341    return simple_channel_event("ChannelCreated", snapshot, tv);
00342 }
00343 
00344 static struct ast_json *channel_destroyed_event(
00345    struct ast_channel_snapshot *snapshot,
00346    const struct timeval *tv)
00347 {
00348    struct ast_json *json_channel = ast_channel_snapshot_to_json(snapshot, stasis_app_get_sanitizer());
00349 
00350    if (!json_channel) {
00351       return NULL;
00352    }
00353 
00354    return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
00355       "type", "ChannelDestroyed",
00356       "timestamp", ast_json_timeval(*tv, NULL),
00357       "cause", snapshot->hangupcause,
00358       "cause_txt", ast_cause2str(snapshot->hangupcause),
00359       "channel", json_channel);
00360 }
00361 
00362 static struct ast_json *channel_state_change_event(
00363    struct ast_channel_snapshot *snapshot,
00364    const struct timeval *tv)
00365 {
00366    return simple_channel_event("ChannelStateChange", snapshot, tv);
00367 }
00368 
00369 /*! \brief Handle channel state changes */
00370 static struct ast_json *channel_state(
00371    struct ast_channel_snapshot *old_snapshot,
00372    struct ast_channel_snapshot *new_snapshot,
00373    const struct timeval *tv)
00374 {
00375    struct ast_channel_snapshot *snapshot = new_snapshot ?
00376       new_snapshot : old_snapshot;
00377 
00378    if (!old_snapshot) {
00379       return channel_created_event(snapshot, tv);
00380    } else if (!new_snapshot) {
00381       return channel_destroyed_event(snapshot, tv);
00382    } else if (old_snapshot->state != new_snapshot->state) {
00383       return channel_state_change_event(snapshot, tv);
00384    }
00385 
00386    return NULL;
00387 }
00388 
00389 static struct ast_json *channel_dialplan(
00390    struct ast_channel_snapshot *old_snapshot,
00391    struct ast_channel_snapshot *new_snapshot,
00392    const struct timeval *tv)
00393 {
00394    struct ast_json *json_channel;
00395 
00396    /* No Newexten event on cache clear or first event */
00397    if (!old_snapshot || !new_snapshot) {
00398       return NULL;
00399    }
00400 
00401    /* Empty application is not valid for a Newexten event */
00402    if (ast_strlen_zero(new_snapshot->appl)) {
00403       return NULL;
00404    }
00405 
00406    if (ast_channel_snapshot_cep_equal(old_snapshot, new_snapshot)) {
00407       return NULL;
00408    }
00409 
00410    json_channel = ast_channel_snapshot_to_json(new_snapshot, stasis_app_get_sanitizer());
00411    if (!json_channel) {
00412       return NULL;
00413    }
00414 
00415    return ast_json_pack("{s: s, s: o, s: s, s: s, s: o}",
00416       "type", "ChannelDialplan",
00417       "timestamp", ast_json_timeval(*tv, NULL),
00418       "dialplan_app", new_snapshot->appl,
00419       "dialplan_app_data", new_snapshot->data,
00420       "channel", json_channel);
00421 }
00422 
00423 static struct ast_json *channel_callerid(
00424    struct ast_channel_snapshot *old_snapshot,
00425    struct ast_channel_snapshot *new_snapshot,
00426    const struct timeval *tv)
00427 {
00428    struct ast_json *json_channel;
00429 
00430    /* No NewCallerid event on cache clear or first event */
00431    if (!old_snapshot || !new_snapshot) {
00432       return NULL;
00433    }
00434 
00435    if (ast_channel_snapshot_caller_id_equal(old_snapshot, new_snapshot)) {
00436       return NULL;
00437    }
00438 
00439    json_channel = ast_channel_snapshot_to_json(new_snapshot, stasis_app_get_sanitizer());
00440    if (!json_channel) {
00441       return NULL;
00442    }
00443 
00444    return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
00445       "type", "ChannelCallerId",
00446       "timestamp", ast_json_timeval(*tv, NULL),
00447       "caller_presentation", new_snapshot->caller_pres,
00448       "caller_presentation_txt", ast_describe_caller_presentation(
00449          new_snapshot->caller_pres),
00450       "channel", json_channel);
00451 }
00452 
00453 static struct ast_json *channel_connected_line(
00454    struct ast_channel_snapshot *old_snapshot,
00455    struct ast_channel_snapshot *new_snapshot,
00456    const struct timeval *tv)
00457 {
00458    struct ast_json *json_channel;
00459 
00460    /* No ChannelConnectedLine event on cache clear or first event */
00461    if (!old_snapshot || !new_snapshot) {
00462       return NULL;
00463    }
00464 
00465    if (ast_channel_snapshot_connected_line_equal(old_snapshot, new_snapshot)) {
00466       return NULL;
00467    }
00468 
00469    json_channel = ast_channel_snapshot_to_json(new_snapshot, stasis_app_get_sanitizer());
00470    if (!json_channel) {
00471       return NULL;
00472    }
00473 
00474    return ast_json_pack("{s: s, s: o, s: o}",
00475       "type", "ChannelConnectedLine",
00476       "timestamp", ast_json_timeval(*tv, NULL),
00477       "channel", json_channel);
00478 }
00479 
00480 static channel_snapshot_monitor channel_monitors[] = {
00481    channel_state,
00482    channel_dialplan,
00483    channel_callerid,
00484    channel_connected_line,
00485 };
00486 
00487 static void sub_channel_update_handler(void *data,
00488    struct stasis_subscription *sub,
00489    struct stasis_message *message)
00490 {
00491    struct stasis_app *app = data;
00492    struct stasis_cache_update *update;
00493    struct ast_channel_snapshot *new_snapshot;
00494    struct ast_channel_snapshot *old_snapshot;
00495    const struct timeval *tv;
00496    int i;
00497 
00498    ast_assert(stasis_message_type(message) == stasis_cache_update_type());
00499 
00500    update = stasis_message_data(message);
00501 
00502    ast_assert(update->type == ast_channel_snapshot_type());
00503 
00504    new_snapshot = stasis_message_data(update->new_snapshot);
00505    old_snapshot = stasis_message_data(update->old_snapshot);
00506 
00507    /* Pull timestamp from the new snapshot, or from the update message
00508     * when there isn't one. */
00509    tv = update->new_snapshot ?
00510       stasis_message_timestamp(update->new_snapshot) :
00511       stasis_message_timestamp(message);
00512 
00513    for (i = 0; i < ARRAY_LEN(channel_monitors); ++i) {
00514       RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
00515 
00516       msg = channel_monitors[i](old_snapshot, new_snapshot, tv);
00517       if (msg) {
00518          app_send(app, msg);
00519       }
00520    }
00521 
00522    if (!new_snapshot && old_snapshot) {
00523       unsubscribe(app, "channel", old_snapshot->uniqueid, 1);
00524    }
00525 }
00526 
00527 static struct ast_json *simple_endpoint_event(
00528    const char *type,
00529    struct ast_endpoint_snapshot *snapshot,
00530    const struct timeval *tv)
00531 {
00532    struct ast_json *json_endpoint = ast_endpoint_snapshot_to_json(snapshot, stasis_app_get_sanitizer());
00533 
00534    if (!json_endpoint) {
00535       return NULL;
00536    }
00537 
00538    return ast_json_pack("{s: s, s: o, s: o}",
00539       "type", type,
00540       "timestamp", ast_json_timeval(*tv, NULL),
00541       "endpoint", json_endpoint);
00542 }
00543 
00544 static int message_received_handler(const char *endpoint_id, struct ast_json *json_msg, void *pvt)
00545 {
00546    RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
00547    struct ast_json *json_endpoint;
00548    struct stasis_app *app = pvt;
00549    char *tech;
00550    char *resource;
00551 
00552    tech = ast_strdupa(endpoint_id);
00553    resource = strchr(tech, '/');
00554    if (resource) {
00555       resource[0] = '\0';
00556       resource++;
00557    }
00558 
00559    if (ast_strlen_zero(tech) || ast_strlen_zero(resource)) {
00560       return -1;
00561    }
00562 
00563    snapshot = ast_endpoint_latest_snapshot(tech, resource);
00564    if (!snapshot) {
00565       return -1;
00566    }
00567 
00568    json_endpoint = ast_endpoint_snapshot_to_json(snapshot, stasis_app_get_sanitizer());
00569    if (!json_endpoint) {
00570       return -1;
00571    }
00572 
00573    app_send(app, ast_json_pack("{s: s, s: o, s: o, s: O}",
00574       "type", "TextMessageReceived",
00575       "timestamp", ast_json_timeval(ast_tvnow(), NULL),
00576       "endpoint", json_endpoint,
00577       "message", json_msg));
00578 
00579    return 0;
00580 }
00581 
00582 static void sub_endpoint_update_handler(void *data,
00583    struct stasis_subscription *sub,
00584    struct stasis_message *message)
00585 {
00586    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
00587    struct stasis_app *app = data;
00588    struct stasis_cache_update *update;
00589    struct ast_endpoint_snapshot *new_snapshot;
00590    struct ast_endpoint_snapshot *old_snapshot;
00591    const struct timeval *tv;
00592 
00593    ast_assert(stasis_message_type(message) == stasis_cache_update_type());
00594 
00595    update = stasis_message_data(message);
00596 
00597    ast_assert(update->type == ast_endpoint_snapshot_type());
00598 
00599    new_snapshot = stasis_message_data(update->new_snapshot);
00600    old_snapshot = stasis_message_data(update->old_snapshot);
00601 
00602    if (new_snapshot) {
00603       tv = stasis_message_timestamp(update->new_snapshot);
00604 
00605       json = simple_endpoint_event("EndpointStateChange", new_snapshot, tv);
00606       if (!json) {
00607          return;
00608       }
00609 
00610       app_send(app, json);
00611    }
00612 
00613    if (!new_snapshot && old_snapshot) {
00614       unsubscribe(app, "endpoint", old_snapshot->id, 1);
00615    }
00616 }
00617 
00618 static struct ast_json *simple_bridge_event(
00619    const char *type,
00620    struct ast_bridge_snapshot *snapshot,
00621    const struct timeval *tv)
00622 {
00623    struct ast_json *json_bridge = ast_bridge_snapshot_to_json(snapshot, stasis_app_get_sanitizer());
00624    if (!json_bridge) {
00625       return NULL;
00626    }
00627 
00628    return ast_json_pack("{s: s, s: o, s: o}",
00629       "type", type,
00630       "timestamp", ast_json_timeval(*tv, NULL),
00631       "bridge", json_bridge);
00632 }
00633 
00634 static void sub_bridge_update_handler(void *data,
00635    struct stasis_subscription *sub,
00636    struct stasis_message *message)
00637 {
00638    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
00639    struct stasis_app *app = data;
00640    struct stasis_cache_update *update;
00641    struct ast_bridge_snapshot *new_snapshot;
00642    struct ast_bridge_snapshot *old_snapshot;
00643    const struct timeval *tv;
00644 
00645    ast_assert(stasis_message_type(message) == stasis_cache_update_type());
00646 
00647    update = stasis_message_data(message);
00648 
00649    ast_assert(update->type == ast_bridge_snapshot_type());
00650 
00651    new_snapshot = stasis_message_data(update->new_snapshot);
00652    old_snapshot = stasis_message_data(update->old_snapshot);
00653    tv = update->new_snapshot ?
00654       stasis_message_timestamp(update->new_snapshot) :
00655       stasis_message_timestamp(message);
00656 
00657    if (!new_snapshot) {
00658       json = simple_bridge_event("BridgeDestroyed", old_snapshot, tv);
00659    } else if (!old_snapshot) {
00660       json = simple_bridge_event("BridgeCreated", new_snapshot, tv);
00661    }
00662 
00663    if (json) {
00664       app_send(app, json);
00665    }
00666 
00667    if (!new_snapshot && old_snapshot) {
00668       unsubscribe(app, "bridge", old_snapshot->uniqueid, 1);
00669    }
00670 }
00671 
00672 
00673 /*! \brief Helper function for determining if the application is subscribed to a given entity */
00674 static int bridge_app_subscribed(struct stasis_app *app, const char *uniqueid)
00675 {
00676    struct app_forwards *forwards = NULL;
00677 
00678    forwards = ao2_find(app->forwards, uniqueid, OBJ_SEARCH_KEY);
00679    if (!forwards) {
00680       return 0;
00681    }
00682 
00683    ao2_ref(forwards, -1);
00684    return 1;
00685 }
00686 
00687 static void bridge_merge_handler(void *data, struct stasis_subscription *sub,
00688    struct stasis_message *message)
00689 {
00690    struct stasis_app *app = data;
00691    struct ast_bridge_merge_message *merge;
00692 
00693    merge = stasis_message_data(message);
00694 
00695    /* Find out if we're subscribed to either bridge */
00696    if (bridge_app_subscribed(app, merge->from->uniqueid) ||
00697       bridge_app_subscribed(app, merge->to->uniqueid)) {
00698       /* Forward the message to the app */
00699       stasis_publish(app->topic, message);
00700    }
00701 }
00702 
00703 /*! \brief Callback function for checking if channels in a bridge are subscribed to */
00704 static int bridge_app_subscribed_involved(struct stasis_app *app, struct ast_bridge_snapshot *snapshot)
00705 {
00706    int subscribed = 0;
00707    struct ao2_iterator iter;
00708    char *uniqueid;
00709 
00710    if (bridge_app_subscribed(app, snapshot->uniqueid)) {
00711       return 1;
00712    }
00713 
00714    iter = ao2_iterator_init(snapshot->channels, 0);
00715    for (; (uniqueid = ao2_iterator_next(&iter)); ao2_ref(uniqueid, -1)) {
00716       if (bridge_app_subscribed(app, uniqueid)) {
00717          subscribed = 1;
00718          ao2_ref(uniqueid, -1);
00719          break;
00720       }
00721    }
00722    ao2_iterator_destroy(&iter);
00723 
00724    return subscribed;
00725 }
00726 
00727 static void bridge_blind_transfer_handler(void *data, struct stasis_subscription *sub,
00728    struct stasis_message *message)
00729 {
00730    struct stasis_app *app = data;
00731    struct ast_blind_transfer_message *transfer_msg = stasis_message_data(message);
00732    struct ast_bridge_snapshot *bridge = transfer_msg->bridge;
00733 
00734    if (bridge_app_subscribed(app, transfer_msg->transferer->uniqueid) ||
00735       (bridge && bridge_app_subscribed_involved(app, bridge))) {
00736       stasis_publish(app->topic, message);
00737    }
00738 }
00739 
00740 static void bridge_attended_transfer_handler(void *data, struct stasis_subscription *sub,
00741    struct stasis_message *message)
00742 {
00743    struct stasis_app *app = data;
00744    struct ast_attended_transfer_message *transfer_msg = stasis_message_data(message);
00745    int subscribed = 0;
00746 
00747    subscribed = bridge_app_subscribed(app, transfer_msg->to_transferee.channel_snapshot->uniqueid);
00748    if (!subscribed) {
00749       subscribed = bridge_app_subscribed(app, transfer_msg->to_transfer_target.channel_snapshot->uniqueid);
00750    }
00751    if (!subscribed && transfer_msg->to_transferee.bridge_snapshot) {
00752       subscribed = bridge_app_subscribed_involved(app, transfer_msg->to_transferee.bridge_snapshot);
00753    }
00754    if (!subscribed && transfer_msg->to_transfer_target.bridge_snapshot) {
00755       subscribed = bridge_app_subscribed_involved(app, transfer_msg->to_transfer_target.bridge_snapshot);
00756    }
00757 
00758    if (!subscribed) {
00759       switch (transfer_msg->dest_type) {
00760       case AST_ATTENDED_TRANSFER_DEST_BRIDGE_MERGE:
00761          subscribed = bridge_app_subscribed(app, transfer_msg->dest.bridge);
00762          break;
00763       case AST_ATTENDED_TRANSFER_DEST_LINK:
00764          subscribed = bridge_app_subscribed(app, transfer_msg->dest.links[0]->uniqueid);
00765          if (!subscribed) {
00766             subscribed = bridge_app_subscribed(app, transfer_msg->dest.links[1]->uniqueid);
00767          }
00768          break;
00769       break;
00770       case AST_ATTENDED_TRANSFER_DEST_THREEWAY:
00771          subscribed = bridge_app_subscribed_involved(app, transfer_msg->dest.threeway.bridge_snapshot);
00772          if (!subscribed) {
00773             subscribed = bridge_app_subscribed(app, transfer_msg->dest.threeway.channel_snapshot->uniqueid);
00774          }
00775          break;
00776       default:
00777          break;
00778       }
00779    }
00780 
00781    if (subscribed) {
00782       stasis_publish(app->topic, message);
00783    }
00784 }
00785 
00786 static void bridge_default_handler(void *data, struct stasis_subscription *sub,
00787    struct stasis_message *message)
00788 {
00789    struct stasis_app *app = data;
00790 
00791    if (stasis_subscription_final_message(sub, message)) {
00792       ao2_cleanup(app);
00793    }
00794 }
00795 
00796 struct stasis_app *app_create(const char *name, stasis_app_cb handler, void *data)
00797 {
00798    RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
00799    size_t size;
00800    int res = 0;
00801 
00802    ast_assert(name != NULL);
00803    ast_assert(handler != NULL);
00804 
00805    ast_verb(1, "Creating Stasis app '%s'\n", name);
00806 
00807    size = sizeof(*app) + strlen(name) + 1;
00808    app = ao2_alloc_options(size, app_dtor, AO2_ALLOC_OPT_LOCK_MUTEX);
00809 
00810    if (!app) {
00811       return NULL;
00812    }
00813 
00814    app->forwards = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_MUTEX,
00815       AO2_CONTAINER_ALLOC_OPT_DUPS_OBJ_REJECT,
00816       forwards_sort, NULL);
00817    if (!app->forwards) {
00818       return NULL;
00819    }
00820 
00821    app->topic = stasis_topic_create(name);
00822    if (!app->topic) {
00823       return NULL;
00824    }
00825 
00826    app->bridge_router = stasis_message_router_create(ast_bridge_topic_all());
00827    if (!app->bridge_router) {
00828       return NULL;
00829    }
00830 
00831    res |= stasis_message_router_add(app->bridge_router,
00832       ast_bridge_merge_message_type(), bridge_merge_handler, app);
00833 
00834    res |= stasis_message_router_add(app->bridge_router,
00835       ast_blind_transfer_type(), bridge_blind_transfer_handler, app);
00836 
00837    res |= stasis_message_router_add(app->bridge_router,
00838       ast_attended_transfer_type(), bridge_attended_transfer_handler, app);
00839 
00840    res |= stasis_message_router_set_default(app->bridge_router,
00841       bridge_default_handler, app);
00842 
00843    if (res != 0) {
00844       return NULL;
00845    }
00846    /* Bridge router holds a reference */
00847    ao2_ref(app, +1);
00848 
00849    app->router = stasis_message_router_create(app->topic);
00850    if (!app->router) {
00851       return NULL;
00852    }
00853 
00854    res |= stasis_message_router_add_cache_update(app->router,
00855       ast_bridge_snapshot_type(), sub_bridge_update_handler, app);
00856 
00857    res |= stasis_message_router_add_cache_update(app->router,
00858       ast_channel_snapshot_type(), sub_channel_update_handler, app);
00859 
00860    res |= stasis_message_router_add_cache_update(app->router,
00861       ast_endpoint_snapshot_type(), sub_endpoint_update_handler, app);
00862 
00863    res |= stasis_message_router_set_default(app->router,
00864       sub_default_handler, app);
00865 
00866    if (res != 0) {
00867       return NULL;
00868    }
00869    /* Router holds a reference */
00870    ao2_ref(app, +1);
00871 
00872    strncpy(app->name, name, size - sizeof(*app));
00873    app->handler = handler;
00874    ao2_ref(data, +1);
00875    app->data = data;
00876 
00877    ao2_ref(app, +1);
00878    return app;
00879 }
00880 
00881 struct stasis_topic *ast_app_get_topic(struct stasis_app *app) {
00882    return app->topic;
00883 }
00884 
00885 /*!
00886  * \brief Send a message to the given application.
00887  * \param app App to send the message to.
00888  * \param message Message to send.
00889  */
00890 void app_send(struct stasis_app *app, struct ast_json *message)
00891 {
00892    stasis_app_cb handler;
00893    RAII_VAR(void *, data, NULL, ao2_cleanup);
00894 
00895    /* Copy off mutable state with lock held */
00896    {
00897       SCOPED_AO2LOCK(lock, app);
00898       handler = app->handler;
00899       if (app->data) {
00900          ao2_ref(app->data, +1);
00901          data = app->data;
00902       }
00903       /* Name is immutable; no need to copy */
00904    }
00905 
00906    if (!handler) {
00907       ast_verb(3,
00908          "Inactive Stasis app '%s' missed message\n", app->name);
00909       return;
00910    }
00911 
00912    handler(data, app->name, message);
00913 }
00914 
00915 void app_deactivate(struct stasis_app *app)
00916 {
00917    SCOPED_AO2LOCK(lock, app);
00918    ast_verb(1, "Deactivating Stasis app '%s'\n", app->name);
00919    app->handler = NULL;
00920    ao2_cleanup(app->data);
00921    app->data = NULL;
00922 }
00923 
00924 void app_shutdown(struct stasis_app *app)
00925 {
00926    SCOPED_AO2LOCK(lock, app);
00927 
00928    ast_assert(app_is_finished(app));
00929 
00930    stasis_message_router_unsubscribe(app->router);
00931    app->router = NULL;
00932    stasis_message_router_unsubscribe(app->bridge_router);
00933    app->bridge_router = NULL;
00934 }
00935 
00936 int app_is_active(struct stasis_app *app)
00937 {
00938    SCOPED_AO2LOCK(lock, app);
00939    return app->handler != NULL;
00940 }
00941 
00942 int app_is_finished(struct stasis_app *app)
00943 {
00944    SCOPED_AO2LOCK(lock, app);
00945 
00946    return app->handler == NULL && ao2_container_count(app->forwards) == 0;
00947 }
00948 
00949 void app_update(struct stasis_app *app, stasis_app_cb handler, void *data)
00950 {
00951    SCOPED_AO2LOCK(lock, app);
00952 
00953    if (app->handler) {
00954       RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
00955 
00956       ast_verb(1, "Replacing Stasis app '%s'\n", app->name);
00957 
00958       msg = ast_json_pack("{s: s, s: s}",
00959          "type", "ApplicationReplaced",
00960          "application", app->name);
00961       if (msg) {
00962          app_send(app, msg);
00963       }
00964    } else {
00965       ast_verb(1, "Activating Stasis app '%s'\n", app->name);
00966    }
00967 
00968    app->handler = handler;
00969    ao2_cleanup(app->data);
00970    if (data) {
00971       ao2_ref(data, +1);
00972    }
00973    app->data = data;
00974 }
00975 
00976 const char *app_name(const struct stasis_app *app)
00977 {
00978    return app->name;
00979 }
00980 
00981 struct ast_json *app_to_json(const struct stasis_app *app)
00982 {
00983    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
00984    struct ast_json *channels;
00985    struct ast_json *bridges;
00986    struct ast_json *endpoints;
00987    struct ao2_iterator i;
00988    void *obj;
00989 
00990    json = ast_json_pack("{s: s, s: [], s: [], s: []}",
00991       "name", app->name,
00992       "channel_ids", "bridge_ids", "endpoint_ids");
00993    channels = ast_json_object_get(json, "channel_ids");
00994    bridges = ast_json_object_get(json, "bridge_ids");
00995    endpoints = ast_json_object_get(json, "endpoint_ids");
00996 
00997    i = ao2_iterator_init(app->forwards, 0);
00998    while ((obj = ao2_iterator_next(&i))) {
00999       RAII_VAR(struct app_forwards *, forwards, obj, ao2_cleanup);
01000       RAII_VAR(struct ast_json *, id, NULL, ast_json_unref);
01001       int append_res = -1;
01002 
01003       id = ast_json_string_create(forwards->id);
01004 
01005       switch (forwards->forward_type) {
01006       case FORWARD_CHANNEL:
01007          append_res = ast_json_array_append(channels,
01008             ast_json_ref(id));
01009          break;
01010       case FORWARD_BRIDGE:
01011          append_res = ast_json_array_append(bridges,
01012             ast_json_ref(id));
01013          break;
01014       case FORWARD_ENDPOINT:
01015          append_res = ast_json_array_append(endpoints,
01016             ast_json_ref(id));
01017          break;
01018       }
01019 
01020       if (append_res != 0) {
01021          ast_log(LOG_ERROR, "Error building response\n");
01022          ao2_iterator_destroy(&i);
01023          return NULL;
01024       }
01025    }
01026    ao2_iterator_destroy(&i);
01027 
01028    return ast_json_ref(json);
01029 }
01030 
01031 int app_subscribe_channel(struct stasis_app *app, struct ast_channel *chan)
01032 {
01033    int res;
01034 
01035    if (!app || !chan) {
01036       return -1;
01037    } else {
01038       RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01039       SCOPED_AO2LOCK(lock, app->forwards);
01040 
01041       forwards = ao2_find(app->forwards, ast_channel_uniqueid(chan),
01042          OBJ_SEARCH_KEY | OBJ_NOLOCK);
01043       if (!forwards) {
01044          /* Forwards not found, create one */
01045          forwards = forwards_create_channel(app, chan);
01046          if (!forwards) {
01047             return -1;
01048          }
01049 
01050          res = ao2_link_flags(app->forwards, forwards,
01051             OBJ_NOLOCK);
01052          if (!res) {
01053             return -1;
01054          }
01055       }
01056 
01057       ++forwards->interested;
01058       ast_debug(3, "Channel '%s' is %d interested in %s\n", ast_channel_uniqueid(chan), forwards->interested, app->name);
01059       return 0;
01060    }
01061 }
01062 
01063 static int subscribe_channel(struct stasis_app *app, void *obj)
01064 {
01065    return app_subscribe_channel(app, obj);
01066 }
01067 
01068 static int unsubscribe(struct stasis_app *app, const char *kind, const char *id, int terminate)
01069 {
01070    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01071    SCOPED_AO2LOCK(lock, app->forwards);
01072 
01073    forwards = ao2_find(app->forwards, id, OBJ_SEARCH_KEY | OBJ_NOLOCK);
01074    if (!forwards) {
01075       ast_debug(3, "App '%s' not subscribed to %s '%s'\n", app->name, kind, id);
01076       return -1;
01077    }
01078    forwards->interested--;
01079 
01080    ast_debug(3, "%s '%s': is %d interested in %s\n", kind, id, forwards->interested, app->name);
01081    if (forwards->interested == 0 || terminate) {
01082       /* No one is interested any more; unsubscribe */
01083       ast_debug(3, "%s '%s' unsubscribed from %s\n", kind, id, app->name);
01084       forwards_unsubscribe(forwards);
01085       ao2_find(app->forwards, forwards,
01086          OBJ_POINTER | OBJ_NOLOCK | OBJ_UNLINK |
01087          OBJ_NODATA);
01088 
01089       if (!strcmp(kind, "endpoint")) {
01090          messaging_app_unsubscribe_endpoint(app->name, id);
01091       }
01092    }
01093 
01094    return 0;
01095 }
01096 
01097 int app_unsubscribe_channel(struct stasis_app *app, struct ast_channel *chan)
01098 {
01099    if (!app || !chan) {
01100       return -1;
01101    }
01102 
01103    return app_unsubscribe_channel_id(app, ast_channel_uniqueid(chan));
01104 }
01105 
01106 int app_unsubscribe_channel_id(struct stasis_app *app, const char *channel_id)
01107 {
01108    if (!app || !channel_id) {
01109       return -1;
01110    }
01111 
01112    return unsubscribe(app, "channel", channel_id, 0);
01113 }
01114 
01115 int app_is_subscribed_channel_id(struct stasis_app *app, const char *channel_id)
01116 {
01117    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01118    forwards = ao2_find(app->forwards, channel_id, OBJ_SEARCH_KEY);
01119    return forwards != NULL;
01120 }
01121 
01122 static void *channel_find(const struct stasis_app *app, const char *id)
01123 {
01124    return ast_channel_get_by_name(id);
01125 }
01126 
01127 struct stasis_app_event_source channel_event_source = {
01128    .scheme = "channel:",
01129    .find = channel_find,
01130    .subscribe = subscribe_channel,
01131    .unsubscribe = app_unsubscribe_channel_id,
01132    .is_subscribed = app_is_subscribed_channel_id
01133 };
01134 
01135 int app_subscribe_bridge(struct stasis_app *app, struct ast_bridge *bridge)
01136 {
01137    if (!app || !bridge) {
01138       return -1;
01139    } else {
01140       RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01141       SCOPED_AO2LOCK(lock, app->forwards);
01142 
01143       forwards = ao2_find(app->forwards, bridge->uniqueid,
01144          OBJ_SEARCH_KEY | OBJ_NOLOCK);
01145 
01146       if (!forwards) {
01147          /* Forwards not found, create one */
01148          forwards = forwards_create_bridge(app, bridge);
01149          if (!forwards) {
01150             return -1;
01151          }
01152          ao2_link_flags(app->forwards, forwards, OBJ_NOLOCK);
01153       }
01154 
01155       ++forwards->interested;
01156       ast_debug(3, "Bridge '%s' is %d interested in %s\n", bridge->uniqueid, forwards->interested, app->name);
01157       return 0;
01158    }
01159 }
01160 
01161 static int subscribe_bridge(struct stasis_app *app, void *obj)
01162 {
01163    return app_subscribe_bridge(app, obj);
01164 }
01165 
01166 int app_unsubscribe_bridge(struct stasis_app *app, struct ast_bridge *bridge)
01167 {
01168    if (!app || !bridge) {
01169       return -1;
01170    }
01171 
01172    return app_unsubscribe_bridge_id(app, bridge->uniqueid);
01173 }
01174 
01175 int app_unsubscribe_bridge_id(struct stasis_app *app, const char *bridge_id)
01176 {
01177    if (!app || !bridge_id) {
01178       return -1;
01179    }
01180 
01181    return unsubscribe(app, "bridge", bridge_id, 0);
01182 }
01183 
01184 int app_is_subscribed_bridge_id(struct stasis_app *app, const char *bridge_id)
01185 {
01186    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01187    forwards = ao2_find(app->forwards, bridge_id, OBJ_SEARCH_KEY);
01188    return forwards != NULL;
01189 }
01190 
01191 static void *bridge_find(const struct stasis_app *app, const char *id)
01192 {
01193    return stasis_app_bridge_find_by_id(id);
01194 }
01195 
01196 struct stasis_app_event_source bridge_event_source = {
01197    .scheme = "bridge:",
01198    .find = bridge_find,
01199    .subscribe = subscribe_bridge,
01200    .unsubscribe = app_unsubscribe_bridge_id,
01201    .is_subscribed = app_is_subscribed_bridge_id
01202 };
01203 
01204 int app_subscribe_endpoint(struct stasis_app *app, struct ast_endpoint *endpoint)
01205 {
01206    if (!app || !endpoint) {
01207       return -1;
01208    } else {
01209       RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01210       SCOPED_AO2LOCK(lock, app->forwards);
01211 
01212       forwards = ao2_find(app->forwards, ast_endpoint_get_id(endpoint),
01213          OBJ_SEARCH_KEY | OBJ_NOLOCK);
01214 
01215       if (!forwards) {
01216          /* Forwards not found, create one */
01217          forwards = forwards_create_endpoint(app, endpoint);
01218          if (!forwards) {
01219             return -1;
01220          }
01221          ao2_link_flags(app->forwards, forwards, OBJ_NOLOCK);
01222 
01223          /* Subscribe for messages */
01224          messaging_app_subscribe_endpoint(app->name, endpoint, &message_received_handler, app);
01225       }
01226 
01227       ++forwards->interested;
01228       ast_debug(3, "Endpoint '%s' is %d interested in %s\n", ast_endpoint_get_id(endpoint), forwards->interested, app->name);
01229       return 0;
01230    }
01231 }
01232 
01233 static int subscribe_endpoint(struct stasis_app *app, void *obj)
01234 {
01235    return app_subscribe_endpoint(app, obj);
01236 }
01237 
01238 int app_unsubscribe_endpoint_id(struct stasis_app *app, const char *endpoint_id)
01239 {
01240    if (!app || !endpoint_id) {
01241       return -1;
01242    }
01243 
01244    return unsubscribe(app, "endpoint", endpoint_id, 0);
01245 }
01246 
01247 int app_is_subscribed_endpoint_id(struct stasis_app *app, const char *endpoint_id)
01248 {
01249    RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
01250    forwards = ao2_find(app->forwards, endpoint_id, OBJ_SEARCH_KEY);
01251    return forwards != NULL;
01252 }
01253 
01254 static void *endpoint_find(const struct stasis_app *app, const char *id)
01255 {
01256    return ast_endpoint_find_by_id(id);
01257 }
01258 
01259 struct stasis_app_event_source endpoint_event_source = {
01260    .scheme = "endpoint:",
01261    .find = endpoint_find,
01262    .subscribe = subscribe_endpoint,
01263    .unsubscribe = app_unsubscribe_endpoint_id,
01264    .is_subscribed = app_is_subscribed_endpoint_id
01265 };
01266 
01267 void stasis_app_register_event_sources(void)
01268 {
01269    stasis_app_register_event_source(&channel_event_source);
01270    stasis_app_register_event_source(&bridge_event_source);
01271    stasis_app_register_event_source(&endpoint_event_source);
01272 }
01273 
01274 int stasis_app_is_core_event_source(struct stasis_app_event_source *obj)
01275 {
01276    return obj == &endpoint_event_source ||
01277       obj == &bridge_event_source ||
01278       obj == &channel_event_source;
01279 }
01280 
01281 void stasis_app_unregister_event_sources(void)
01282 {
01283    stasis_app_unregister_event_source(&endpoint_event_source);
01284    stasis_app_unregister_event_source(&bridge_event_source);
01285    stasis_app_unregister_event_source(&channel_event_source);
01286 }
01287 
01288 

Generated on Thu Apr 16 06:27:08 2015 for Asterisk - The Open Source Telephony Project by  doxygen 1.5.6