res/res_stasis.c

   1 /*
   2  * Asterisk -- An open source telephony toolkit.
   3  *
   4  * Copyright (C) 2012 - 2013, Digium, Inc.
   5  *
   6  * David M. Lee, II <dlee@digium.com>
   7  *
   8  * See http://www.asterisk.org for more information about
   9  * the Asterisk project. Please do not directly contact
  10  * any of the maintainers of this project for assistance;
  11  * the project provides a web site, mailing lists and IRC
  12  * channels for your use.
  13  *
  14  * This program is free software, distributed under the terms of
  15  * the GNU General Public License Version 2. See the LICENSE file
  16  * at the top of the source tree.
  17  */
  18
  19 /*! \file
  20  *
  21  * \brief Stasis application support.
  22  *
  23  * \author David M. Lee, II <dlee@digium.com>
  24  *
  25  * <code>res_stasis.so</code> brings together the various components of the
  26  * Stasis application infrastructure.
  27  *
  28  * First, there's the Stasis application handler, stasis_app_exec(). This is
  29  * called by <code>app_stasis.so</code> to give control of a channel to the
  30  * Stasis application code from the dialplan.
  31  *
  32  * While a channel is in stasis_app_exec(), it has a \ref stasis_app_control
  33  * object, which may be used to control the channel.
  34  *
  35  * To control the channel, commands may be sent to channel using
  36  * stasis_app_send_command() and stasis_app_send_async_command().
  37  *
  38  * Alongside this, applications may be registered/unregistered using
  39  * stasis_app_register()/stasis_app_unregister(). While a channel is in Stasis,
  40  * events received on the channel's topic are converted to JSON and forwarded to
  41  * the \ref stasis_app_cb. The application may also subscribe to the channel to
  42  * continue to receive messages even after the channel has left Stasis, but it
  43  * will not be able to control it.
  44  *
  45  * Given all the stuff that comes together in this module, it's been broken up
  46  * into several pieces that are in <code>res/stasis/</code> and compiled into
  47  * <code>res_stasis.so</code>.
  48  */
  49
  50 /*** MODULEINFO
  51         <support_level>core</support_level>
  52  ***/
  53
  54 #include "asterisk.h"
  55
  56 ASTERISK_REGISTER_FILE()
  57
  58 #include "asterisk/astobj2.h"
  59 #include "asterisk/callerid.h"
  60 #include "asterisk/module.h"
  61 #include "asterisk/stasis_app_impl.h"
  62 #include "asterisk/stasis_channels.h"
  63 #include "asterisk/stasis_bridges.h"
  64 #include "asterisk/stasis_endpoints.h"
  65 #include "asterisk/stasis_message_router.h"
  66 #include "asterisk/strings.h"
  67 #include "stasis/app.h"
  68 #include "stasis/control.h"
  69 #include "stasis/messaging.h"
  70 #include "stasis/stasis_bridge.h"
  71 #include "asterisk/core_unreal.h"
  72 #include "asterisk/musiconhold.h"
  73 #include "asterisk/causes.h"
  74 #include "asterisk/stringfields.h"
  75 #include "asterisk/bridge_after.h"
  76 #include "asterisk/format_cache.h"
  77
  78 /*! Time to wait for a frame in the application */
  79 #define MAX_WAIT_MS 200
  80
  81 /*!
  82  * \brief Number of buckets for the Stasis application hash table.  Remember to
  83  * keep it a prime number!
  84  */
  85 #define APPS_NUM_BUCKETS 127
  86
  87 /*!
  88  * \brief Number of buckets for the Stasis application hash table.  Remember to
  89  * keep it a prime number!
  90  */
  91 #define CONTROLS_NUM_BUCKETS 127
  92
  93 /*!
  94  * \brief Number of buckets for the Stasis bridges hash table.  Remember to
  95  * keep it a prime number!
  96  */
  97 #define BRIDGES_NUM_BUCKETS 127
  98
  99 /*!
 100  * \brief Stasis application container.
 101  */
 102 struct ao2_container *apps_registry;
 103
 104 struct ao2_container *app_controls;
 105
 106 struct ao2_container *app_bridges;
 107
 108 struct ao2_container *app_bridges_moh;
 109
 110 struct ao2_container *app_bridges_playback;
 111
 112 /*!
 113  * \internal \brief List of registered event sources.
 114  */
 115 AST_RWLIST_HEAD_STATIC(event_sources, stasis_app_event_source);
 116
 117 static struct ast_json *stasis_end_to_json(struct stasis_message *message,
 118                 const struct stasis_message_sanitizer *sanitize)
 119 {
 120         struct ast_channel_blob *payload = stasis_message_data(message);
 121
 122         if (sanitize && sanitize->channel_snapshot &&
 123                         sanitize->channel_snapshot(payload->snapshot)) {
 124                 return NULL;
 125         }
 126
 127         return ast_json_pack("{s: s, s: o, s: o}",
 128                 "type", "StasisEnd",
 129                 "timestamp", ast_json_timeval(ast_tvnow(), NULL),
 130                 "channel", ast_channel_snapshot_to_json(payload->snapshot, sanitize));
 131 }
 132
 133 STASIS_MESSAGE_TYPE_DEFN_LOCAL(end_message_type,
 134         .to_json = stasis_end_to_json);
 135
 136 struct start_message_blob {
 137         struct ast_channel_snapshot *channel;           /*!< Channel that is entering Stasis() */
 138         struct ast_channel_snapshot *replace_channel;   /*!< Channel that is being replaced (optional) */
 139         struct ast_json *blob;                          /*!< JSON blob containing timestamp and args */
 140 };
 141
 142 static struct ast_json *stasis_start_to_json(struct stasis_message *message,
 143                 const struct stasis_message_sanitizer *sanitize)
 144 {
 145         struct start_message_blob *payload = stasis_message_data(message);
 146         struct ast_json *msg;
 147
 148         if (sanitize && sanitize->channel_snapshot &&
 149                         sanitize->channel_snapshot(payload->channel)) {
 150                 return NULL;
 151         }
 152
 153         msg = ast_json_pack("{s: s, s: o, s: o, s: o}",
 154                 "type", "StasisStart",
 155                 "timestamp", ast_json_copy(ast_json_object_get(payload->blob, "timestamp")),
 156                 "args", ast_json_deep_copy(ast_json_object_get(payload->blob, "args")),
 157                 "channel", ast_channel_snapshot_to_json(payload->channel, NULL));
 158         if (!msg) {
 159                 ast_log(LOG_ERROR, "Failed to pack JSON for StasisStart message\n");
 160                 return NULL;
 161         }
 162
 163         if (payload->replace_channel) {
 164                 int res = ast_json_object_set(msg, "replace_channel",
 165                         ast_channel_snapshot_to_json(payload->replace_channel, NULL));
 166
 167                 if (res) {
 168                         ast_json_unref(msg);
 169                         ast_log(LOG_ERROR, "Failed to append JSON for StasisStart message\n");
 170                         return NULL;
 171                 }
 172         }
 173
 174         return msg;
 175 }
 176
 177 STASIS_MESSAGE_TYPE_DEFN_LOCAL(start_message_type,
 178         .to_json = stasis_start_to_json);
 179
 180 const char *stasis_app_name(const struct stasis_app *app)
 181 {
 182         return app_name(app);
 183 }
 184
 185 /*! AO2 hash function for \ref app */
 186 static int app_hash(const void *obj, const int flags)
 187 {
 188         const struct stasis_app *app;
 189         const char *key;
 190
 191         switch (flags & OBJ_SEARCH_MASK) {
 192         case OBJ_SEARCH_KEY:
 193                 key = obj;
 194                 break;
 195         case OBJ_SEARCH_OBJECT:
 196                 app = obj;
 197                 key = stasis_app_name(app);
 198                 break;
 199         default:
 200                 /* Hash can only work on something with a full key. */
 201                 ast_assert(0);
 202                 return 0;
 203         }
 204         return ast_str_hash(key);
 205 }
 206
 207 /*! AO2 comparison function for \ref app */
 208 static int app_compare(void *obj, void *arg, int flags)
 209 {
 210         const struct stasis_app *object_left = obj;
 211         const struct stasis_app *object_right = arg;
 212         const char *right_key = arg;
 213         int cmp;
 214
 215         switch (flags & OBJ_SEARCH_MASK) {
 216         case OBJ_SEARCH_OBJECT:
 217                 right_key = stasis_app_name(object_right);
 218                 /* Fall through */
 219         case OBJ_SEARCH_KEY:
 220                 cmp = strcmp(stasis_app_name(object_left), right_key);
 221                 break;
 222         case OBJ_SEARCH_PARTIAL_KEY:
 223                 /*
 224                  * We could also use a partial key struct containing a length
 225                  * so strlen() does not get called for every comparison instead.
 226                  */
 227                 cmp = strncmp(stasis_app_name(object_left), right_key, strlen(right_key));
 228                 break;
 229         default:
 230                 /*
 231                  * What arg points to is specific to this traversal callback
 232                  * and has no special meaning to astobj2.
 233                  */
 234                 cmp = 0;
 235                 break;
 236         }
 237         if (cmp) {
 238                 return 0;
 239         }
 240         /*
 241          * At this point the traversal callback is identical to a sorted
 242          * container.
 243          */
 244         return CMP_MATCH;
 245 }
 246
 247 /*! AO2 hash function for \ref stasis_app_control */
 248 static int control_hash(const void *obj, const int flags)
 249 {
 250         const struct stasis_app_control *control;
 251         const char *key;
 252
 253         switch (flags & OBJ_SEARCH_MASK) {
 254         case OBJ_SEARCH_KEY:
 255                 key = obj;
 256                 break;
 257         case OBJ_SEARCH_OBJECT:
 258                 control = obj;
 259                 key = stasis_app_control_get_channel_id(control);
 260                 break;
 261         default:
 262                 /* Hash can only work on something with a full key. */
 263                 ast_assert(0);
 264                 return 0;
 265         }
 266         return ast_str_hash(key);
 267 }
 268
 269 /*! AO2 comparison function for \ref stasis_app_control */
 270 static int control_compare(void *obj, void *arg, int flags)
 271 {
 272         const struct stasis_app_control *object_left = obj;
 273         const struct stasis_app_control *object_right = arg;
 274         const char *right_key = arg;
 275         int cmp;
 276
 277         switch (flags & OBJ_SEARCH_MASK) {
 278         case OBJ_SEARCH_OBJECT:
 279                 right_key = stasis_app_control_get_channel_id(object_right);
 280                 /* Fall through */
 281         case OBJ_SEARCH_KEY:
 282                 cmp = strcmp(stasis_app_control_get_channel_id(object_left), right_key);
 283                 break;
 284         case OBJ_SEARCH_PARTIAL_KEY:
 285                 /*
 286                  * We could also use a partial key struct containing a length
 287                  * so strlen() does not get called for every comparison instead.
 288                  */
 289                 cmp = strncmp(stasis_app_control_get_channel_id(object_left), right_key, strlen(right_key));
 290                 break;
 291         default:
 292                 /*
 293                  * What arg points to is specific to this traversal callback
 294                  * and has no special meaning to astobj2.
 295                  */
 296                 cmp = 0;
 297                 break;
 298         }
 299         if (cmp) {
 300                 return 0;
 301         }
 302         /*
 303          * At this point the traversal callback is identical to a sorted
 304          * container.
 305          */
 306         return CMP_MATCH;
 307 }
 308
 309 static int cleanup_cb(void *obj, void *arg, int flags)
 310 {
 311         struct stasis_app *app = obj;
 312
 313         if (!app_is_finished(app)) {
 314                 return 0;
 315         }
 316
 317         ast_verb(1, "Shutting down application '%s'\n", stasis_app_name(app));
 318         app_shutdown(app);
 319
 320         return CMP_MATCH;
 321
 322 }
 323
 324 /*!
 325  * \brief Clean up any old apps that we don't need any more.
 326  */
 327 static void cleanup(void)
 328 {
 329         ao2_callback(apps_registry, OBJ_MULTIPLE | OBJ_NODATA | OBJ_UNLINK,
 330                 cleanup_cb, NULL);
 331 }
 332
 333 struct stasis_app_control *stasis_app_control_create(struct ast_channel *chan)
 334 {
 335         return control_create(chan, NULL);
 336 }
 337
 338 struct stasis_app_control *stasis_app_control_find_by_channel(
 339         const struct ast_channel *chan)
 340 {
 341         if (chan == NULL) {
 342                 return NULL;
 343         }
 344
 345         return stasis_app_control_find_by_channel_id(
 346                 ast_channel_uniqueid(chan));
 347 }
 348
 349 struct stasis_app_control *stasis_app_control_find_by_channel_id(
 350         const char *channel_id)
 351 {
 352         return ao2_find(app_controls, channel_id, OBJ_SEARCH_KEY);
 353 }
 354
 355 /*! AO2 hash function for bridges container  */
 356 static int bridges_hash(const void *obj, const int flags)
 357 {
 358         const struct ast_bridge *bridge;
 359         const char *key;
 360
 361         switch (flags & OBJ_SEARCH_MASK) {
 362         case OBJ_SEARCH_KEY:
 363                 key = obj;
 364                 break;
 365         case OBJ_SEARCH_OBJECT:
 366                 bridge = obj;
 367                 key = bridge->uniqueid;
 368                 break;
 369         default:
 370                 /* Hash can only work on something with a full key. */
 371                 ast_assert(0);
 372                 return 0;
 373         }
 374         return ast_str_hash(key);
 375 }
 376
 377 /*! AO2 comparison function for bridges container */
 378 static int bridges_compare(void *obj, void *arg, int flags)
 379 {
 380         const struct ast_bridge *object_left = obj;
 381         const struct ast_bridge *object_right = arg;
 382         const char *right_key = arg;
 383         int cmp;
 384
 385         switch (flags & OBJ_SEARCH_MASK) {
 386         case OBJ_SEARCH_OBJECT:
 387                 right_key = object_right->uniqueid;
 388                 /* Fall through */
 389         case OBJ_SEARCH_KEY:
 390                 cmp = strcmp(object_left->uniqueid, right_key);
 391                 break;
 392         case OBJ_SEARCH_PARTIAL_KEY:
 393                 /*
 394                  * We could also use a partial key struct containing a length
 395                  * so strlen() does not get called for every comparison instead.
 396                  */
 397                 cmp = strncmp(object_left->uniqueid, right_key, strlen(right_key));
 398                 break;
 399         default:
 400                 /*
 401                  * What arg points to is specific to this traversal callback
 402                  * and has no special meaning to astobj2.
 403                  */
 404                 cmp = 0;
 405                 break;
 406         }
 407         if (cmp) {
 408                 return 0;
 409         }
 410         /*
 411          * At this point the traversal callback is identical to a sorted
 412          * container.
 413          */
 414         return CMP_MATCH;
 415 }
 416
 417 /*!
 418  *  Used with app_bridges_moh and app_bridge_control, they provide links
 419  *  between bridges and channels used for ARI application purposes
 420  */
 421 struct stasis_app_bridge_channel_wrapper {
 422         AST_DECLARE_STRING_FIELDS(
 423                 AST_STRING_FIELD(channel_id);
 424                 AST_STRING_FIELD(bridge_id);
 425         );
 426 };
 427
 428 static void stasis_app_bridge_channel_wrapper_destructor(void *obj)
 429 {
 430         struct stasis_app_bridge_channel_wrapper *wrapper = obj;
 431         ast_string_field_free_memory(wrapper);
 432 }
 433
 434 /*! AO2 hash function for the bridges moh container */
 435 static int bridges_channel_hash_fn(const void *obj, const int flags)
 436 {
 437         const struct stasis_app_bridge_channel_wrapper *wrapper;
 438         const char *key;
 439
 440         switch (flags & OBJ_SEARCH_MASK) {
 441         case OBJ_SEARCH_KEY:
 442                 key = obj;
 443                 break;
 444         case OBJ_SEARCH_OBJECT:
 445                 wrapper = obj;
 446                 key = wrapper->bridge_id;
 447                 break;
 448         default:
 449                 /* Hash can only work on something with a full key. */
 450                 ast_assert(0);
 451                 return 0;
 452         }
 453         return ast_str_hash(key);
 454 }
 455
 456 static int bridges_channel_sort_fn(const void *obj_left, const void *obj_right, const int flags)
 457 {
 458         const struct stasis_app_bridge_channel_wrapper *left = obj_left;
 459         const struct stasis_app_bridge_channel_wrapper *right = obj_right;
 460         const char *right_key = obj_right;
 461         int cmp;
 462
 463         switch (flags & OBJ_SEARCH_MASK) {
 464         case OBJ_SEARCH_OBJECT:
 465                 right_key = right->bridge_id;
 466                 /* Fall through */
 467         case OBJ_SEARCH_KEY:
 468                 cmp = strcmp(left->bridge_id, right_key);
 469                 break;
 470         case OBJ_SEARCH_PARTIAL_KEY:
 471                 cmp = strncmp(left->bridge_id, right_key, strlen(right_key));
 472                 break;
 473         default:
 474                 /* Sort can only work on something with a full or partial key. */
 475                 ast_assert(0);
 476                 cmp = 0;
 477                 break;
 478         }
 479         return cmp;
 480 }
 481
 482 /*! Removes the bridge to music on hold channel link */
 483 static void remove_bridge_moh(char *bridge_id)
 484 {
 485         ao2_find(app_bridges_moh, bridge_id, OBJ_SEARCH_KEY | OBJ_UNLINK | OBJ_NODATA);
 486         ast_free(bridge_id);
 487 }
 488
 489 /*! After bridge failure callback for moh channels */
 490 static void moh_after_bridge_cb_failed(enum ast_bridge_after_cb_reason reason, void *data)
 491 {
 492         char *bridge_id = data;
 493
 494         remove_bridge_moh(bridge_id);
 495 }
 496
 497 /*! After bridge callback for moh channels */
 498 static void moh_after_bridge_cb(struct ast_channel *chan, void *data)
 499 {
 500         char *bridge_id = data;
 501
 502         remove_bridge_moh(bridge_id);
 503 }
 504
 505 /*! Request a bridge MOH channel */
 506 static struct ast_channel *prepare_bridge_moh_channel(void)
 507 {
 508         RAII_VAR(struct ast_format_cap *, cap, NULL, ao2_cleanup);
 509
 510         cap = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
 511         if (!cap) {
 512                 return NULL;
 513         }
 514
 515         ast_format_cap_append(cap, ast_format_slin, 0);
 516
 517         return ast_request("Announcer", cap, NULL, NULL, "ARI_MOH", NULL);
 518 }
 519
 520 /*! Provides the moh channel with a thread so it can actually play its music */
 521 static void *moh_channel_thread(void *data)
 522 {
 523         struct ast_channel *moh_channel = data;
 524
 525         while (!ast_safe_sleep(moh_channel, 1000)) {
 526         }
 527
 528         ast_moh_stop(moh_channel);
 529         ast_hangup(moh_channel);
 530
 531         return NULL;
 532 }
 533
 534 /*!
 535  * \internal
 536  * \brief Creates, pushes, and links a channel for playing music on hold to bridge
 537  *
 538  * \param bridge Which bridge this moh channel exists for
 539  *
 540  * \retval NULL if the channel could not be created, pushed, or linked
 541  * \retval Reference to the channel on success
 542  */
 543 static struct ast_channel *bridge_moh_create(struct ast_bridge *bridge)
 544 {
 545         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, new_wrapper, NULL, ao2_cleanup);
 546         RAII_VAR(char *, bridge_id, ast_strdup(bridge->uniqueid), ast_free);
 547         struct ast_channel *chan;
 548         pthread_t threadid;
 549
 550         if (!bridge_id) {
 551                 return NULL;
 552         }
 553
 554         chan = prepare_bridge_moh_channel();
 555         if (!chan) {
 556                 return NULL;
 557         }
 558
 559         if (stasis_app_channel_unreal_set_internal(chan)) {
 560                 ast_hangup(chan);
 561                 return NULL;
 562         }
 563
 564         /* The after bridge callback assumes responsibility of the bridge_id. */
 565         if (ast_bridge_set_after_callback(chan,
 566                 moh_after_bridge_cb, moh_after_bridge_cb_failed, bridge_id)) {
 567                 ast_hangup(chan);
 568                 return NULL;
 569         }
 570         bridge_id = NULL;
 571
 572         if (ast_unreal_channel_push_to_bridge(chan, bridge,
 573                 AST_BRIDGE_CHANNEL_FLAG_IMMOVABLE | AST_BRIDGE_CHANNEL_FLAG_LONELY)) {
 574                 ast_hangup(chan);
 575                 return NULL;
 576         }
 577
 578         new_wrapper = ao2_alloc_options(sizeof(*new_wrapper),
 579                 stasis_app_bridge_channel_wrapper_destructor, AO2_ALLOC_OPT_LOCK_NOLOCK);
 580         if (!new_wrapper) {
 581                 ast_hangup(chan);
 582                 return NULL;
 583         }
 584
 585         if (ast_string_field_init(new_wrapper, 32)) {
 586                 ast_hangup(chan);
 587                 return NULL;
 588         }
 589         ast_string_field_set(new_wrapper, bridge_id, bridge->uniqueid);
 590         ast_string_field_set(new_wrapper, channel_id, ast_channel_uniqueid(chan));
 591
 592         if (!ao2_link_flags(app_bridges_moh, new_wrapper, OBJ_NOLOCK)) {
 593                 ast_hangup(chan);
 594                 return NULL;
 595         }
 596
 597         if (ast_pthread_create_detached(&threadid, NULL, moh_channel_thread, chan)) {
 598                 ast_log(LOG_ERROR, "Failed to create channel thread. Abandoning MOH channel creation.\n");
 599                 ao2_unlink_flags(app_bridges_moh, new_wrapper, OBJ_NOLOCK);
 600                 ast_hangup(chan);
 601                 return NULL;
 602         }
 603
 604         return chan;
 605 }
 606
 607 struct ast_channel *stasis_app_bridge_moh_channel(struct ast_bridge *bridge)
 608 {
 609         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, moh_wrapper, NULL, ao2_cleanup);
 610
 611         {
 612                 SCOPED_AO2LOCK(lock, app_bridges_moh);
 613
 614                 moh_wrapper = ao2_find(app_bridges_moh, bridge->uniqueid, OBJ_SEARCH_KEY | OBJ_NOLOCK);
 615                 if (!moh_wrapper) {
 616                         return bridge_moh_create(bridge);
 617                 }
 618         }
 619
 620         return ast_channel_get_by_name(moh_wrapper->channel_id);
 621 }
 622
 623 int stasis_app_bridge_moh_stop(struct ast_bridge *bridge)
 624 {
 625         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, moh_wrapper, NULL, ao2_cleanup);
 626         struct ast_channel *chan;
 627
 628         moh_wrapper = ao2_find(app_bridges_moh, bridge->uniqueid, OBJ_SEARCH_KEY | OBJ_UNLINK);
 629         if (!moh_wrapper) {
 630                 return -1;
 631         }
 632
 633         chan = ast_channel_get_by_name(moh_wrapper->channel_id);
 634         if (!chan) {
 635                 return -1;
 636         }
 637
 638         ast_moh_stop(chan);
 639         ast_softhangup(chan, AST_CAUSE_NORMAL_CLEARING);
 640         ao2_cleanup(chan);
 641
 642         return 0;
 643 }
 644
 645 /*! Removes the bridge to playback channel link */
 646 static void remove_bridge_playback(char *bridge_id)
 647 {
 648         struct stasis_app_bridge_channel_wrapper *wrapper;
 649         struct stasis_app_control *control;
 650
 651         wrapper = ao2_find(app_bridges_playback, bridge_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
 652
 653         if (wrapper) {
 654                 control = stasis_app_control_find_by_channel_id(wrapper->channel_id);
 655                 if (control) {
 656                         ao2_unlink(app_controls, control);
 657                         ao2_ref(control, -1);
 658                 }
 659                 ao2_ref(wrapper, -1);
 660         }
 661         ast_free(bridge_id);
 662 }
 663
 664 static void playback_after_bridge_cb_failed(enum ast_bridge_after_cb_reason reason, void *data)
 665 {
 666         char *bridge_id = data;
 667
 668         remove_bridge_playback(bridge_id);
 669 }
 670
 671 static void playback_after_bridge_cb(struct ast_channel *chan, void *data)
 672 {
 673         char *bridge_id = data;
 674
 675         remove_bridge_playback(bridge_id);
 676 }
 677
 678 int stasis_app_bridge_playback_channel_add(struct ast_bridge *bridge,
 679         struct ast_channel *chan,
 680         struct stasis_app_control *control)
 681 {
 682         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, new_wrapper, NULL, ao2_cleanup);
 683         char *bridge_id = ast_strdup(bridge->uniqueid);
 684
 685         if (!bridge_id) {
 686                 return -1;
 687         }
 688
 689         if (ast_bridge_set_after_callback(chan,
 690                 playback_after_bridge_cb, playback_after_bridge_cb_failed, bridge_id)) {
 691                 ast_free(bridge_id);
 692                 return -1;
 693         }
 694
 695         new_wrapper = ao2_alloc_options(sizeof(*new_wrapper),
 696                 stasis_app_bridge_channel_wrapper_destructor, AO2_ALLOC_OPT_LOCK_NOLOCK);
 697         if (!new_wrapper) {
 698                 return -1;
 699         }
 700
 701         if (ast_string_field_init(new_wrapper, 32)) {
 702                 return -1;
 703         }
 704
 705         ast_string_field_set(new_wrapper, bridge_id, bridge->uniqueid);
 706         ast_string_field_set(new_wrapper, channel_id, ast_channel_uniqueid(chan));
 707
 708         if (!ao2_link(app_bridges_playback, new_wrapper)) {
 709                 return -1;
 710         }
 711
 712         ao2_link(app_controls, control);
 713         return 0;
 714 }
 715
 716 void stasis_app_bridge_playback_channel_remove(char *bridge_id,
 717         struct stasis_app_control *control)
 718 {
 719         struct stasis_app_bridge_channel_wrapper *wrapper;
 720
 721         wrapper = ao2_find(app_bridges_playback, bridge_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
 722         if (wrapper) {
 723                 /* If wrapper is not found, then that means the after bridge callback has been
 724                  * called or is in progress. No need to unlink the control here since that has
 725                  * been done or is about to be done in the after bridge callback
 726                  */
 727                 ao2_unlink(app_controls, control);
 728                 ao2_ref(wrapper, -1);
 729         }
 730 }
 731
 732 struct ast_channel *stasis_app_bridge_playback_channel_find(struct ast_bridge *bridge)
 733 {
 734         struct stasis_app_bridge_channel_wrapper *playback_wrapper;
 735         struct ast_channel *chan;
 736
 737         playback_wrapper = ao2_find(app_bridges_playback, bridge->uniqueid, OBJ_SEARCH_KEY);
 738         if (!playback_wrapper) {
 739                 return NULL;
 740         }
 741
 742         chan = ast_channel_get_by_name(playback_wrapper->channel_id);
 743         ao2_ref(playback_wrapper, -1);
 744         return chan;
 745 }
 746
 747 struct ast_bridge *stasis_app_bridge_find_by_id(
 748         const char *bridge_id)
 749 {
 750         return ao2_find(app_bridges, bridge_id, OBJ_SEARCH_KEY);
 751 }
 752
 753
 754 /*!
 755  * \brief In addition to running ao2_cleanup(), this function also removes the
 756  * object from the app_controls container.
 757  */
 758 static void control_unlink(struct stasis_app_control *control)
 759 {
 760         if (!control) {
 761                 return;
 762         }
 763
 764         ao2_unlink(app_controls, control);
 765         ao2_cleanup(control);
 766 }
 767
 768 static struct ast_bridge *bridge_create_common(const char *type, const char *name, const char *id, int invisible)
 769 {
 770         struct ast_bridge *bridge;
 771         char *requested_type, *requested_types = ast_strdupa(S_OR(type, "mixing"));
 772         int capabilities = 0;
 773         int flags = AST_BRIDGE_FLAG_MERGE_INHIBIT_FROM | AST_BRIDGE_FLAG_MERGE_INHIBIT_TO
 774                 | AST_BRIDGE_FLAG_SWAP_INHIBIT_FROM | AST_BRIDGE_FLAG_SWAP_INHIBIT_TO
 775                 | AST_BRIDGE_FLAG_TRANSFER_BRIDGE_ONLY;
 776
 777         if (invisible) {
 778                 flags |= AST_BRIDGE_FLAG_INVISIBLE;
 779         }
 780
 781         while ((requested_type = strsep(&requested_types, ","))) {
 782                 requested_type = ast_strip(requested_type);
 783
 784                 if (!strcmp(requested_type, "mixing")) {
 785                         capabilities |= STASIS_BRIDGE_MIXING_CAPABILITIES;
 786                         flags |= AST_BRIDGE_FLAG_SMART;
 787                 } else if (!strcmp(requested_type, "holding")) {
 788                         capabilities |= AST_BRIDGE_CAPABILITY_HOLDING;
 789                 } else if (!strcmp(requested_type, "dtmf_events") ||
 790                         !strcmp(requested_type, "proxy_media")) {
 791                         capabilities &= ~AST_BRIDGE_CAPABILITY_NATIVE;
 792                 }
 793         }
 794
 795         if (!capabilities
 796                 /* Holding and mixing capabilities don't mix. */
 797                 || ((capabilities & AST_BRIDGE_CAPABILITY_HOLDING)
 798                         && (capabilities & (STASIS_BRIDGE_MIXING_CAPABILITIES)))) {
 799                 return NULL;
 800         }
 801
 802         bridge = bridge_stasis_new(capabilities, flags, name, id);
 803         if (bridge) {
 804                 if (!ao2_link(app_bridges, bridge)) {
 805                         ast_bridge_destroy(bridge, 0);
 806                         bridge = NULL;
 807                 }
 808         }
 809         return bridge;
 810 }
 811
 812 struct ast_bridge *stasis_app_bridge_create(const char *type, const char *name, const char *id)
 813 {
 814         return bridge_create_common(type, name, id, 0);
 815 }
 816
 817 struct ast_bridge *stasis_app_bridge_create_invisible(const char *type, const char *name, const char *id)
 818 {
 819         return bridge_create_common(type, name, id, 1);
 820 }
 821
 822 void stasis_app_bridge_destroy(const char *bridge_id)
 823 {
 824         struct ast_bridge *bridge = stasis_app_bridge_find_by_id(bridge_id);
 825         if (!bridge) {
 826                 return;
 827         }
 828         ao2_unlink(app_bridges, bridge);
 829         ast_bridge_destroy(bridge, 0);
 830 }
 831
 832 struct replace_channel_store {
 833         struct ast_channel_snapshot *snapshot;
 834         char *app;
 835 };
 836
 837 static void replace_channel_destroy(void *obj)
 838 {
 839         struct replace_channel_store *replace = obj;
 840
 841         ao2_cleanup(replace->snapshot);
 842         ast_free(replace->app);
 843         ast_free(replace);
 844 }
 845
 846 static const struct ast_datastore_info replace_channel_store_info = {
 847         .type = "replace-channel-store",
 848         .destroy = replace_channel_destroy,
 849 };
 850
 851 static struct replace_channel_store *get_replace_channel_store(struct ast_channel *chan, int no_create)
 852 {
 853         struct ast_datastore *datastore;
 854
 855         SCOPED_CHANNELLOCK(lock, chan);
 856         datastore = ast_channel_datastore_find(chan, &replace_channel_store_info, NULL);
 857         if (!datastore) {
 858                 if (no_create) {
 859                         return NULL;
 860                 }
 861
 862                 datastore = ast_datastore_alloc(&replace_channel_store_info, NULL);
 863                 if (!datastore) {
 864                         return NULL;
 865                 }
 866                 ast_channel_datastore_add(chan, datastore);
 867         }
 868
 869         if (!datastore->data) {
 870                 datastore->data = ast_calloc(1, sizeof(struct replace_channel_store));
 871         }
 872         return datastore->data;
 873 }
 874
 875 int app_set_replace_channel_snapshot(struct ast_channel *chan, struct ast_channel_snapshot *replace_snapshot)
 876 {
 877         struct replace_channel_store *replace = get_replace_channel_store(chan, 0);
 878
 879         if (!replace) {
 880                 return -1;
 881         }
 882
 883         ao2_replace(replace->snapshot, replace_snapshot);
 884         return 0;
 885 }
 886
 887 int app_set_replace_channel_app(struct ast_channel *chan, const char *replace_app)
 888 {
 889         struct replace_channel_store *replace = get_replace_channel_store(chan, 0);
 890
 891         if (!replace) {
 892                 return -1;
 893         }
 894
 895         ast_free(replace->app);
 896         replace->app = NULL;
 897
 898         if (replace_app) {
 899                 replace->app = ast_strdup(replace_app);
 900                 if (!replace->app) {
 901                         return -1;
 902                 }
 903         }
 904
 905         return 0;
 906 }
 907
 908 static struct ast_channel_snapshot *get_replace_channel_snapshot(struct ast_channel *chan)
 909 {
 910         struct replace_channel_store *replace = get_replace_channel_store(chan, 1);
 911         struct ast_channel_snapshot *replace_channel_snapshot;
 912
 913         if (!replace) {
 914                 return NULL;
 915         }
 916
 917         replace_channel_snapshot = replace->snapshot;
 918         replace->snapshot = NULL;
 919
 920         return replace_channel_snapshot;
 921 }
 922
 923 char *app_get_replace_channel_app(struct ast_channel *chan)
 924 {
 925         struct replace_channel_store *replace = get_replace_channel_store(chan, 1);
 926         char *replace_channel_app;
 927
 928         if (!replace) {
 929                 return NULL;
 930         }
 931
 932         replace_channel_app = replace->app;
 933         replace->app = NULL;
 934
 935         return replace_channel_app;
 936 }
 937
 938 static void start_message_blob_dtor(void *obj)
 939 {
 940         struct start_message_blob *payload = obj;
 941
 942         ao2_cleanup(payload->channel);
 943         ao2_cleanup(payload->replace_channel);
 944         ast_json_unref(payload->blob);
 945 }
 946
 947 static int send_start_msg_snapshots(struct ast_channel *chan, struct stasis_app *app,
 948         int argc, char *argv[], struct ast_channel_snapshot *snapshot,
 949         struct ast_channel_snapshot *replace_channel_snapshot)
 950 {
 951         RAII_VAR(struct ast_json *, json_blob, NULL, ast_json_unref);
 952         struct ast_json *json_args;
 953         RAII_VAR(struct start_message_blob *, payload, NULL, ao2_cleanup);
 954         struct stasis_message *msg;
 955         int i;
 956
 957         if (app_subscribe_channel(app, chan)) {
 958                 ast_log(LOG_ERROR, "Error subscribing app '%s' to channel '%s'\n",
 959                         app_name(app), ast_channel_name(chan));
 960                 return -1;
 961         }
 962
 963         payload = ao2_alloc(sizeof(*payload), start_message_blob_dtor);
 964         if (!payload) {
 965                 ast_log(LOG_ERROR, "Error packing JSON for StasisStart message\n");
 966                 return -1;
 967         }
 968
 969         payload->channel = ao2_bump(snapshot);
 970         payload->replace_channel = ao2_bump(replace_channel_snapshot);
 971
 972         json_blob = ast_json_pack("{s: s, s: o, s: []}",
 973                 "app", app_name(app),
 974                 "timestamp", ast_json_timeval(ast_tvnow(), NULL),
 975                 "args");
 976         if (!json_blob) {
 977                 ast_log(LOG_ERROR, "Error packing JSON for StasisStart message\n");
 978                 return -1;
 979         }
 980
 981         /* Append arguments to args array */
 982         json_args = ast_json_object_get(json_blob, "args");
 983         ast_assert(json_args != NULL);
 984         for (i = 0; i < argc; ++i) {
 985                 int r = ast_json_array_append(json_args,
 986                                               ast_json_string_create(argv[i]));
 987                 if (r != 0) {
 988                         ast_log(LOG_ERROR, "Error appending to StasisStart message\n");
 989                         return -1;
 990                 }
 991         }
 992
 993         payload->blob = ast_json_ref(json_blob);
 994
 995         msg = stasis_message_create(start_message_type(), payload);
 996         if (!msg) {
 997                 ast_log(LOG_ERROR, "Error sending StasisStart message\n");
 998                 return -1;
 999         }
1000
1001         if (replace_channel_snapshot) {
1002                 app_unsubscribe_channel_id(app, replace_channel_snapshot->uniqueid);
1003         }
1004         stasis_publish(ast_app_get_topic(app), msg);
1005         ao2_ref(msg, -1);
1006         return 0;
1007 }
1008
1009 static int send_start_msg(struct stasis_app *app, struct ast_channel *chan,
1010         int argc, char *argv[])
1011 {
1012         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
1013         RAII_VAR(struct ast_channel_snapshot *, replace_channel_snapshot,
1014                 NULL, ao2_cleanup);
1015
1016         ast_assert(chan != NULL);
1017
1018         replace_channel_snapshot = get_replace_channel_snapshot(chan);
1019
1020         /* Set channel info */
1021         ast_channel_lock(chan);
1022         snapshot = ast_channel_snapshot_create(chan);
1023         ast_channel_unlock(chan);
1024         if (!snapshot) {
1025                 return -1;
1026         }
1027         return send_start_msg_snapshots(chan, app, argc, argv, snapshot, replace_channel_snapshot);
1028 }
1029
1030 static void remove_masquerade_store(struct ast_channel *chan);
1031
1032 int app_send_end_msg(struct stasis_app *app, struct ast_channel *chan)
1033 {
1034         struct stasis_message_sanitizer *sanitize = stasis_app_get_sanitizer();
1035         struct ast_json *blob;
1036         struct stasis_message *msg;
1037
1038         if (sanitize && sanitize->channel
1039                 && sanitize->channel(chan)) {
1040                 return 0;
1041         }
1042
1043         blob = ast_json_pack("{s: s}", "app", app_name(app));
1044         if (!blob) {
1045                 ast_log(LOG_ERROR, "Error packing JSON for StasisEnd message\n");
1046                 return -1;
1047         }
1048
1049         remove_masquerade_store(chan);
1050         app_unsubscribe_channel(app, chan);
1051         msg = ast_channel_blob_create(chan, end_message_type(), blob);
1052         if (msg) {
1053                 stasis_publish(ast_app_get_topic(app), msg);
1054         }
1055         ao2_cleanup(msg);
1056         ast_json_unref(blob);
1057
1058         return 0;
1059 }
1060
1061 static int masq_match_cb(void *obj, void *data, int flags)
1062 {
1063         struct stasis_app_control *control = obj;
1064         struct ast_channel *chan = data;
1065
1066         if (!strcmp(ast_channel_uniqueid(chan),
1067                 stasis_app_control_get_channel_id(control))) {
1068                 return CMP_MATCH;
1069         }
1070
1071         return 0;
1072 }
1073
1074 static void channel_stolen_cb(void *data, struct ast_channel *old_chan, struct ast_channel *new_chan)
1075 {
1076         struct stasis_app_control *control;
1077
1078         /* find control */
1079         control = ao2_callback(app_controls, 0, masq_match_cb, old_chan);
1080         if (!control) {
1081                 ast_log(LOG_ERROR, "Could not find control for masqueraded channel\n");
1082                 return;
1083         }
1084
1085         /* send the StasisEnd message to the app */
1086         stasis_app_channel_set_stasis_end_published(new_chan);
1087         app_send_end_msg(control_app(control), new_chan);
1088
1089         /* remove the datastore */
1090         remove_masquerade_store(old_chan);
1091
1092         ao2_cleanup(control);
1093 }
1094
1095 static void channel_replaced_cb(void *data, struct ast_channel *old_chan, struct ast_channel *new_chan)
1096 {
1097         RAII_VAR(struct ast_channel_snapshot *, new_snapshot, NULL, ao2_cleanup);
1098         RAII_VAR(struct ast_channel_snapshot *, old_snapshot, NULL, ao2_cleanup);
1099         struct stasis_app_control *control;
1100
1101         /* At this point, new_chan is the channel pointer that is in Stasis() and
1102          * has the unknown channel's name in it while old_chan is the channel pointer
1103          * that is not in Stasis(), but has the guts of the channel that Stasis() knows
1104          * about */
1105
1106         /* grab a snapshot for the channel that is jumping into Stasis() */
1107         new_snapshot = ast_channel_snapshot_get_latest(ast_channel_uniqueid(new_chan));
1108         if (!new_snapshot) {
1109                 ast_log(LOG_ERROR, "Could not get snapshot for masquerading channel\n");
1110                 return;
1111         }
1112
1113         /* grab a snapshot for the channel that has been kicked out of Stasis() */
1114         old_snapshot = ast_channel_snapshot_get_latest(ast_channel_uniqueid(old_chan));
1115         if (!old_snapshot) {
1116                 ast_log(LOG_ERROR, "Could not get snapshot for masqueraded channel\n");
1117                 return;
1118         }
1119
1120         /* find, unlink, and relink control since the channel has a new name and
1121          * its hash has likely changed */
1122         control = ao2_callback(app_controls, OBJ_UNLINK, masq_match_cb, new_chan);
1123         if (!control) {
1124                 ast_log(LOG_ERROR, "Could not find control for masquerading channel\n");
1125                 return;
1126         }
1127         ao2_link(app_controls, control);
1128
1129
1130         /* send the StasisStart with replace_channel to the app */
1131         send_start_msg_snapshots(new_chan, control_app(control), 0, NULL, new_snapshot,
1132                 old_snapshot);
1133         /* send the StasisEnd message to the app */
1134         app_send_end_msg(control_app(control), old_chan);
1135
1136         ao2_cleanup(control);
1137 }
1138
1139 static const struct ast_datastore_info masquerade_store_info = {
1140         .type = "stasis-masqerade",
1141         .chan_fixup = channel_stolen_cb,
1142         .chan_breakdown = channel_replaced_cb,
1143 };
1144
1145 static int has_masquerade_store(struct ast_channel *chan)
1146 {
1147         SCOPED_CHANNELLOCK(lock, chan);
1148         return !!ast_channel_datastore_find(chan, &masquerade_store_info, NULL);
1149 }
1150
1151 static int add_masquerade_store(struct ast_channel *chan)
1152 {
1153         struct ast_datastore *datastore;
1154
1155         SCOPED_CHANNELLOCK(lock, chan);
1156         if (ast_channel_datastore_find(chan, &masquerade_store_info, NULL)) {
1157                 return 0;
1158         }
1159
1160         datastore = ast_datastore_alloc(&masquerade_store_info, NULL);
1161         if (!datastore) {
1162                 return -1;
1163         }
1164
1165         ast_channel_datastore_add(chan, datastore);
1166
1167         return 0;
1168 }
1169
1170 static void remove_masquerade_store(struct ast_channel *chan)
1171 {
1172         struct ast_datastore *datastore;
1173
1174         SCOPED_CHANNELLOCK(lock, chan);
1175         datastore = ast_channel_datastore_find(chan, &masquerade_store_info, NULL);
1176         if (!datastore) {
1177                 return;
1178         }
1179
1180         ast_channel_datastore_remove(chan, datastore);
1181         ast_datastore_free(datastore);
1182 }
1183
1184 void stasis_app_control_execute_until_exhausted(struct ast_channel *chan, struct stasis_app_control *control)
1185 {
1186         while (!control_is_done(control)) {
1187                 int command_count;
1188                 command_count = control_dispatch_all(control, chan);
1189
1190                 ao2_lock(control);
1191
1192                 if (control_command_count(control)) {
1193                         /* If the command queue isn't empty, something added to the queue before it was locked. */
1194                         ao2_unlock(control);
1195                         continue;
1196                 }
1197
1198                 if (command_count == 0 || ast_channel_fdno(chan) == -1) {
1199                         control_mark_done(control);
1200                         ao2_unlock(control);
1201                         break;
1202                 }
1203                 ao2_unlock(control);
1204         }
1205 }
1206
1207 int stasis_app_control_is_done(struct stasis_app_control *control)
1208 {
1209         return control_is_done(control);
1210 }
1211
1212 void stasis_app_control_flush_queue(struct stasis_app_control *control)
1213 {
1214         control_flush_queue(control);
1215 }
1216
1217 struct ast_datastore_info set_end_published_info = {
1218         .type = "stasis_end_published",
1219 };
1220
1221 void stasis_app_channel_set_stasis_end_published(struct ast_channel *chan)
1222 {
1223         struct ast_datastore *datastore;
1224
1225         datastore = ast_datastore_alloc(&set_end_published_info, NULL);
1226         if (datastore) {
1227                 ast_channel_lock(chan);
1228                 ast_channel_datastore_add(chan, datastore);
1229                 ast_channel_unlock(chan);
1230         }
1231 }
1232
1233 int stasis_app_channel_is_stasis_end_published(struct ast_channel *chan)
1234 {
1235         struct ast_datastore *datastore;
1236
1237         ast_channel_lock(chan);
1238         datastore = ast_channel_datastore_find(chan, &set_end_published_info, NULL);
1239         ast_channel_unlock(chan);
1240
1241         return datastore ? 1 : 0;
1242 }
1243
1244 static void remove_stasis_end_published(struct ast_channel *chan)
1245 {
1246         struct ast_datastore *datastore;
1247
1248         ast_channel_lock(chan);
1249         datastore = ast_channel_datastore_find(chan, &set_end_published_info, NULL);
1250         if (datastore) {
1251                 ast_channel_datastore_remove(chan, datastore);
1252                 ast_datastore_free(datastore);
1253         }
1254         ast_channel_unlock(chan);
1255 }
1256
1257 /*! /brief Stasis dialplan application callback */
1258 int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
1259                     char *argv[])
1260 {
1261         SCOPED_MODULE_USE(ast_module_info->self);
1262
1263         RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
1264         RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
1265         struct ast_bridge *bridge = NULL;
1266         int res = 0;
1267         int needs_depart;
1268
1269         ast_assert(chan != NULL);
1270
1271         /* Just in case there's a lingering indication that the channel has had a stasis
1272          * end published on it, remove that now.
1273          */
1274         remove_stasis_end_published(chan);
1275
1276         if (!apps_registry) {
1277                 return -1;
1278         }
1279
1280         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1281         if (!app) {
1282                 ast_log(LOG_ERROR,
1283                         "Stasis app '%s' not registered\n", app_name);
1284                 return -1;
1285         }
1286         if (!app_is_active(app)) {
1287                 ast_log(LOG_ERROR,
1288                         "Stasis app '%s' not active\n", app_name);
1289                 return -1;
1290         }
1291
1292         control = control_create(chan, app);
1293         if (!control) {
1294                 ast_log(LOG_ERROR, "Allocated failed\n");
1295                 return -1;
1296         }
1297         ao2_link(app_controls, control);
1298
1299         if (add_masquerade_store(chan)) {
1300                 ast_log(LOG_ERROR, "Failed to attach masquerade detector\n");
1301                 return -1;
1302         }
1303
1304         res = send_start_msg(app, chan, argc, argv);
1305         if (res != 0) {
1306                 ast_log(LOG_ERROR,
1307                         "Error sending start message to '%s'\n", app_name);
1308                 remove_masquerade_store(chan);
1309                 return -1;
1310         }
1311
1312         /* Pull queued prestart commands and execute */
1313         control_prestart_dispatch_all(control, chan);
1314
1315         while (!control_is_done(control)) {
1316                 RAII_VAR(struct ast_frame *, f, NULL, ast_frame_dtor);
1317                 int r;
1318                 int command_count;
1319                 RAII_VAR(struct ast_bridge *, last_bridge, NULL, ao2_cleanup);
1320
1321                 /* Check to see if a bridge absorbed our hangup frame */
1322                 if (ast_check_hangup_locked(chan)) {
1323                         control_mark_done(control);
1324                         break;
1325                 }
1326
1327                 last_bridge = bridge;
1328                 bridge = ao2_bump(stasis_app_get_bridge(control));
1329
1330                 if (bridge != last_bridge) {
1331                         app_unsubscribe_bridge(app, last_bridge);
1332                         if (bridge) {
1333                                 app_subscribe_bridge(app, bridge);
1334                         }
1335                 }
1336
1337                 if (bridge) {
1338                         /* Bridge/dial is handling channel frames */
1339                         control_wait(control);
1340                         control_dispatch_all(control, chan);
1341                         continue;
1342                 }
1343
1344                 r = ast_waitfor(chan, MAX_WAIT_MS);
1345
1346                 if (r < 0) {
1347                         ast_debug(3, "%s: Poll error\n",
1348                                   ast_channel_uniqueid(chan));
1349                         control_mark_done(control);
1350                         break;
1351                 }
1352
1353                 command_count = control_dispatch_all(control, chan);
1354
1355                 if (command_count > 0 && ast_channel_fdno(chan) == -1) {
1356                         /* Command drained the channel; wait for next frame */
1357                         continue;
1358                 }
1359
1360                 if (r == 0) {
1361                         /* Timeout */
1362                         continue;
1363                 }
1364
1365                 f = ast_read(chan);
1366                 if (!f) {
1367                         /* Continue on in the dialplan */
1368                         ast_debug(3, "%s: Hangup (no more frames)\n",
1369                                 ast_channel_uniqueid(chan));
1370                         control_mark_done(control);
1371                         break;
1372                 }
1373
1374                 if (f->frametype == AST_FRAME_CONTROL) {
1375                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
1376                                 /* Continue on in the dialplan */
1377                                 ast_debug(3, "%s: Hangup\n",
1378                                         ast_channel_uniqueid(chan));
1379                                 control_mark_done(control);
1380                                 break;
1381                         }
1382                 }
1383         }
1384
1385         ast_channel_lock(chan);
1386         needs_depart = (ast_channel_internal_bridge_channel(chan) != NULL);
1387         ast_channel_unlock(chan);
1388         if (needs_depart) {
1389                 ast_bridge_depart(chan);
1390         }
1391
1392         app_unsubscribe_bridge(app, stasis_app_get_bridge(control));
1393         ao2_cleanup(bridge);
1394
1395         /* Only publish a stasis_end event if it hasn't already been published */
1396         if (!stasis_app_channel_is_stasis_end_published(chan)) {
1397                 /* A masquerade has occurred and this message will be wrong so it
1398                  * has already been sent elsewhere. */
1399                 res = has_masquerade_store(chan) && app_send_end_msg(app, chan);
1400                 if (res != 0) {
1401                         ast_log(LOG_ERROR,
1402                                 "Error sending end message to %s\n", app_name);
1403                         return res;
1404                 }
1405         } else {
1406                 remove_stasis_end_published(chan);
1407         }
1408
1409         control_flush_queue(control);
1410
1411         /* Stop any lingering silence generator */
1412         control_silence_stop_now(control);
1413
1414         /* There's an off chance that app is ready for cleanup. Go ahead
1415          * and clean up, just in case
1416          */
1417         cleanup();
1418
1419         /* The control needs to be removed from the controls container in
1420          * case a new PBX is started and ends up coming back into Stasis.
1421          */
1422         ao2_cleanup(app);
1423         app = NULL;
1424         control_unlink(control);
1425         control = NULL;
1426
1427         if (!ast_channel_pbx(chan)) {
1428                 int chan_hungup;
1429
1430                 /* The ASYNCGOTO softhangup flag may have broken the channel out of
1431                  * its bridge to run dialplan, so if there's no pbx on the channel
1432                  * let it run dialplan here. Otherwise, it will run when this
1433                  * application exits. */
1434                 ast_channel_lock(chan);
1435                 ast_channel_clear_softhangup(chan, AST_SOFTHANGUP_ASYNCGOTO);
1436                 chan_hungup = ast_check_hangup(chan);
1437                 ast_channel_unlock(chan);
1438
1439                 if (!chan_hungup) {
1440                         struct ast_pbx_args pbx_args;
1441
1442                         memset(&pbx_args, 0, sizeof(pbx_args));
1443                         pbx_args.no_hangup_chan = 1;
1444
1445                         res = ast_pbx_run_args(chan, &pbx_args);
1446                 }
1447         }
1448
1449         return res;
1450 }
1451
1452 int stasis_app_send(const char *app_name, struct ast_json *message)
1453 {
1454         RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
1455
1456         if (!apps_registry) {
1457                 return -1;
1458         }
1459
1460         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1461         if (!app) {
1462                 /* XXX We can do a better job handling late binding, queueing up
1463                  * the call for a few seconds to wait for the app to register.
1464                  */
1465                 ast_log(LOG_WARNING,
1466                         "Stasis app '%s' not registered\n", app_name);
1467                 return -1;
1468         }
1469         app_send(app, message);
1470         return 0;
1471 }
1472
1473 static struct stasis_app *find_app_by_name(const char *app_name)
1474 {
1475         struct stasis_app *res = NULL;
1476
1477         if (!apps_registry) {
1478                 return NULL;
1479         }
1480
1481         if (!ast_strlen_zero(app_name)) {
1482                 res = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1483         }
1484
1485         if (!res) {
1486                 ast_log(LOG_WARNING, "Could not find app '%s'\n",
1487                         app_name ? : "(null)");
1488         }
1489         return res;
1490 }
1491
1492 static int append_name(void *obj, void *arg, int flags)
1493 {
1494         struct stasis_app *app = obj;
1495         struct ao2_container *apps = arg;
1496
1497         ast_str_container_add(apps, stasis_app_name(app));
1498         return 0;
1499 }
1500
1501 struct ao2_container *stasis_app_get_all(void)
1502 {
1503         RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
1504
1505         if (!apps_registry) {
1506                 return NULL;
1507         }
1508
1509         apps = ast_str_container_alloc(1);
1510         if (!apps) {
1511                 return NULL;
1512         }
1513
1514         ao2_callback(apps_registry, OBJ_NODATA, append_name, apps);
1515
1516         return ao2_bump(apps);
1517 }
1518
1519 static int __stasis_app_register(const char *app_name, stasis_app_cb handler, void *data, int all_events)
1520 {
1521         RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
1522
1523         if (!apps_registry) {
1524                 return -1;
1525         }
1526
1527         ao2_lock(apps_registry);
1528         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
1529         if (app) {
1530                 app_update(app, handler, data);
1531         } else {
1532                 app = app_create(app_name, handler, data, all_events ? STASIS_APP_SUBSCRIBE_ALL : STASIS_APP_SUBSCRIBE_MANUAL);
1533                 if (app) {
1534                         if (all_events) {
1535                                 struct stasis_app_event_source *source;
1536                                 SCOPED_LOCK(lock, &event_sources, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
1537
1538                                 AST_LIST_TRAVERSE(&event_sources, source, next) {
1539                                         if (!source->subscribe) {
1540                                                 continue;
1541                                         }
1542
1543                                         source->subscribe(app, NULL);
1544                                 }
1545                         }
1546                         ao2_link_flags(apps_registry, app, OBJ_NOLOCK);
1547                 } else {
1548                         ao2_unlock(apps_registry);
1549                         return -1;
1550                 }
1551         }
1552
1553         /* We lazily clean up the apps_registry, because it's good enough to
1554          * prevent memory leaks, and we're lazy.
1555          */
1556         cleanup();
1557         ao2_unlock(apps_registry);
1558         return 0;
1559 }
1560
1561 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
1562 {
1563         return __stasis_app_register(app_name, handler, data, 0);
1564 }
1565
1566 int stasis_app_register_all(const char *app_name, stasis_app_cb handler, void *data)
1567 {
1568         return __stasis_app_register(app_name, handler, data, 1);
1569 }
1570
1571 void stasis_app_unregister(const char *app_name)
1572 {
1573         RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
1574
1575         if (!app_name) {
1576                 return;
1577         }
1578
1579         if (!apps_registry) {
1580                 return;
1581         }
1582
1583         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1584         if (!app) {
1585                 ast_log(LOG_ERROR,
1586                         "Stasis app '%s' not registered\n", app_name);
1587                 return;
1588         }
1589
1590         app_deactivate(app);
1591
1592         /* There's a decent chance that app is ready for cleanup. Go ahead
1593          * and clean up, just in case
1594          */
1595         cleanup();
1596 }
1597
1598 void stasis_app_register_event_source(struct stasis_app_event_source *obj)
1599 {
1600         SCOPED_LOCK(lock, &event_sources, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
1601         AST_LIST_INSERT_TAIL(&event_sources, obj, next);
1602         /* only need to bump the module ref on non-core sources because the
1603            core ones are [un]registered by this module. */
1604         if (!stasis_app_is_core_event_source(obj)) {
1605                 ast_module_ref(ast_module_info->self);
1606         }
1607 }
1608
1609 void stasis_app_unregister_event_source(struct stasis_app_event_source *obj)
1610 {
1611         struct stasis_app_event_source *source;
1612         SCOPED_LOCK(lock, &event_sources, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
1613         AST_RWLIST_TRAVERSE_SAFE_BEGIN(&event_sources, source, next) {
1614                 if (source == obj) {
1615                         AST_RWLIST_REMOVE_CURRENT(next);
1616                         if (!stasis_app_is_core_event_source(obj)) {
1617                                 ast_module_unref(ast_module_info->self);
1618                         }
1619                         break;
1620                 }
1621         }
1622         AST_RWLIST_TRAVERSE_SAFE_END;
1623 }
1624
1625 /*!
1626  * \internal
1627  * \brief Convert event source data to JSON.
1628  *
1629  * Calls each event source that has a "to_json" handler allowing each
1630  * source to add data to the given JSON object.
1631  *
1632  * \param app application associated with the event source
1633  * \param json a json object to "fill"
1634  *
1635  * \retval The given json object.
1636  */
1637 static struct ast_json *app_event_sources_to_json(
1638         const struct stasis_app *app, struct ast_json *json)
1639 {
1640         struct stasis_app_event_source *source;
1641         SCOPED_LOCK(lock, &event_sources, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
1642         AST_LIST_TRAVERSE(&event_sources, source, next) {
1643                 if (source->to_json) {
1644                         source->to_json(app, json);
1645                 }
1646         }
1647         return json;
1648 }
1649
1650 static struct ast_json *stasis_app_object_to_json(struct stasis_app *app)
1651 {
1652         if (!app) {
1653                 return NULL;
1654         }
1655
1656         return app_event_sources_to_json(app, app_to_json(app));
1657 }
1658
1659 struct ast_json *stasis_app_to_json(const char *app_name)
1660 {
1661         RAII_VAR(struct stasis_app *, app, find_app_by_name(app_name), ao2_cleanup);
1662
1663         return stasis_app_object_to_json(app);
1664 }
1665
1666 /*!
1667  * \internal
1668  * \brief Finds an event source that matches a uri scheme.
1669  *
1670  * Uri(s) should begin with a particular scheme that can be matched
1671  * against an event source.
1672  *
1673  * \param uri uri containing a scheme to match
1674  *
1675  * \retval an event source if found, NULL otherwise.
1676  */
1677 static struct stasis_app_event_source *app_event_source_find(const char *uri)
1678 {
1679         struct stasis_app_event_source *source;
1680         SCOPED_LOCK(lock, &event_sources, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
1681         AST_LIST_TRAVERSE(&event_sources, source, next) {
1682                 if (ast_begins_with(uri, source->scheme)) {
1683                         return source;
1684                 }
1685         }
1686         return NULL;
1687 }
1688
1689 /*!
1690  * \internal
1691  * \brief Callback for subscription handling
1692  *
1693  * \param app [un]subscribing application
1694  * \param uri scheme:id of an event source
1695  * \param event_source being [un]subscribed [from]to
1696  *
1697  * \retval stasis_app_subscribe_res return code.
1698  */
1699 typedef enum stasis_app_subscribe_res (*app_subscription_handler)(
1700         struct stasis_app *app, const char *uri,
1701         struct stasis_app_event_source *event_source);
1702
1703 /*!
1704  * \internal
1705  * \brief Subscriptions handler for application [un]subscribing.
1706  *
1707  * \param app_name Name of the application to subscribe.
1708  * \param event_source_uris URIs for the event sources to subscribe to.
1709  * \param event_sources_count Array size of event_source_uris.
1710  * \param json Optional output pointer for JSON representation of the app
1711  *             after adding the subscription.
1712  * \param handler [un]subscribe handler
1713  *
1714  * \retval stasis_app_subscribe_res return code.
1715  */
1716 static enum stasis_app_subscribe_res app_handle_subscriptions(
1717         const char *app_name, const char **event_source_uris,
1718         int event_sources_count, struct ast_json **json,
1719         app_subscription_handler handler)
1720 {
1721         RAII_VAR(struct stasis_app *, app, find_app_by_name(app_name), ao2_cleanup);
1722         int i;
1723
1724         if (!app) {
1725                 return STASIS_ASR_APP_NOT_FOUND;
1726         }
1727
1728         for (i = 0; i < event_sources_count; ++i) {
1729                 const char *uri = event_source_uris[i];
1730                 enum stasis_app_subscribe_res res = STASIS_ASR_INTERNAL_ERROR;
1731                 struct stasis_app_event_source *event_source;
1732
1733                 if (!(event_source = app_event_source_find(uri))) {
1734                         ast_log(LOG_WARNING, "Invalid scheme: %s\n", uri);
1735                         return STASIS_ASR_EVENT_SOURCE_BAD_SCHEME;
1736                 }
1737
1738                 if (handler &&
1739                     ((res = handler(app, uri, event_source)))) {
1740                         return res;
1741                 }
1742         }
1743
1744         if (json) {
1745                 ast_debug(3, "%s: Successful; setting results\n", app_name);
1746                 *json = stasis_app_object_to_json(app);
1747         }
1748
1749         return STASIS_ASR_OK;
1750 }
1751
1752 enum stasis_app_subscribe_res stasis_app_subscribe_channel(const char *app_name,
1753         struct ast_channel *chan)
1754 {
1755         RAII_VAR(struct stasis_app *, app, find_app_by_name(app_name), ao2_cleanup);
1756         int res;
1757
1758         if (!app) {
1759                 return STASIS_ASR_APP_NOT_FOUND;
1760         }
1761
1762         ast_debug(3, "%s: Subscribing to %s\n", app_name, ast_channel_uniqueid(chan));
1763
1764         res = app_subscribe_channel(app, chan);
1765         if (res != 0) {
1766                 ast_log(LOG_ERROR, "Error subscribing app '%s' to channel '%s'\n",
1767                         app_name, ast_channel_uniqueid(chan));
1768                 return STASIS_ASR_INTERNAL_ERROR;
1769         }
1770
1771         return STASIS_ASR_OK;
1772 }
1773
1774
1775 /*!
1776  * \internal
1777  * \brief Subscribe an app to an event source.
1778  *
1779  * \param app subscribing application
1780  * \param uri scheme:id of an event source
1781  * \param event_source being subscribed to
1782  *
1783  * \retval stasis_app_subscribe_res return code.
1784  */
1785 static enum stasis_app_subscribe_res app_subscribe(
1786         struct stasis_app *app, const char *uri,
1787         struct stasis_app_event_source *event_source)
1788 {
1789         const char *app_name = stasis_app_name(app);
1790         RAII_VAR(void *, obj, NULL, ao2_cleanup);
1791
1792         ast_debug(3, "%s: Checking %s\n", app_name, uri);
1793
1794         if (!ast_strlen_zero(uri + strlen(event_source->scheme)) &&
1795             (!event_source->find || (!(obj = event_source->find(app, uri + strlen(event_source->scheme)))))) {
1796                 ast_log(LOG_WARNING, "Event source not found: %s\n", uri);
1797                 return STASIS_ASR_EVENT_SOURCE_NOT_FOUND;
1798         }
1799
1800         ast_debug(3, "%s: Subscribing to %s\n", app_name, uri);
1801
1802         if (!event_source->subscribe || (event_source->subscribe(app, obj))) {
1803                 ast_log(LOG_WARNING, "Error subscribing app '%s' to '%s'\n",
1804                         app_name, uri);
1805                 return STASIS_ASR_INTERNAL_ERROR;
1806         }
1807
1808         return STASIS_ASR_OK;
1809 }
1810
1811 enum stasis_app_subscribe_res stasis_app_subscribe(const char *app_name,
1812         const char **event_source_uris, int event_sources_count,
1813         struct ast_json **json)
1814 {
1815         return app_handle_subscriptions(
1816                 app_name, event_source_uris, event_sources_count,
1817                 json, app_subscribe);
1818 }
1819
1820 /*!
1821  * \internal
1822  * \brief Unsubscribe an app from an event source.
1823  *
1824  * \param app application to unsubscribe
1825  * \param uri scheme:id of an event source
1826  * \param event_source being unsubscribed from
1827  *
1828  * \retval stasis_app_subscribe_res return code.
1829  */
1830 static enum stasis_app_subscribe_res app_unsubscribe(
1831         struct stasis_app *app, const char *uri,
1832         struct stasis_app_event_source *event_source)
1833 {
1834         const char *app_name = stasis_app_name(app);
1835         const char *id = uri + strlen(event_source->scheme);
1836
1837         if (!event_source->is_subscribed ||
1838             (!event_source->is_subscribed(app, id))) {
1839                 return STASIS_ASR_EVENT_SOURCE_NOT_FOUND;
1840         }
1841
1842         ast_debug(3, "%s: Unsubscribing from %s\n", app_name, uri);
1843
1844         if (!event_source->unsubscribe || (event_source->unsubscribe(app, id))) {
1845                 ast_log(LOG_WARNING, "Error unsubscribing app '%s' to '%s'\n",
1846                         app_name, uri);
1847                 return -1;
1848         }
1849         return 0;
1850 }
1851
1852 enum stasis_app_subscribe_res stasis_app_unsubscribe(const char *app_name,
1853         const char **event_source_uris, int event_sources_count,
1854         struct ast_json **json)
1855 {
1856         return app_handle_subscriptions(
1857                 app_name, event_source_uris, event_sources_count,
1858                 json, app_unsubscribe);
1859 }
1860
1861 enum stasis_app_user_event_res stasis_app_user_event(const char *app_name,
1862         const char *event_name,
1863         const char **source_uris, int sources_count,
1864         struct ast_json *json_variables)
1865 {
1866         RAII_VAR(struct stasis_app *, app, find_app_by_name(app_name), ao2_cleanup);
1867         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
1868         RAII_VAR(struct ast_multi_object_blob *, multi, NULL, ao2_cleanup);
1869         RAII_VAR(void *, obj, NULL, ao2_cleanup);
1870         RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
1871         enum stasis_app_user_event_res res = STASIS_APP_USER_INTERNAL_ERROR;
1872         struct ast_json *json_value;
1873         int have_channel = 0;
1874         int i;
1875
1876         if (!app) {
1877                 ast_log(LOG_WARNING, "App %s not found\n", app_name);
1878                 return STASIS_APP_USER_APP_NOT_FOUND;
1879         }
1880
1881         if (!ast_multi_user_event_type()) {
1882                 return res;
1883         }
1884
1885         blob = json_variables;
1886         if (!blob) {
1887                 blob = ast_json_pack("{}");
1888         } else {
1889                 ast_json_ref(blob);
1890         }
1891         json_value = ast_json_string_create(event_name);
1892         if (!json_value) {
1893                 ast_log(LOG_ERROR, "unable to create json string\n");
1894                 return res;
1895         }
1896         if (ast_json_object_set(blob, "eventname", json_value)) {
1897                 ast_log(LOG_ERROR, "unable to set eventname to blob\n");
1898                 return res;
1899         }
1900
1901         multi = ast_multi_object_blob_create(blob);
1902
1903         for (i = 0; i < sources_count; ++i) {
1904                 const char *uri = source_uris[i];
1905                 void *snapshot=NULL;
1906                 enum stasis_user_multi_object_snapshot_type type;
1907
1908                 if (ast_begins_with(uri, "channel:")) {
1909                         type = STASIS_UMOS_CHANNEL;
1910                         snapshot = ast_channel_snapshot_get_latest(uri + 8);
1911                         have_channel = 1;
1912                 } else if (ast_begins_with(uri, "bridge:")) {
1913                         type = STASIS_UMOS_BRIDGE;
1914                         snapshot = ast_bridge_snapshot_get_latest(uri + 7);
1915                 } else if (ast_begins_with(uri, "endpoint:")) {
1916                         type = STASIS_UMOS_ENDPOINT;
1917                         snapshot = ast_endpoint_latest_snapshot(uri + 9, NULL);
1918                 } else {
1919                         ast_log(LOG_WARNING, "Invalid scheme: %s\n", uri);
1920                         return STASIS_APP_USER_EVENT_SOURCE_BAD_SCHEME;
1921                 }
1922                 if (!snapshot) {
1923                         ast_log(LOG_ERROR, "Unable to get snapshot for %s\n", uri);
1924                         return STASIS_APP_USER_EVENT_SOURCE_NOT_FOUND;
1925                 }
1926                 ast_multi_object_blob_add(multi, type, snapshot);
1927         }
1928
1929         message = stasis_message_create(ast_multi_user_event_type(), multi);
1930         if (!message) {
1931                 ast_log(LOG_ERROR, "Unable to create stasis user event message\n");
1932                 return res;
1933         }
1934
1935         /*
1936          * Publishing to two different topics is normally to be avoided -- except
1937          * in this case both are final destinations with no forwards (only listeners).
1938          * The message has to be delivered to the application topic for ARI, but a
1939          * copy is also delivered directly to the manager for AMI if there is a channel.
1940          */
1941         stasis_publish(ast_app_get_topic(app), message);
1942
1943         if (have_channel) {
1944                 stasis_publish(ast_manager_get_topic(), message);
1945         }
1946
1947         return STASIS_APP_USER_OK;
1948 }
1949
1950 void stasis_app_ref(void)
1951 {
1952         ast_module_ref(ast_module_info->self);
1953 }
1954
1955 void stasis_app_unref(void)
1956 {
1957         ast_module_unref(ast_module_info->self);
1958 }
1959
1960 static int unload_module(void)
1961 {
1962         stasis_app_unregister_event_sources();
1963
1964         messaging_cleanup();
1965
1966         cleanup();
1967         ao2_cleanup(apps_registry);
1968         apps_registry = NULL;
1969
1970         ao2_cleanup(app_controls);
1971         app_controls = NULL;
1972
1973         ao2_cleanup(app_bridges);
1974         app_bridges = NULL;
1975
1976         ao2_cleanup(app_bridges_moh);
1977         app_bridges_moh = NULL;
1978
1979         ao2_cleanup(app_bridges_playback);
1980         app_bridges_playback = NULL;
1981
1982         stasis_app_control_shutdown();
1983
1984         STASIS_MESSAGE_TYPE_CLEANUP(end_message_type);
1985         STASIS_MESSAGE_TYPE_CLEANUP(start_message_type);
1986
1987         return 0;
1988 }
1989
1990 /* \brief Sanitization callback for channel snapshots */
1991 static int channel_snapshot_sanitizer(const struct ast_channel_snapshot *snapshot)
1992 {
1993         if (!snapshot || !(snapshot->tech_properties & AST_CHAN_TP_INTERNAL)) {
1994                 return 0;
1995         }
1996         return 1;
1997 }
1998
1999 /* \brief Sanitization callback for channels */
2000 static int channel_sanitizer(const struct ast_channel *chan)
2001 {
2002         if (!chan || !(ast_channel_tech(chan)->properties & AST_CHAN_TP_INTERNAL)) {
2003                 return 0;
2004         }
2005         return 1;
2006 }
2007
2008 /* \brief Sanitization callback for channel unique IDs */
2009 static int channel_id_sanitizer(const char *id)
2010 {
2011         RAII_VAR(struct ast_channel_snapshot *, snapshot, ast_channel_snapshot_get_latest(id), ao2_cleanup);
2012
2013         return channel_snapshot_sanitizer(snapshot);
2014 }
2015
2016 /* \brief Sanitization callbacks for communication to Stasis applications */
2017 struct stasis_message_sanitizer app_sanitizer = {
2018         .channel_id = channel_id_sanitizer,
2019         .channel_snapshot = channel_snapshot_sanitizer,
2020         .channel = channel_sanitizer,
2021 };
2022
2023 struct stasis_message_sanitizer *stasis_app_get_sanitizer(void)
2024 {
2025         return &app_sanitizer;
2026 }
2027
2028 static const struct ast_datastore_info stasis_internal_channel_info = {
2029         .type = "stasis-internal-channel",
2030 };
2031
2032 static int set_internal_datastore(struct ast_channel *chan)
2033 {
2034         struct ast_datastore *datastore;
2035
2036         datastore = ast_channel_datastore_find(chan, &stasis_internal_channel_info, NULL);
2037         if (!datastore) {
2038                 datastore = ast_datastore_alloc(&stasis_internal_channel_info, NULL);
2039                 if (!datastore) {
2040                         return -1;
2041                 }
2042                 ast_channel_datastore_add(chan, datastore);
2043         }
2044         return 0;
2045 }
2046
2047 int stasis_app_channel_unreal_set_internal(struct ast_channel *chan)
2048 {
2049         struct ast_channel *outchan = NULL, *outowner = NULL;
2050         int res = 0;
2051         struct ast_unreal_pvt *unreal_pvt = ast_channel_tech_pvt(chan);
2052
2053         ao2_ref(unreal_pvt, +1);
2054         ast_unreal_lock_all(unreal_pvt, &outowner, &outchan);
2055         if (outowner) {
2056                 res |= set_internal_datastore(outowner);
2057                 ast_channel_unlock(outowner);
2058                 ast_channel_unref(outowner);
2059         }
2060         if (outchan) {
2061                 res |= set_internal_datastore(outchan);
2062                 ast_channel_unlock(outchan);
2063                 ast_channel_unref(outchan);
2064         }
2065         ao2_unlock(unreal_pvt);
2066         ao2_ref(unreal_pvt, -1);
2067         return res;
2068 }
2069
2070 int stasis_app_channel_set_internal(struct ast_channel *chan)
2071 {
2072         int res;
2073
2074         ast_channel_lock(chan);
2075         res = set_internal_datastore(chan);
2076         ast_channel_unlock(chan);
2077
2078         return res;
2079 }
2080
2081 int stasis_app_channel_is_internal(struct ast_channel *chan)
2082 {
2083         struct ast_datastore *datastore;
2084         int res = 0;
2085
2086         ast_channel_lock(chan);
2087         datastore = ast_channel_datastore_find(chan, &stasis_internal_channel_info, NULL);
2088         if (datastore) {
2089                 res = 1;
2090         }
2091         ast_channel_unlock(chan);
2092
2093         return res;
2094 }
2095
2096 static int load_module(void)
2097 {
2098         if (STASIS_MESSAGE_TYPE_INIT(start_message_type) != 0) {
2099                 return AST_MODULE_LOAD_DECLINE;
2100         }
2101         if (STASIS_MESSAGE_TYPE_INIT(end_message_type) != 0) {
2102                 return AST_MODULE_LOAD_DECLINE;
2103         }
2104         apps_registry = ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
2105         app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS, control_hash, control_compare);
2106         app_bridges = ao2_container_alloc(BRIDGES_NUM_BUCKETS, bridges_hash, bridges_compare);
2107         app_bridges_moh = ao2_container_alloc_hash(
2108                 AO2_ALLOC_OPT_LOCK_MUTEX, AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT,
2109                 37, bridges_channel_hash_fn, bridges_channel_sort_fn, NULL);
2110         app_bridges_playback = ao2_container_alloc_hash(
2111                 AO2_ALLOC_OPT_LOCK_MUTEX, AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT,
2112                 37, bridges_channel_hash_fn, bridges_channel_sort_fn, NULL);
2113         if (!apps_registry || !app_controls || !app_bridges || !app_bridges_moh || !app_bridges_playback) {
2114                 unload_module();
2115                 return AST_MODULE_LOAD_FAILURE;
2116         }
2117
2118         if (messaging_init()) {
2119                 unload_module();
2120                 return AST_MODULE_LOAD_FAILURE;
2121         }
2122
2123         bridge_stasis_init();
2124
2125         stasis_app_register_event_sources();
2126
2127         return AST_MODULE_LOAD_SUCCESS;
2128 }
2129
2130 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application support",
2131         .load_pri = AST_MODPRI_APP_DEPEND,
2132         .support_level = AST_MODULE_SUPPORT_CORE,
2133         .load = load_module,
2134         .unload = unload_module,
2135 );