res_resolver_unbound: Fix config documentation.
[asterisk/asterisk.git] / 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 );