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