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