loader: Add dependency fields to module structures.
[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         struct ast_channel *chan;
502         struct ast_format_cap *cap;
503
504         cap = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
505         if (!cap) {
506                 return NULL;
507         }
508
509         ast_format_cap_append(cap, ast_format_slin, 0);
510
511         chan = ast_request("Announcer", cap, NULL, NULL, "ARI_MOH", NULL);
512         ao2_ref(cap, -1);
513
514         return chan;
515 }
516
517 /*! Provides the moh channel with a thread so it can actually play its music */
518 static void *moh_channel_thread(void *data)
519 {
520         struct ast_channel *moh_channel = data;
521
522         while (!ast_safe_sleep(moh_channel, 1000)) {
523         }
524
525         ast_moh_stop(moh_channel);
526         ast_hangup(moh_channel);
527
528         return NULL;
529 }
530
531 /*!
532  * \internal
533  * \brief Creates, pushes, and links a channel for playing music on hold to bridge
534  *
535  * \param bridge Which bridge this moh channel exists for
536  *
537  * \retval NULL if the channel could not be created, pushed, or linked
538  * \retval Reference to the channel on success
539  */
540 static struct ast_channel *bridge_moh_create(struct ast_bridge *bridge)
541 {
542         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, new_wrapper, NULL, ao2_cleanup);
543         RAII_VAR(char *, bridge_id, ast_strdup(bridge->uniqueid), ast_free);
544         struct ast_channel *chan;
545         pthread_t threadid;
546
547         if (!bridge_id) {
548                 return NULL;
549         }
550
551         chan = prepare_bridge_moh_channel();
552         if (!chan) {
553                 return NULL;
554         }
555
556         if (stasis_app_channel_unreal_set_internal(chan)) {
557                 ast_hangup(chan);
558                 return NULL;
559         }
560
561         /* The after bridge callback assumes responsibility of the bridge_id. */
562         if (ast_bridge_set_after_callback(chan,
563                 moh_after_bridge_cb, moh_after_bridge_cb_failed, bridge_id)) {
564                 ast_hangup(chan);
565                 return NULL;
566         }
567         bridge_id = NULL;
568
569         if (ast_unreal_channel_push_to_bridge(chan, bridge,
570                 AST_BRIDGE_CHANNEL_FLAG_IMMOVABLE | AST_BRIDGE_CHANNEL_FLAG_LONELY)) {
571                 ast_hangup(chan);
572                 return NULL;
573         }
574
575         new_wrapper = ao2_alloc_options(sizeof(*new_wrapper),
576                 stasis_app_bridge_channel_wrapper_destructor, AO2_ALLOC_OPT_LOCK_NOLOCK);
577         if (!new_wrapper) {
578                 ast_hangup(chan);
579                 return NULL;
580         }
581
582         if (ast_string_field_init(new_wrapper, 32)) {
583                 ast_hangup(chan);
584                 return NULL;
585         }
586         ast_string_field_set(new_wrapper, bridge_id, bridge->uniqueid);
587         ast_string_field_set(new_wrapper, channel_id, ast_channel_uniqueid(chan));
588
589         if (!ao2_link_flags(app_bridges_moh, new_wrapper, OBJ_NOLOCK)) {
590                 ast_hangup(chan);
591                 return NULL;
592         }
593
594         if (ast_pthread_create_detached(&threadid, NULL, moh_channel_thread, chan)) {
595                 ast_log(LOG_ERROR, "Failed to create channel thread. Abandoning MOH channel creation.\n");
596                 ao2_unlink_flags(app_bridges_moh, new_wrapper, OBJ_NOLOCK);
597                 ast_hangup(chan);
598                 return NULL;
599         }
600
601         return chan;
602 }
603
604 struct ast_channel *stasis_app_bridge_moh_channel(struct ast_bridge *bridge)
605 {
606         struct ast_channel *chan;
607         struct stasis_app_bridge_channel_wrapper *moh_wrapper;
608
609         ao2_lock(app_bridges_moh);
610         moh_wrapper = ao2_find(app_bridges_moh, bridge->uniqueid, OBJ_SEARCH_KEY | OBJ_NOLOCK);
611         if (!moh_wrapper) {
612                 chan = bridge_moh_create(bridge);
613         }
614         ao2_unlock(app_bridges_moh);
615
616         if (moh_wrapper) {
617                 chan = ast_channel_get_by_name(moh_wrapper->channel_id);
618                 ao2_ref(moh_wrapper, -1);
619         }
620
621         return chan;
622 }
623
624 int stasis_app_bridge_moh_stop(struct ast_bridge *bridge)
625 {
626         struct stasis_app_bridge_channel_wrapper *moh_wrapper;
627         struct ast_channel *chan;
628
629         moh_wrapper = ao2_find(app_bridges_moh, bridge->uniqueid, OBJ_SEARCH_KEY | OBJ_UNLINK);
630         if (!moh_wrapper) {
631                 return -1;
632         }
633
634         chan = ast_channel_get_by_name(moh_wrapper->channel_id);
635         ao2_ref(moh_wrapper, -1);
636         if (!chan) {
637                 return -1;
638         }
639
640         ast_moh_stop(chan);
641         ast_softhangup(chan, AST_CAUSE_NORMAL_CLEARING);
642         ao2_cleanup(chan);
643
644         return 0;
645 }
646
647 /*! Removes the bridge to playback channel link */
648 static void remove_bridge_playback(char *bridge_id)
649 {
650         struct stasis_app_bridge_channel_wrapper *wrapper;
651         struct stasis_app_control *control;
652
653         wrapper = ao2_find(app_bridges_playback, bridge_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
654
655         if (wrapper) {
656                 control = stasis_app_control_find_by_channel_id(wrapper->channel_id);
657                 if (control) {
658                         ao2_unlink(app_controls, control);
659                         ao2_ref(control, -1);
660                 }
661                 ao2_ref(wrapper, -1);
662         }
663         ast_free(bridge_id);
664 }
665
666 static void playback_after_bridge_cb_failed(enum ast_bridge_after_cb_reason reason, void *data)
667 {
668         char *bridge_id = data;
669
670         remove_bridge_playback(bridge_id);
671 }
672
673 static void playback_after_bridge_cb(struct ast_channel *chan, void *data)
674 {
675         char *bridge_id = data;
676
677         remove_bridge_playback(bridge_id);
678 }
679
680 int stasis_app_bridge_playback_channel_add(struct ast_bridge *bridge,
681         struct ast_channel *chan,
682         struct stasis_app_control *control)
683 {
684         RAII_VAR(struct stasis_app_bridge_channel_wrapper *, new_wrapper, NULL, ao2_cleanup);
685         char *bridge_id = ast_strdup(bridge->uniqueid);
686
687         if (!bridge_id) {
688                 return -1;
689         }
690
691         if (ast_bridge_set_after_callback(chan,
692                 playback_after_bridge_cb, playback_after_bridge_cb_failed, bridge_id)) {
693                 ast_free(bridge_id);
694                 return -1;
695         }
696
697         new_wrapper = ao2_alloc_options(sizeof(*new_wrapper),
698                 stasis_app_bridge_channel_wrapper_destructor, AO2_ALLOC_OPT_LOCK_NOLOCK);
699         if (!new_wrapper) {
700                 return -1;
701         }
702
703         if (ast_string_field_init(new_wrapper, 32)) {
704                 return -1;
705         }
706
707         ast_string_field_set(new_wrapper, bridge_id, bridge->uniqueid);
708         ast_string_field_set(new_wrapper, channel_id, ast_channel_uniqueid(chan));
709
710         if (!ao2_link(app_bridges_playback, new_wrapper)) {
711                 return -1;
712         }
713
714         ao2_link(app_controls, control);
715         return 0;
716 }
717
718 void stasis_app_bridge_playback_channel_remove(char *bridge_id,
719         struct stasis_app_control *control)
720 {
721         struct stasis_app_bridge_channel_wrapper *wrapper;
722
723         wrapper = ao2_find(app_bridges_playback, bridge_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
724         if (wrapper) {
725                 /* If wrapper is not found, then that means the after bridge callback has been
726                  * called or is in progress. No need to unlink the control here since that has
727                  * been done or is about to be done in the after bridge callback
728                  */
729                 ao2_unlink(app_controls, control);
730                 ao2_ref(wrapper, -1);
731         }
732 }
733
734 struct ast_channel *stasis_app_bridge_playback_channel_find(struct ast_bridge *bridge)
735 {
736         struct stasis_app_bridge_channel_wrapper *playback_wrapper;
737         struct ast_channel *chan;
738
739         playback_wrapper = ao2_find(app_bridges_playback, bridge->uniqueid, OBJ_SEARCH_KEY);
740         if (!playback_wrapper) {
741                 return NULL;
742         }
743
744         chan = ast_channel_get_by_name(playback_wrapper->channel_id);
745         ao2_ref(playback_wrapper, -1);
746         return chan;
747 }
748
749 struct ast_bridge *stasis_app_bridge_find_by_id(
750         const char *bridge_id)
751 {
752         return ao2_find(app_bridges, bridge_id, OBJ_SEARCH_KEY);
753 }
754
755
756 /*!
757  * \brief In addition to running ao2_cleanup(), this function also removes the
758  * object from the app_controls container.
759  */
760 static void control_unlink(struct stasis_app_control *control)
761 {
762         if (!control) {
763                 return;
764         }
765
766         ao2_unlink(app_controls, control);
767         ao2_cleanup(control);
768 }
769
770 static struct ast_bridge *bridge_create_common(const char *type, const char *name, const char *id, int invisible)
771 {
772         struct ast_bridge *bridge;
773         char *requested_type, *requested_types = ast_strdupa(S_OR(type, "mixing"));
774         int capabilities = 0;
775         int flags = AST_BRIDGE_FLAG_MERGE_INHIBIT_FROM | AST_BRIDGE_FLAG_MERGE_INHIBIT_TO
776                 | AST_BRIDGE_FLAG_SWAP_INHIBIT_FROM | AST_BRIDGE_FLAG_SWAP_INHIBIT_TO
777                 | AST_BRIDGE_FLAG_TRANSFER_BRIDGE_ONLY;
778         enum ast_bridge_video_mode_type video_mode = AST_BRIDGE_VIDEO_MODE_TALKER_SRC;
779
780         if (invisible) {
781                 flags |= AST_BRIDGE_FLAG_INVISIBLE;
782         }
783
784         while ((requested_type = strsep(&requested_types, ","))) {
785                 requested_type = ast_strip(requested_type);
786
787                 if (!strcmp(requested_type, "mixing")) {
788                         capabilities |= STASIS_BRIDGE_MIXING_CAPABILITIES;
789                         flags |= AST_BRIDGE_FLAG_SMART;
790                 } else if (!strcmp(requested_type, "holding")) {
791                         capabilities |= AST_BRIDGE_CAPABILITY_HOLDING;
792                 } else if (!strcmp(requested_type, "dtmf_events") ||
793                         !strcmp(requested_type, "proxy_media")) {
794                         capabilities &= ~AST_BRIDGE_CAPABILITY_NATIVE;
795                 } else if (!strcmp(requested_type, "video_sfu")) {
796                         video_mode = AST_BRIDGE_VIDEO_MODE_SFU;
797                 }
798         }
799
800         /* For an SFU video bridge we ensure it always remains in multimix for the best experience. */
801         if (video_mode == AST_BRIDGE_VIDEO_MODE_SFU) {
802                 capabilities = AST_BRIDGE_CAPABILITY_MULTIMIX;
803                 flags &= ~AST_BRIDGE_FLAG_SMART;
804         }
805
806         if (!capabilities
807                 /* Holding and mixing capabilities don't mix. */
808                 || ((capabilities & AST_BRIDGE_CAPABILITY_HOLDING)
809                         && (capabilities & (STASIS_BRIDGE_MIXING_CAPABILITIES)))) {
810                 return NULL;
811         }
812
813         bridge = bridge_stasis_new(capabilities, flags, name, id);
814         if (bridge) {
815                 if (video_mode == AST_BRIDGE_VIDEO_MODE_SFU) {
816                         ast_bridge_set_sfu_video_mode(bridge);
817                         /* We require a minimum 5 seconds between video updates to stop floods from clients,
818                          * this should rarely be changed but should become configurable in the future.
819                          */
820                         ast_bridge_set_video_update_discard(bridge, 5);
821                 } else {
822                         ast_bridge_set_talker_src_video_mode(bridge);
823                 }
824                 if (!ao2_link(app_bridges, bridge)) {
825                         ast_bridge_destroy(bridge, 0);
826                         bridge = NULL;
827                 }
828         }
829         return bridge;
830 }
831
832 struct ast_bridge *stasis_app_bridge_create(const char *type, const char *name, const char *id)
833 {
834         return bridge_create_common(type, name, id, 0);
835 }
836
837 struct ast_bridge *stasis_app_bridge_create_invisible(const char *type, const char *name, const char *id)
838 {
839         return bridge_create_common(type, name, id, 1);
840 }
841
842 void stasis_app_bridge_destroy(const char *bridge_id)
843 {
844         struct ast_bridge *bridge = stasis_app_bridge_find_by_id(bridge_id);
845         if (!bridge) {
846                 return;
847         }
848         ao2_unlink(app_bridges, bridge);
849         ast_bridge_destroy(bridge, 0);
850 }
851
852 struct replace_channel_store {
853         struct ast_channel_snapshot *snapshot;
854         char *app;
855 };
856
857 static void replace_channel_destroy(void *obj)
858 {
859         struct replace_channel_store *replace = obj;
860
861         ao2_cleanup(replace->snapshot);
862         ast_free(replace->app);
863         ast_free(replace);
864 }
865
866 static const struct ast_datastore_info replace_channel_store_info = {
867         .type = "replace-channel-store",
868         .destroy = replace_channel_destroy,
869 };
870
871 static struct replace_channel_store *get_replace_channel_store(struct ast_channel *chan, int no_create)
872 {
873         struct ast_datastore *datastore;
874         struct replace_channel_store *ret;
875
876         ast_channel_lock(chan);
877         datastore = ast_channel_datastore_find(chan, &replace_channel_store_info, NULL);
878         if (!datastore && !no_create) {
879                 datastore = ast_datastore_alloc(&replace_channel_store_info, NULL);
880                 if (datastore) {
881                         ast_channel_datastore_add(chan, datastore);
882                 }
883         }
884
885         if (!datastore) {
886                 ast_channel_unlock(chan);
887                 return NULL;
888         }
889
890         if (!datastore->data) {
891                 datastore->data = ast_calloc(1, sizeof(struct replace_channel_store));
892         }
893
894         ret = datastore->data;
895         ast_channel_unlock(chan);
896
897         return ret;
898 }
899
900 int app_set_replace_channel_snapshot(struct ast_channel *chan, struct ast_channel_snapshot *replace_snapshot)
901 {
902         struct replace_channel_store *replace = get_replace_channel_store(chan, 0);
903
904         if (!replace) {
905                 return -1;
906         }
907
908         ao2_replace(replace->snapshot, replace_snapshot);
909         return 0;
910 }
911
912 int app_set_replace_channel_app(struct ast_channel *chan, const char *replace_app)
913 {
914         struct replace_channel_store *replace = get_replace_channel_store(chan, 0);
915
916         if (!replace) {
917                 return -1;
918         }
919
920         ast_free(replace->app);
921         replace->app = NULL;
922
923         if (replace_app) {
924                 replace->app = ast_strdup(replace_app);
925                 if (!replace->app) {
926                         return -1;
927                 }
928         }
929
930         return 0;
931 }
932
933 static struct ast_channel_snapshot *get_replace_channel_snapshot(struct ast_channel *chan)
934 {
935         struct replace_channel_store *replace = get_replace_channel_store(chan, 1);
936         struct ast_channel_snapshot *replace_channel_snapshot;
937
938         if (!replace) {
939                 return NULL;
940         }
941
942         replace_channel_snapshot = replace->snapshot;
943         replace->snapshot = NULL;
944
945         return replace_channel_snapshot;
946 }
947
948 char *app_get_replace_channel_app(struct ast_channel *chan)
949 {
950         struct replace_channel_store *replace = get_replace_channel_store(chan, 1);
951         char *replace_channel_app;
952
953         if (!replace) {
954                 return NULL;
955         }
956
957         replace_channel_app = replace->app;
958         replace->app = NULL;
959
960         return replace_channel_app;
961 }
962
963 static void start_message_blob_dtor(void *obj)
964 {
965         struct start_message_blob *payload = obj;
966
967         ao2_cleanup(payload->channel);
968         ao2_cleanup(payload->replace_channel);
969         ast_json_unref(payload->blob);
970 }
971
972 static int send_start_msg_snapshots(struct ast_channel *chan, struct stasis_app *app,
973         int argc, char *argv[], struct ast_channel_snapshot *snapshot,
974         struct ast_channel_snapshot *replace_channel_snapshot)
975 {
976         struct ast_json *json_blob;
977         struct ast_json *json_args;
978         struct start_message_blob *payload;
979         struct stasis_message *msg;
980         int i;
981
982         if (app_subscribe_channel(app, chan)) {
983                 ast_log(LOG_ERROR, "Error subscribing app '%s' to channel '%s'\n",
984                         stasis_app_name(app), ast_channel_name(chan));
985                 return -1;
986         }
987
988         payload = ao2_alloc(sizeof(*payload), start_message_blob_dtor);
989         if (!payload) {
990                 ast_log(LOG_ERROR, "Error packing JSON for StasisStart message\n");
991                 return -1;
992         }
993
994         payload->channel = ao2_bump(snapshot);
995         payload->replace_channel = ao2_bump(replace_channel_snapshot);
996
997         json_blob = ast_json_pack("{s: s, s: o, s: []}",
998                 "app", stasis_app_name(app),
999                 "timestamp", ast_json_timeval(ast_tvnow(), NULL),
1000                 "args");
1001         if (!json_blob) {
1002                 ast_log(LOG_ERROR, "Error packing JSON for StasisStart message\n");
1003                 ao2_ref(payload, -1);
1004                 return -1;
1005         }
1006         payload->blob = json_blob;
1007
1008
1009         /* Append arguments to args array */
1010         json_args = ast_json_object_get(json_blob, "args");
1011         ast_assert(json_args != NULL);
1012         for (i = 0; i < argc; ++i) {
1013                 int r = ast_json_array_append(json_args,
1014                                               ast_json_string_create(argv[i]));
1015                 if (r != 0) {
1016                         ast_log(LOG_ERROR, "Error appending to StasisStart message\n");
1017                         ao2_ref(payload, -1);
1018                         return -1;
1019                 }
1020         }
1021
1022
1023         msg = stasis_message_create(start_message_type(), payload);
1024         ao2_ref(payload, -1);
1025         if (!msg) {
1026                 ast_log(LOG_ERROR, "Error sending StasisStart message\n");
1027                 return -1;
1028         }
1029
1030         if (replace_channel_snapshot) {
1031                 app_unsubscribe_channel_id(app, replace_channel_snapshot->uniqueid);
1032         }
1033         stasis_publish(ast_app_get_topic(app), msg);
1034         ao2_ref(msg, -1);
1035         return 0;
1036 }
1037
1038 static int send_start_msg(struct stasis_app *app, struct ast_channel *chan,
1039         int argc, char *argv[])
1040 {
1041         int ret = -1;
1042         struct ast_channel_snapshot *snapshot;
1043         struct ast_channel_snapshot *replace_channel_snapshot;
1044
1045         ast_assert(chan != NULL);
1046
1047         replace_channel_snapshot = get_replace_channel_snapshot(chan);
1048
1049         /* Set channel info */
1050         ast_channel_lock(chan);
1051         snapshot = ast_channel_snapshot_create(chan);
1052         ast_channel_unlock(chan);
1053         if (snapshot) {
1054                 ret = send_start_msg_snapshots(chan, app, argc, argv, snapshot, replace_channel_snapshot);
1055                 ao2_ref(snapshot, -1);
1056         }
1057         ao2_cleanup(replace_channel_snapshot);
1058
1059         return ret;
1060 }
1061
1062 static void remove_masquerade_store(struct ast_channel *chan);
1063
1064 int app_send_end_msg(struct stasis_app *app, struct ast_channel *chan)
1065 {
1066         struct stasis_message_sanitizer *sanitize = stasis_app_get_sanitizer();
1067         struct ast_json *blob;
1068         struct stasis_message *msg;
1069
1070         if (sanitize && sanitize->channel
1071                 && sanitize->channel(chan)) {
1072                 return 0;
1073         }
1074
1075         blob = ast_json_pack("{s: s}", "app", stasis_app_name(app));
1076         if (!blob) {
1077                 ast_log(LOG_ERROR, "Error packing JSON for StasisEnd message\n");
1078                 return -1;
1079         }
1080
1081         remove_masquerade_store(chan);
1082         app_unsubscribe_channel(app, chan);
1083         msg = ast_channel_blob_create(chan, end_message_type(), blob);
1084         if (msg) {
1085                 stasis_publish(ast_app_get_topic(app), msg);
1086         }
1087         ao2_cleanup(msg);
1088         ast_json_unref(blob);
1089
1090         return 0;
1091 }
1092
1093 static int masq_match_cb(void *obj, void *data, int flags)
1094 {
1095         struct stasis_app_control *control = obj;
1096         struct ast_channel *chan = data;
1097
1098         if (!strcmp(ast_channel_uniqueid(chan),
1099                 stasis_app_control_get_channel_id(control))) {
1100                 return CMP_MATCH;
1101         }
1102
1103         return 0;
1104 }
1105
1106 static void channel_stolen_cb(void *data, struct ast_channel *old_chan, struct ast_channel *new_chan)
1107 {
1108         struct stasis_app_control *control;
1109
1110         /*
1111          * At this point, old_chan is the channel pointer that is in Stasis() and
1112          * has the unknown channel's name in it while new_chan is the channel pointer
1113          * that is not in Stasis(), but has the guts of the channel that Stasis() knows
1114          * about.
1115          *
1116          * Find and unlink control since the channel has a new name/uniqueid
1117          * and its hash has changed.  Since the channel is leaving stasis don't
1118          * bother putting it back into the container.  Nobody is going to
1119          * remove it from the container later.
1120          */
1121         control = ao2_callback(app_controls, OBJ_UNLINK, masq_match_cb, old_chan);
1122         if (!control) {
1123                 ast_log(LOG_ERROR, "Could not find control for masqueraded channel\n");
1124                 return;
1125         }
1126
1127         /* send the StasisEnd message to the app */
1128         stasis_app_channel_set_stasis_end_published(new_chan);
1129         app_send_end_msg(control_app(control), new_chan);
1130
1131         /* remove the datastore */
1132         remove_masquerade_store(old_chan);
1133
1134         ao2_cleanup(control);
1135 }
1136
1137 static void channel_replaced_cb(void *data, struct ast_channel *old_chan, struct ast_channel *new_chan)
1138 {
1139         RAII_VAR(struct ast_channel_snapshot *, new_snapshot, NULL, ao2_cleanup);
1140         RAII_VAR(struct ast_channel_snapshot *, old_snapshot, NULL, ao2_cleanup);
1141         struct stasis_app_control *control;
1142
1143         /* At this point, new_chan is the channel pointer that is in Stasis() and
1144          * has the unknown channel's name in it while old_chan is the channel pointer
1145          * that is not in Stasis(), but has the guts of the channel that Stasis() knows
1146          * about */
1147
1148         /* grab a snapshot for the channel that is jumping into Stasis() */
1149         new_snapshot = ast_channel_snapshot_get_latest(ast_channel_uniqueid(new_chan));
1150         if (!new_snapshot) {
1151                 ast_log(LOG_ERROR, "Could not get snapshot for masquerading channel\n");
1152                 return;
1153         }
1154
1155         /* grab a snapshot for the channel that has been kicked out of Stasis() */
1156         old_snapshot = ast_channel_snapshot_get_latest(ast_channel_uniqueid(old_chan));
1157         if (!old_snapshot) {
1158                 ast_log(LOG_ERROR, "Could not get snapshot for masqueraded channel\n");
1159                 return;
1160         }
1161
1162         /*
1163          * Find, unlink, and relink control since the channel has a new
1164          * name/uniqueid and its hash has changed.
1165          */
1166         control = ao2_callback(app_controls, OBJ_UNLINK, masq_match_cb, new_chan);
1167         if (!control) {
1168                 ast_log(LOG_ERROR, "Could not find control for masquerading channel\n");
1169                 return;
1170         }
1171         ao2_link(app_controls, control);
1172
1173
1174         /* send the StasisStart with replace_channel to the app */
1175         send_start_msg_snapshots(new_chan, control_app(control), 0, NULL, new_snapshot,
1176                 old_snapshot);
1177         /* send the StasisEnd message to the app */
1178         app_send_end_msg(control_app(control), old_chan);
1179
1180         ao2_cleanup(control);
1181 }
1182
1183 static const struct ast_datastore_info masquerade_store_info = {
1184         .type = "stasis-masqerade",
1185         .chan_fixup = channel_stolen_cb,
1186         .chan_breakdown = channel_replaced_cb,
1187 };
1188
1189 static int has_masquerade_store(struct ast_channel *chan)
1190 {
1191         SCOPED_CHANNELLOCK(lock, chan);
1192         return !!ast_channel_datastore_find(chan, &masquerade_store_info, NULL);
1193 }
1194
1195 static int add_masquerade_store(struct ast_channel *chan)
1196 {
1197         struct ast_datastore *datastore;
1198
1199         SCOPED_CHANNELLOCK(lock, chan);
1200         if (ast_channel_datastore_find(chan, &masquerade_store_info, NULL)) {
1201                 return 0;
1202         }
1203
1204         datastore = ast_datastore_alloc(&masquerade_store_info, NULL);
1205         if (!datastore) {
1206                 return -1;
1207         }
1208
1209         ast_channel_datastore_add(chan, datastore);
1210
1211         return 0;
1212 }
1213
1214 static void remove_masquerade_store(struct ast_channel *chan)
1215 {
1216         struct ast_datastore *datastore;
1217
1218         SCOPED_CHANNELLOCK(lock, chan);
1219         datastore = ast_channel_datastore_find(chan, &masquerade_store_info, NULL);
1220         if (!datastore) {
1221                 return;
1222         }
1223
1224         ast_channel_datastore_remove(chan, datastore);
1225         ast_datastore_free(datastore);
1226 }
1227
1228 void stasis_app_control_execute_until_exhausted(struct ast_channel *chan, struct stasis_app_control *control)
1229 {
1230         while (!control_is_done(control)) {
1231                 int command_count;
1232                 command_count = control_dispatch_all(control, chan);
1233
1234                 ao2_lock(control);
1235
1236                 if (control_command_count(control)) {
1237                         /* If the command queue isn't empty, something added to the queue before it was locked. */
1238                         ao2_unlock(control);
1239                         continue;
1240                 }
1241
1242                 if (command_count == 0 || ast_channel_fdno(chan) == -1) {
1243                         control_mark_done(control);
1244                         ao2_unlock(control);
1245                         break;
1246                 }
1247                 ao2_unlock(control);
1248         }
1249 }
1250
1251 int stasis_app_control_is_done(struct stasis_app_control *control)
1252 {
1253         return control_is_done(control);
1254 }
1255
1256 void stasis_app_control_flush_queue(struct stasis_app_control *control)
1257 {
1258         control_flush_queue(control);
1259 }
1260
1261 struct ast_datastore_info set_end_published_info = {
1262         .type = "stasis_end_published",
1263 };
1264
1265 void stasis_app_channel_set_stasis_end_published(struct ast_channel *chan)
1266 {
1267         struct ast_datastore *datastore;
1268
1269         datastore = ast_datastore_alloc(&set_end_published_info, NULL);
1270         if (datastore) {
1271                 ast_channel_lock(chan);
1272                 ast_channel_datastore_add(chan, datastore);
1273                 ast_channel_unlock(chan);
1274         }
1275 }
1276
1277 int stasis_app_channel_is_stasis_end_published(struct ast_channel *chan)
1278 {
1279         struct ast_datastore *datastore;
1280
1281         ast_channel_lock(chan);
1282         datastore = ast_channel_datastore_find(chan, &set_end_published_info, NULL);
1283         ast_channel_unlock(chan);
1284
1285         return datastore ? 1 : 0;
1286 }
1287
1288 static void remove_stasis_end_published(struct ast_channel *chan)
1289 {
1290         struct ast_datastore *datastore;
1291
1292         ast_channel_lock(chan);
1293         datastore = ast_channel_datastore_find(chan, &set_end_published_info, NULL);
1294         if (datastore) {
1295                 ast_channel_datastore_remove(chan, datastore);
1296                 ast_datastore_free(datastore);
1297         }
1298         ast_channel_unlock(chan);
1299 }
1300
1301 /*! /brief Stasis dialplan application callback */
1302 int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
1303                     char *argv[])
1304 {
1305         SCOPED_MODULE_USE(ast_module_info->self);
1306
1307         RAII_VAR(struct stasis_app *, app, NULL, ao2_cleanup);
1308         RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
1309         struct ast_bridge *bridge = NULL;
1310         int res = 0;
1311         int needs_depart;
1312
1313         ast_assert(chan != NULL);
1314
1315         /* Just in case there's a lingering indication that the channel has had a stasis
1316          * end published on it, remove that now.
1317          */
1318         remove_stasis_end_published(chan);
1319
1320         if (!apps_registry) {
1321                 return -1;
1322         }
1323
1324         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1325         if (!app) {
1326                 ast_log(LOG_ERROR,
1327                         "Stasis app '%s' not registered\n", app_name);
1328                 return -1;
1329         }
1330         if (!app_is_active(app)) {
1331                 ast_log(LOG_ERROR,
1332                         "Stasis app '%s' not active\n", app_name);
1333                 return -1;
1334         }
1335
1336         control = control_create(chan, app);
1337         if (!control) {
1338                 ast_log(LOG_ERROR, "Allocated failed\n");
1339                 return -1;
1340         }
1341         ao2_link(app_controls, control);
1342
1343         if (add_masquerade_store(chan)) {
1344                 ast_log(LOG_ERROR, "Failed to attach masquerade detector\n");
1345                 return -1;
1346         }
1347
1348         res = send_start_msg(app, chan, argc, argv);
1349         if (res != 0) {
1350                 ast_log(LOG_ERROR,
1351                         "Error sending start message to '%s'\n", app_name);
1352                 remove_masquerade_store(chan);
1353                 return -1;
1354         }
1355
1356         /* Pull queued prestart commands and execute */
1357         control_prestart_dispatch_all(control, chan);
1358
1359         while (!control_is_done(control)) {
1360                 RAII_VAR(struct ast_frame *, f, NULL, ast_frame_dtor);
1361                 int r;
1362                 int command_count;
1363                 RAII_VAR(struct ast_bridge *, last_bridge, NULL, ao2_cleanup);
1364
1365                 /* Check to see if a bridge absorbed our hangup frame */
1366                 if (ast_check_hangup_locked(chan)) {
1367                         control_mark_done(control);
1368                         break;
1369                 }
1370
1371                 last_bridge = bridge;
1372                 bridge = ao2_bump(stasis_app_get_bridge(control));
1373
1374                 if (bridge != last_bridge) {
1375                         if (last_bridge) {
1376                                 app_unsubscribe_bridge(app, last_bridge);
1377                         }
1378                         if (bridge) {
1379                                 app_subscribe_bridge(app, bridge);
1380                         }
1381                 }
1382
1383                 if (bridge) {
1384                         /* Bridge/dial is handling channel frames */
1385                         control_wait(control);
1386                         control_dispatch_all(control, chan);
1387                         continue;
1388                 }
1389
1390                 r = ast_waitfor(chan, MAX_WAIT_MS);
1391
1392                 if (r < 0) {
1393                         ast_debug(3, "%s: Poll error\n",
1394                                   ast_channel_uniqueid(chan));
1395                         control_mark_done(control);
1396                         break;
1397                 }
1398
1399                 command_count = control_dispatch_all(control, chan);
1400
1401                 if (command_count > 0 && ast_channel_fdno(chan) == -1) {
1402                         /* Command drained the channel; wait for next frame */
1403                         continue;
1404                 }
1405
1406                 if (r == 0) {
1407                         /* Timeout */
1408                         continue;
1409                 }
1410
1411                 f = ast_read(chan);
1412                 if (!f) {
1413                         /* Continue on in the dialplan */
1414                         ast_debug(3, "%s: Hangup (no more frames)\n",
1415                                 ast_channel_uniqueid(chan));
1416                         control_mark_done(control);
1417                         break;
1418                 }
1419
1420                 if (f->frametype == AST_FRAME_CONTROL) {
1421                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
1422                                 /* Continue on in the dialplan */
1423                                 ast_debug(3, "%s: Hangup\n",
1424                                         ast_channel_uniqueid(chan));
1425                                 control_mark_done(control);
1426                                 break;
1427                         }
1428                 }
1429         }
1430
1431         ast_channel_lock(chan);
1432         needs_depart = (ast_channel_internal_bridge_channel(chan) != NULL);
1433         ast_channel_unlock(chan);
1434         if (needs_depart) {
1435                 ast_bridge_depart(chan);
1436         }
1437
1438         if (stasis_app_get_bridge(control)) {
1439                 app_unsubscribe_bridge(app, stasis_app_get_bridge(control));
1440         }
1441         ao2_cleanup(bridge);
1442
1443         /* Only publish a stasis_end event if it hasn't already been published */
1444         if (!stasis_app_channel_is_stasis_end_published(chan)) {
1445                 /* A masquerade has occurred and this message will be wrong so it
1446                  * has already been sent elsewhere. */
1447                 res = has_masquerade_store(chan) && app_send_end_msg(app, chan);
1448                 if (res != 0) {
1449                         ast_log(LOG_ERROR,
1450                                 "Error sending end message to %s\n", app_name);
1451                         return res;
1452                 }
1453         } else {
1454                 remove_stasis_end_published(chan);
1455         }
1456
1457         control_flush_queue(control);
1458
1459         /* Stop any lingering silence generator */
1460         control_silence_stop_now(control);
1461
1462         /* There's an off chance that app is ready for cleanup. Go ahead
1463          * and clean up, just in case
1464          */
1465         cleanup();
1466
1467         /* The control needs to be removed from the controls container in
1468          * case a new PBX is started and ends up coming back into Stasis.
1469          */
1470         ao2_cleanup(app);
1471         app = NULL;
1472         control_unlink(control);
1473         control = NULL;
1474
1475         if (!ast_channel_pbx(chan)) {
1476                 int chan_hungup;
1477
1478                 /* The ASYNCGOTO softhangup flag may have broken the channel out of
1479                  * its bridge to run dialplan, so if there's no pbx on the channel
1480                  * let it run dialplan here. Otherwise, it will run when this
1481                  * application exits. */
1482                 ast_channel_lock(chan);
1483                 ast_channel_clear_softhangup(chan, AST_SOFTHANGUP_ASYNCGOTO);
1484                 chan_hungup = ast_check_hangup(chan);
1485                 ast_channel_unlock(chan);
1486
1487                 if (!chan_hungup) {
1488                         struct ast_pbx_args pbx_args;
1489
1490                         memset(&pbx_args, 0, sizeof(pbx_args));
1491                         pbx_args.no_hangup_chan = 1;
1492
1493                         res = ast_pbx_run_args(chan, &pbx_args);
1494                 }
1495         }
1496
1497         return res;
1498 }
1499
1500 int stasis_app_send(const char *app_name, struct ast_json *message)
1501 {
1502         struct stasis_app *app;
1503
1504         if (!apps_registry) {
1505                 return -1;
1506         }
1507
1508         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1509         if (!app) {
1510                 /* XXX We can do a better job handling late binding, queueing up
1511                  * the call for a few seconds to wait for the app to register.
1512                  */
1513                 ast_log(LOG_WARNING,
1514                         "Stasis app '%s' not registered\n", app_name);
1515                 return -1;
1516         }
1517         app_send(app, message);
1518         ao2_ref(app, -1);
1519
1520         return 0;
1521 }
1522
1523 static struct stasis_app *find_app_by_name(const char *app_name)
1524 {
1525         struct stasis_app *res = NULL;
1526
1527         if (!apps_registry) {
1528                 return NULL;
1529         }
1530
1531         if (!ast_strlen_zero(app_name)) {
1532                 res = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1533         }
1534
1535         return res;
1536 }
1537
1538 struct stasis_app *stasis_app_get_by_name(const char *name)
1539 {
1540         return find_app_by_name(name);
1541 }
1542
1543 static int append_name(void *obj, void *arg, int flags)
1544 {
1545         struct stasis_app *app = obj;
1546         struct ao2_container *apps = arg;
1547
1548         ast_str_container_add(apps, stasis_app_name(app));
1549         return 0;
1550 }
1551
1552 struct ao2_container *stasis_app_get_all(void)
1553 {
1554         struct ao2_container *apps;
1555
1556         if (!apps_registry) {
1557                 return NULL;
1558         }
1559
1560         apps = ast_str_container_alloc(1);
1561         if (!apps) {
1562                 return NULL;
1563         }
1564
1565         ao2_callback(apps_registry, OBJ_NODATA, append_name, apps);
1566
1567         return apps;
1568 }
1569
1570 static int __stasis_app_register(const char *app_name, stasis_app_cb handler, void *data, int all_events)
1571 {
1572         struct stasis_app *app;
1573
1574         if (!apps_registry) {
1575                 return -1;
1576         }
1577
1578         ao2_lock(apps_registry);
1579         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
1580         if (app) {
1581                 app_update(app, handler, data);
1582         } else {
1583                 app = app_create(app_name, handler, data, all_events ? STASIS_APP_SUBSCRIBE_ALL : STASIS_APP_SUBSCRIBE_MANUAL);
1584                 if (!app) {
1585                         ao2_unlock(apps_registry);
1586                         return -1;
1587                 }
1588
1589                 if (all_events) {
1590                         struct stasis_app_event_source *source;
1591
1592                         AST_RWLIST_RDLOCK(&event_sources);
1593                         AST_LIST_TRAVERSE(&event_sources, source, next) {
1594                                 if (!source->subscribe) {
1595                                         continue;
1596                                 }
1597
1598                                 source->subscribe(app, NULL);
1599                         }
1600                         AST_RWLIST_UNLOCK(&event_sources);
1601                 }
1602                 ao2_link_flags(apps_registry, app, OBJ_NOLOCK);
1603         }
1604
1605         /* We lazily clean up the apps_registry, because it's good enough to
1606          * prevent memory leaks, and we're lazy.
1607          */
1608         cleanup();
1609         ao2_unlock(apps_registry);
1610         ao2_ref(app, -1);
1611         return 0;
1612 }
1613
1614 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
1615 {
1616         return __stasis_app_register(app_name, handler, data, 0);
1617 }
1618
1619 int stasis_app_register_all(const char *app_name, stasis_app_cb handler, void *data)
1620 {
1621         return __stasis_app_register(app_name, handler, data, 1);
1622 }
1623
1624 void stasis_app_unregister(const char *app_name)
1625 {
1626         struct stasis_app *app;
1627
1628         if (!app_name) {
1629                 return;
1630         }
1631
1632         if (!apps_registry) {
1633                 return;
1634         }
1635
1636         app = ao2_find(apps_registry, app_name, OBJ_SEARCH_KEY);
1637         if (!app) {
1638                 ast_log(LOG_ERROR,
1639                         "Stasis app '%s' not registered\n", app_name);
1640                 return;
1641         }
1642
1643         app_deactivate(app);
1644
1645         /* There's a decent chance that app is ready for cleanup. Go ahead
1646          * and clean up, just in case
1647          */
1648         cleanup();
1649
1650         ao2_ref(app, -1);
1651 }
1652
1653 void stasis_app_register_event_source(struct stasis_app_event_source *obj)
1654 {
1655         AST_RWLIST_WRLOCK(&event_sources);
1656         AST_LIST_INSERT_TAIL(&event_sources, obj, next);
1657         /* only need to bump the module ref on non-core sources because the
1658            core ones are [un]registered by this module. */
1659         if (!stasis_app_is_core_event_source(obj)) {
1660                 ast_module_ref(ast_module_info->self);
1661         }
1662         AST_RWLIST_UNLOCK(&event_sources);
1663 }
1664
1665 void stasis_app_unregister_event_source(struct stasis_app_event_source *obj)
1666 {
1667         struct stasis_app_event_source *source;
1668
1669         AST_RWLIST_WRLOCK(&event_sources);
1670         AST_RWLIST_TRAVERSE_SAFE_BEGIN(&event_sources, source, next) {
1671                 if (source == obj) {
1672                         AST_RWLIST_REMOVE_CURRENT(next);
1673                         if (!stasis_app_is_core_event_source(obj)) {
1674                                 ast_module_unref(ast_module_info->self);
1675                         }
1676                         break;
1677                 }
1678         }
1679         AST_RWLIST_TRAVERSE_SAFE_END;
1680         AST_RWLIST_UNLOCK(&event_sources);
1681 }
1682
1683 /*!
1684  * \internal
1685  * \brief Convert event source data to JSON.
1686  *
1687  * Calls each event source that has a "to_json" handler allowing each
1688  * source to add data to the given JSON object.
1689  *
1690  * \param app application associated with the event source
1691  * \param json a json object to "fill"
1692  *
1693  * \retval The given json object.
1694  */
1695 static struct ast_json *app_event_sources_to_json(
1696         const struct stasis_app *app, struct ast_json *json)
1697 {
1698         struct stasis_app_event_source *source;
1699
1700         AST_RWLIST_RDLOCK(&event_sources);
1701         AST_LIST_TRAVERSE(&event_sources, source, next) {
1702                 if (source->to_json) {
1703                         source->to_json(app, json);
1704                 }
1705         }
1706         AST_RWLIST_UNLOCK(&event_sources);
1707
1708         return json;
1709 }
1710
1711 static struct ast_json *stasis_app_object_to_json(struct stasis_app *app)
1712 {
1713         if (!app) {
1714                 return NULL;
1715         }
1716
1717         return app_event_sources_to_json(app, app_to_json(app));
1718 }
1719
1720 struct ast_json *stasis_app_to_json(const char *app_name)
1721 {
1722         struct stasis_app *app = find_app_by_name(app_name);
1723         struct ast_json *json = stasis_app_object_to_json(app);
1724
1725         ao2_cleanup(app);
1726
1727         return json;
1728 }
1729
1730 /*!
1731  * \internal
1732  * \brief Finds an event source that matches a uri scheme.
1733  *
1734  * Uri(s) should begin with a particular scheme that can be matched
1735  * against an event source.
1736  *
1737  * \param uri uri containing a scheme to match
1738  *
1739  * \retval an event source if found, NULL otherwise.
1740  */
1741 static struct stasis_app_event_source *app_event_source_find(const char *uri)
1742 {
1743         struct stasis_app_event_source *source;
1744
1745         AST_RWLIST_RDLOCK(&event_sources);
1746         AST_LIST_TRAVERSE(&event_sources, source, next) {
1747                 if (ast_begins_with(uri, source->scheme)) {
1748                         break;
1749                 }
1750         }
1751         AST_RWLIST_UNLOCK(&event_sources);
1752
1753         return source;
1754 }
1755
1756 /*!
1757  * \internal
1758  * \brief Callback for subscription handling
1759  *
1760  * \param app [un]subscribing application
1761  * \param uri scheme:id of an event source
1762  * \param event_source being [un]subscribed [from]to
1763  *
1764  * \retval stasis_app_subscribe_res return code.
1765  */
1766 typedef enum stasis_app_subscribe_res (*app_subscription_handler)(
1767         struct stasis_app *app, const char *uri,
1768         struct stasis_app_event_source *event_source);
1769
1770 /*!
1771  * \internal
1772  * \brief Subscriptions handler for application [un]subscribing.
1773  *
1774  * \param app_name Name of the application to subscribe.
1775  * \param event_source_uris URIs for the event sources to subscribe to.
1776  * \param event_sources_count Array size of event_source_uris.
1777  * \param json Optional output pointer for JSON representation of the app
1778  *             after adding the subscription.
1779  * \param handler [un]subscribe handler
1780  *
1781  * \retval stasis_app_subscribe_res return code.
1782  */
1783 static enum stasis_app_subscribe_res app_handle_subscriptions(
1784         const char *app_name, const char **event_source_uris,
1785         int event_sources_count, struct ast_json **json,
1786         app_subscription_handler handler)
1787 {
1788         struct stasis_app *app = find_app_by_name(app_name);
1789         int i;
1790
1791         ast_assert(handler != NULL);
1792
1793         if (!app) {
1794                 return STASIS_ASR_APP_NOT_FOUND;
1795         }
1796
1797         for (i = 0; i < event_sources_count; ++i) {
1798                 const char *uri = event_source_uris[i];
1799                 struct stasis_app_event_source *event_source;
1800                 enum stasis_app_subscribe_res res;
1801
1802                 event_source = app_event_source_find(uri);
1803                 if (!event_source) {
1804                         ast_log(LOG_WARNING, "Invalid scheme: %s\n", uri);
1805                         ao2_ref(app, -1);
1806
1807                         return STASIS_ASR_EVENT_SOURCE_BAD_SCHEME;
1808                 }
1809
1810                 res = handler(app, uri, event_source);
1811                 if (res != STASIS_ASR_OK) {
1812                         ao2_ref(app, -1);
1813
1814                         return res;
1815                 }
1816         }
1817
1818         if (json) {
1819                 ast_debug(3, "%s: Successful; setting results\n", app_name);
1820                 *json = stasis_app_object_to_json(app);
1821         }
1822
1823         ao2_ref(app, -1);
1824
1825         return STASIS_ASR_OK;
1826 }
1827
1828 enum stasis_app_subscribe_res stasis_app_subscribe_channel(const char *app_name,
1829         struct ast_channel *chan)
1830 {
1831         struct stasis_app *app = find_app_by_name(app_name);
1832         int res;
1833
1834         if (!app) {
1835                 return STASIS_ASR_APP_NOT_FOUND;
1836         }
1837
1838         ast_debug(3, "%s: Subscribing to %s\n", app_name, ast_channel_uniqueid(chan));
1839
1840         res = app_subscribe_channel(app, chan);
1841         ao2_ref(app, -1);
1842
1843         if (res != 0) {
1844                 ast_log(LOG_ERROR, "Error subscribing app '%s' to channel '%s'\n",
1845                         app_name, ast_channel_uniqueid(chan));
1846                 return STASIS_ASR_INTERNAL_ERROR;
1847         }
1848
1849         return STASIS_ASR_OK;
1850 }
1851
1852
1853 /*!
1854  * \internal
1855  * \brief Subscribe an app to an event source.
1856  *
1857  * \param app subscribing application
1858  * \param uri scheme:id of an event source
1859  * \param event_source being subscribed to
1860  *
1861  * \retval stasis_app_subscribe_res return code.
1862  */
1863 static enum stasis_app_subscribe_res app_subscribe(
1864         struct stasis_app *app, const char *uri,
1865         struct stasis_app_event_source *event_source)
1866 {
1867         const char *app_name = stasis_app_name(app);
1868         RAII_VAR(void *, obj, NULL, ao2_cleanup);
1869
1870         ast_debug(3, "%s: Checking %s\n", app_name, uri);
1871
1872         if (!ast_strlen_zero(uri + strlen(event_source->scheme)) &&
1873             (!event_source->find || (!(obj = event_source->find(app, uri + strlen(event_source->scheme)))))) {
1874                 ast_log(LOG_WARNING, "Event source not found: %s\n", uri);
1875                 return STASIS_ASR_EVENT_SOURCE_NOT_FOUND;
1876         }
1877
1878         ast_debug(3, "%s: Subscribing to %s\n", app_name, uri);
1879
1880         if (!event_source->subscribe || (event_source->subscribe(app, obj))) {
1881                 ast_log(LOG_WARNING, "Error subscribing app '%s' to '%s'\n",
1882                         app_name, uri);
1883                 return STASIS_ASR_INTERNAL_ERROR;
1884         }
1885
1886         return STASIS_ASR_OK;
1887 }
1888
1889 enum stasis_app_subscribe_res stasis_app_subscribe(const char *app_name,
1890         const char **event_source_uris, int event_sources_count,
1891         struct ast_json **json)
1892 {
1893         return app_handle_subscriptions(
1894                 app_name, event_source_uris, event_sources_count,
1895                 json, app_subscribe);
1896 }
1897
1898 /*!
1899  * \internal
1900  * \brief Unsubscribe an app from an event source.
1901  *
1902  * \param app application to unsubscribe
1903  * \param uri scheme:id of an event source
1904  * \param event_source being unsubscribed from
1905  *
1906  * \retval stasis_app_subscribe_res return code.
1907  */
1908 static enum stasis_app_subscribe_res app_unsubscribe(
1909         struct stasis_app *app, const char *uri,
1910         struct stasis_app_event_source *event_source)
1911 {
1912         const char *app_name = stasis_app_name(app);
1913         const char *id = uri + strlen(event_source->scheme);
1914
1915         if (!event_source->is_subscribed ||
1916             (!event_source->is_subscribed(app, id))) {
1917                 return STASIS_ASR_EVENT_SOURCE_NOT_FOUND;
1918         }
1919
1920         ast_debug(3, "%s: Unsubscribing from %s\n", app_name, uri);
1921
1922         if (!event_source->unsubscribe || (event_source->unsubscribe(app, id))) {
1923                 ast_log(LOG_WARNING, "Error unsubscribing app '%s' to '%s'\n",
1924                         app_name, uri);
1925                 return -1;
1926         }
1927         return 0;
1928 }
1929
1930 enum stasis_app_subscribe_res stasis_app_unsubscribe(const char *app_name,
1931         const char **event_source_uris, int event_sources_count,
1932         struct ast_json **json)
1933 {
1934         return app_handle_subscriptions(
1935                 app_name, event_source_uris, event_sources_count,
1936                 json, app_unsubscribe);
1937 }
1938
1939 enum stasis_app_user_event_res stasis_app_user_event(const char *app_name,
1940         const char *event_name,
1941         const char **source_uris, int sources_count,
1942         struct ast_json *json_variables)
1943 {
1944         RAII_VAR(struct stasis_app *, app, find_app_by_name(app_name), ao2_cleanup);
1945         struct ast_json *blob = NULL;
1946         struct ast_multi_object_blob *multi;
1947         struct stasis_message *message;
1948         enum stasis_app_user_event_res res = STASIS_APP_USER_INTERNAL_ERROR;
1949         int have_channel = 0;
1950         int i;
1951
1952         if (!app) {
1953                 ast_log(LOG_WARNING, "App %s not found\n", app_name);
1954                 return STASIS_APP_USER_APP_NOT_FOUND;
1955         }
1956
1957         if (!ast_multi_user_event_type()) {
1958                 return res;
1959         }
1960
1961         if (json_variables) {
1962                 struct ast_json *json_value = ast_json_string_create(event_name);
1963
1964                 if (json_value && !ast_json_object_set(json_variables, "eventname", json_value)) {
1965                         blob = ast_json_ref(json_variables);
1966                 }
1967         } else {
1968                 blob = ast_json_pack("{s: s}", "eventname", event_name);
1969         }
1970
1971         if (!blob) {
1972                 ast_log(LOG_ERROR, "Failed to initialize blob\n");
1973
1974                 return res;
1975         }
1976
1977         multi = ast_multi_object_blob_create(blob);
1978         ast_json_unref(blob);
1979         if (!multi) {
1980                 ast_log(LOG_ERROR, "Failed to initialize multi\n");
1981
1982                 return res;
1983         }
1984
1985         for (i = 0; i < sources_count; ++i) {
1986                 const char *uri = source_uris[i];
1987                 void *snapshot=NULL;
1988                 enum stasis_user_multi_object_snapshot_type type;
1989
1990                 if (ast_begins_with(uri, "channel:")) {
1991                         type = STASIS_UMOS_CHANNEL;
1992                         snapshot = ast_channel_snapshot_get_latest(uri + 8);
1993                         have_channel = 1;
1994                 } else if (ast_begins_with(uri, "bridge:")) {
1995                         type = STASIS_UMOS_BRIDGE;
1996                         snapshot = ast_bridge_snapshot_get_latest(uri + 7);
1997                 } else if (ast_begins_with(uri, "endpoint:")) {
1998                         type = STASIS_UMOS_ENDPOINT;
1999                         snapshot = ast_endpoint_latest_snapshot(uri + 9, NULL);
2000                 } else {
2001                         ast_log(LOG_WARNING, "Invalid scheme: %s\n", uri);
2002                         ao2_ref(multi, -1);
2003
2004                         return STASIS_APP_USER_EVENT_SOURCE_BAD_SCHEME;
2005                 }
2006                 if (!snapshot) {
2007                         ast_log(LOG_ERROR, "Unable to get snapshot for %s\n", uri);
2008                         ao2_ref(multi, -1);
2009
2010                         return STASIS_APP_USER_EVENT_SOURCE_NOT_FOUND;
2011                 }
2012                 ast_multi_object_blob_add(multi, type, snapshot);
2013         }
2014
2015         message = stasis_message_create(ast_multi_user_event_type(), multi);
2016         ao2_ref(multi, -1);
2017
2018         if (!message) {
2019                 ast_log(LOG_ERROR, "Unable to create stasis user event message\n");
2020                 return res;
2021         }
2022
2023         /*
2024          * Publishing to two different topics is normally to be avoided -- except
2025          * in this case both are final destinations with no forwards (only listeners).
2026          * The message has to be delivered to the application topic for ARI, but a
2027          * copy is also delivered directly to the manager for AMI if there is a channel.
2028          */
2029         stasis_publish(ast_app_get_topic(app), message);
2030
2031         if (have_channel) {
2032                 stasis_publish(ast_manager_get_topic(), message);
2033         }
2034         ao2_ref(message, -1);
2035
2036         return STASIS_APP_USER_OK;
2037 }
2038
2039 void stasis_app_ref(void)
2040 {
2041         ast_module_ref(ast_module_info->self);
2042 }
2043
2044 void stasis_app_unref(void)
2045 {
2046         ast_module_unref(ast_module_info->self);
2047 }
2048
2049 static int unload_module(void)
2050 {
2051         stasis_app_unregister_event_sources();
2052
2053         messaging_cleanup();
2054
2055         cleanup();
2056
2057         stasis_app_control_shutdown();
2058
2059         ao2_cleanup(apps_registry);
2060         apps_registry = NULL;
2061
2062         ao2_cleanup(app_controls);
2063         app_controls = NULL;
2064
2065         ao2_cleanup(app_bridges);
2066         app_bridges = NULL;
2067
2068         ao2_cleanup(app_bridges_moh);
2069         app_bridges_moh = NULL;
2070
2071         ao2_cleanup(app_bridges_playback);
2072         app_bridges_playback = NULL;
2073
2074         STASIS_MESSAGE_TYPE_CLEANUP(end_message_type);
2075         STASIS_MESSAGE_TYPE_CLEANUP(start_message_type);
2076
2077         return 0;
2078 }
2079
2080 /* \brief Sanitization callback for channel snapshots */
2081 static int channel_snapshot_sanitizer(const struct ast_channel_snapshot *snapshot)
2082 {
2083         if (!snapshot || !(snapshot->tech_properties & AST_CHAN_TP_INTERNAL)) {
2084                 return 0;
2085         }
2086         return 1;
2087 }
2088
2089 /* \brief Sanitization callback for channels */
2090 static int channel_sanitizer(const struct ast_channel *chan)
2091 {
2092         if (!chan || !(ast_channel_tech(chan)->properties & AST_CHAN_TP_INTERNAL)) {
2093                 return 0;
2094         }
2095         return 1;
2096 }
2097
2098 /* \brief Sanitization callback for channel unique IDs */
2099 static int channel_id_sanitizer(const char *id)
2100 {
2101         struct ast_channel_snapshot *snapshot;
2102         int ret;
2103
2104         snapshot = ast_channel_snapshot_get_latest(id);
2105         ret = channel_snapshot_sanitizer(snapshot);
2106         ao2_cleanup(snapshot);
2107
2108         return ret;
2109 }
2110
2111 /* \brief Sanitization callbacks for communication to Stasis applications */
2112 struct stasis_message_sanitizer app_sanitizer = {
2113         .channel_id = channel_id_sanitizer,
2114         .channel_snapshot = channel_snapshot_sanitizer,
2115         .channel = channel_sanitizer,
2116 };
2117
2118 struct stasis_message_sanitizer *stasis_app_get_sanitizer(void)
2119 {
2120         return &app_sanitizer;
2121 }
2122
2123 static const struct ast_datastore_info stasis_internal_channel_info = {
2124         .type = "stasis-internal-channel",
2125 };
2126
2127 static int set_internal_datastore(struct ast_channel *chan)
2128 {
2129         struct ast_datastore *datastore;
2130
2131         datastore = ast_channel_datastore_find(chan, &stasis_internal_channel_info, NULL);
2132         if (!datastore) {
2133                 datastore = ast_datastore_alloc(&stasis_internal_channel_info, NULL);
2134                 if (!datastore) {
2135                         return -1;
2136                 }
2137                 ast_channel_datastore_add(chan, datastore);
2138         }
2139         return 0;
2140 }
2141
2142 int stasis_app_channel_unreal_set_internal(struct ast_channel *chan)
2143 {
2144         struct ast_channel *outchan = NULL, *outowner = NULL;
2145         int res = 0;
2146         struct ast_unreal_pvt *unreal_pvt = ast_channel_tech_pvt(chan);
2147
2148         ao2_ref(unreal_pvt, +1);
2149         ast_unreal_lock_all(unreal_pvt, &outowner, &outchan);
2150         if (outowner) {
2151                 res |= set_internal_datastore(outowner);
2152                 ast_channel_unlock(outowner);
2153                 ast_channel_unref(outowner);
2154         }
2155         if (outchan) {
2156                 res |= set_internal_datastore(outchan);
2157                 ast_channel_unlock(outchan);
2158                 ast_channel_unref(outchan);
2159         }
2160         ao2_unlock(unreal_pvt);
2161         ao2_ref(unreal_pvt, -1);
2162         return res;
2163 }
2164
2165 int stasis_app_channel_set_internal(struct ast_channel *chan)
2166 {
2167         int res;
2168
2169         ast_channel_lock(chan);
2170         res = set_internal_datastore(chan);
2171         ast_channel_unlock(chan);
2172
2173         return res;
2174 }
2175
2176 int stasis_app_channel_is_internal(struct ast_channel *chan)
2177 {
2178         struct ast_datastore *datastore;
2179         int res = 0;
2180
2181         ast_channel_lock(chan);
2182         datastore = ast_channel_datastore_find(chan, &stasis_internal_channel_info, NULL);
2183         if (datastore) {
2184                 res = 1;
2185         }
2186         ast_channel_unlock(chan);
2187
2188         return res;
2189 }
2190
2191 static int load_module(void)
2192 {
2193         if (STASIS_MESSAGE_TYPE_INIT(start_message_type) != 0) {
2194                 return AST_MODULE_LOAD_DECLINE;
2195         }
2196         if (STASIS_MESSAGE_TYPE_INIT(end_message_type) != 0) {
2197                 return AST_MODULE_LOAD_DECLINE;
2198         }
2199         apps_registry = ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
2200         app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS, control_hash, control_compare);
2201         app_bridges = ao2_container_alloc(BRIDGES_NUM_BUCKETS, bridges_hash, bridges_compare);
2202         app_bridges_moh = ao2_container_alloc_hash(
2203                 AO2_ALLOC_OPT_LOCK_MUTEX, AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT,
2204                 37, bridges_channel_hash_fn, bridges_channel_sort_fn, NULL);
2205         app_bridges_playback = ao2_container_alloc_hash(
2206                 AO2_ALLOC_OPT_LOCK_MUTEX, AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT,
2207                 37, bridges_channel_hash_fn, bridges_channel_sort_fn, NULL);
2208         if (!apps_registry || !app_controls || !app_bridges || !app_bridges_moh || !app_bridges_playback) {
2209                 unload_module();
2210                 return AST_MODULE_LOAD_DECLINE;
2211         }
2212
2213         if (messaging_init()) {
2214                 unload_module();
2215                 return AST_MODULE_LOAD_DECLINE;
2216         }
2217
2218         bridge_stasis_init();
2219
2220         stasis_app_register_event_sources();
2221
2222         return AST_MODULE_LOAD_SUCCESS;
2223 }
2224
2225 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER, "Stasis application support",
2226         .load_pri = AST_MODPRI_APP_DEPEND - 1,
2227         .support_level = AST_MODULE_SUPPORT_CORE,
2228         .load = load_module,
2229         .unload = unload_module,
2230 );