Bridge API: Set a cause code on a channel when it is ejected from a bridge.
[asterisk/asterisk.git] / res / res_stasis.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2012 - 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 /*! \file
20  *
21  * \brief Stasis application support.
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  *
25  * <code>res_stasis.so</code> brings together the various components of the
26  * Stasis application infrastructure.
27  *
28  * First, there's the Stasis application handler, stasis_app_exec(). This is
29  * called by <code>app_stasis.so</code> to give control of a channel to the
30  * Stasis application code from the dialplan.
31  *
32  * While a channel is in stasis_app_exec(), it has a \ref stasis_app_control
33  * object, which may be used to control the channel.
34  *
35  * To control the channel, commands may be sent to channel using
36  * stasis_app_send_command() and stasis_app_send_async_command().
37  *
38  * Alongside this, applications may be registered/unregistered using
39  * stasis_app_register()/stasis_app_unregister(). While a channel is in Stasis,
40  * events received on the channel's topic are converted to JSON and forwarded to
41  * the \ref stasis_app_cb. The application may also subscribe to the channel to
42  * continue to receive messages even after the channel has left Stasis, but it
43  * will not be able to control it.
44  *
45  * Given all the stuff that comes together in this module, it's been broken up
46  * into several pieces that are in <code>res/stasis/</code> and compiled into
47  * <code>res_stasis.so</code>.
48  */
49
50 /*** MODULEINFO
51         <support_level>core</support_level>
52  ***/
53
54 #include "asterisk.h"
55
56 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
57
58 #include "asterisk/astobj2.h"
59 #include "asterisk/callerid.h"
60 #include "asterisk/module.h"
61 #include "asterisk/stasis_app_impl.h"
62 #include "asterisk/stasis_channels.h"
63 #include "asterisk/stasis_bridges.h"
64 #include "asterisk/stasis_message_router.h"
65 #include "asterisk/strings.h"
66 #include "stasis/app.h"
67 #include "stasis/control.h"
68
69 /*! Time to wait for a frame in the application */
70 #define MAX_WAIT_MS 200
71
72 /*!
73  * \brief Number of buckets for the Stasis application hash table.  Remember to
74  * keep it a prime number!
75  */
76 #define APPS_NUM_BUCKETS 127
77
78 /*!
79  * \brief Number of buckets for the Stasis application hash table.  Remember to
80  * keep it a prime number!
81  */
82 #define CONTROLS_NUM_BUCKETS 127
83
84 /*!
85  * \brief Stasis application container.
86  */
87 struct ao2_container *apps_registry;
88
89 struct ao2_container *app_controls;
90
91 struct ao2_container *app_bridges;
92
93 /*! \brief Message router for the channel caching topic */
94 struct stasis_message_router *channel_router;
95
96 /*! \brief Message router for the bridge caching topic */
97 struct stasis_message_router *bridge_router;
98
99 /*! AO2 hash function for \ref app */
100 static int app_hash(const void *obj, const int flags)
101 {
102         const struct app *app = obj;
103         const char *name = flags & OBJ_KEY ? obj : app_name(app);
104
105         return ast_str_hash(name);
106 }
107
108 /*! AO2 comparison function for \ref app */
109 static int app_compare(void *lhs, void *rhs, int flags)
110 {
111         const struct app *lhs_app = lhs;
112         const struct app *rhs_app = rhs;
113         const char *lhs_name = app_name(lhs_app);
114         const char *rhs_name = flags & OBJ_KEY ? rhs : app_name(rhs_app);
115
116         if (strcmp(lhs_name, rhs_name) == 0) {
117                 return CMP_MATCH | CMP_STOP;
118         } else {
119                 return 0;
120         }
121 }
122
123 /*! AO2 hash function for \ref stasis_app_control */
124 static int control_hash(const void *obj, const int flags)
125 {
126         const struct stasis_app_control *control = obj;
127         const char *id = flags & OBJ_KEY ?
128                 obj : stasis_app_control_get_channel_id(control);
129
130         return ast_str_hash(id);
131 }
132
133 /*! AO2 comparison function for \ref stasis_app_control */
134 static int control_compare(void *lhs, void *rhs, int flags)
135 {
136         const struct stasis_app_control *lhs_control = lhs;
137         const struct stasis_app_control *rhs_control = rhs;
138         const char *lhs_id = stasis_app_control_get_channel_id(lhs_control);
139         const char *rhs_id = flags & OBJ_KEY ?
140                 rhs : stasis_app_control_get_channel_id(rhs_control);
141
142         if (strcmp(lhs_id, rhs_id) == 0) {
143                 return CMP_MATCH | CMP_STOP;
144         } else {
145                 return 0;
146         }
147 }
148
149 struct stasis_app_control *stasis_app_control_create(struct ast_channel *chan)
150 {
151         return control_create(chan);
152 }
153
154 struct stasis_app_control *stasis_app_control_find_by_channel(
155         const struct ast_channel *chan)
156 {
157         if (chan == NULL) {
158                 return NULL;
159         }
160
161         return stasis_app_control_find_by_channel_id(
162                 ast_channel_uniqueid(chan));
163 }
164
165 struct stasis_app_control *stasis_app_control_find_by_channel_id(
166         const char *channel_id)
167 {
168         return ao2_find(app_controls, channel_id, OBJ_KEY);
169 }
170
171 /*! AO2 hash function for bridges container  */
172 static int bridges_hash(const void *obj, const int flags)
173 {
174         const struct ast_bridge *bridge = obj;
175         const char *id = flags & OBJ_KEY ?
176                 obj : bridge->uniqueid;
177
178         return ast_str_hash(id);
179 }
180
181 /*! AO2 comparison function for bridges container */
182 static int bridges_compare(void *lhs, void *rhs, int flags)
183 {
184         const struct ast_bridge *lhs_bridge = lhs;
185         const struct ast_bridge *rhs_bridge = rhs;
186         const char *lhs_id = lhs_bridge->uniqueid;
187         const char *rhs_id = flags & OBJ_KEY ?
188                 rhs : rhs_bridge->uniqueid;
189
190         if (strcmp(lhs_id, rhs_id) == 0) {
191                 return CMP_MATCH | CMP_STOP;
192         } else {
193                 return 0;
194         }
195 }
196
197 struct ast_bridge *stasis_app_bridge_find_by_id(
198         const char *bridge_id)
199 {
200         return ao2_find(app_bridges, bridge_id, OBJ_KEY);
201 }
202
203 /*! \brief Typedef for blob handler callbacks */
204 typedef struct ast_json *(*channel_blob_handler_cb)(struct ast_channel_blob *);
205
206 /*! \brief Callback to check whether an app is watching a given channel */
207 static int app_watching_channel_cb(void *obj, void *arg, int flags)
208 {
209         struct app *app = obj;
210         char *uniqueid = arg;
211
212         return app_is_watching_channel(app, uniqueid) ? CMP_MATCH : 0;
213 }
214
215 /*! \brief Get a container full of apps that are interested in the specified channel */
216 static struct ao2_container *get_apps_watching_channel(const char *uniqueid)
217 {
218         struct ao2_container *watching_apps;
219         char *uniqueid_dup;
220         RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy);
221         ast_assert(uniqueid != NULL);
222
223         uniqueid_dup = ast_strdupa(uniqueid);
224
225         watching_apps_iter = ao2_callback(apps_registry, OBJ_MULTIPLE, app_watching_channel_cb, uniqueid_dup);
226         watching_apps = watching_apps_iter->c;
227
228         if (!ao2_container_count(watching_apps)) {
229                 return NULL;
230         }
231
232         ao2_ref(watching_apps, +1);
233         return watching_apps_iter->c;
234 }
235
236 /*! \brief Typedef for callbacks that get called on channel snapshot updates */
237 typedef struct ast_json *(*channel_snapshot_monitor)(
238         struct ast_channel_snapshot *old_snapshot,
239         struct ast_channel_snapshot *new_snapshot,
240         const struct timeval *tv);
241
242 static struct ast_json *simple_channel_event(
243         const char *type,
244         struct ast_channel_snapshot *snapshot,
245         const struct timeval *tv)
246 {
247         return ast_json_pack("{s: s, s: o, s: o}",
248                 "type", type,
249                 "timestamp", ast_json_timeval(*tv, NULL),
250                 "channel", ast_channel_snapshot_to_json(snapshot));
251 }
252
253 static struct ast_json *channel_created_event(
254         struct ast_channel_snapshot *snapshot,
255         const struct timeval *tv)
256 {
257         return simple_channel_event("ChannelCreated", snapshot, tv);
258 }
259
260 static struct ast_json *channel_destroyed_event(
261         struct ast_channel_snapshot *snapshot,
262         const struct timeval *tv)
263 {
264         return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
265                 "type", "ChannelDestroyed",
266                 "timestamp", ast_json_timeval(*tv, NULL),
267                 "cause", snapshot->hangupcause,
268                 "cause_txt", ast_cause2str(snapshot->hangupcause),
269                 "channel", ast_channel_snapshot_to_json(snapshot));
270 }
271
272 static struct ast_json *channel_state_change_event(
273         struct ast_channel_snapshot *snapshot,
274         const struct timeval *tv)
275 {
276         return simple_channel_event("ChannelStateChange", snapshot, tv);
277 }
278
279 /*! \brief Handle channel state changes */
280 static struct ast_json *channel_state(
281         struct ast_channel_snapshot *old_snapshot,
282         struct ast_channel_snapshot *new_snapshot,
283         const struct timeval *tv)
284 {
285         struct ast_channel_snapshot *snapshot = new_snapshot ? new_snapshot : old_snapshot;
286
287         if (!old_snapshot) {
288                 return channel_created_event(snapshot, tv);
289         } else if (!new_snapshot) {
290                 return channel_destroyed_event(snapshot, tv);
291         } else if (old_snapshot->state != new_snapshot->state) {
292                 return channel_state_change_event(snapshot, tv);
293         }
294
295         return NULL;
296 }
297
298 static struct ast_json *channel_dialplan(
299         struct ast_channel_snapshot *old_snapshot,
300         struct ast_channel_snapshot *new_snapshot,
301         const struct timeval *tv)
302 {
303         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
304
305         /* No Newexten event on cache clear */
306         if (!new_snapshot) {
307                 return NULL;
308         }
309
310         /* Empty application is not valid for a Newexten event */
311         if (ast_strlen_zero(new_snapshot->appl)) {
312                 return NULL;
313         }
314
315         if (old_snapshot && ast_channel_snapshot_cep_equal(old_snapshot, new_snapshot)) {
316                 return NULL;
317         }
318
319         return ast_json_pack("{s: s, s: o, s: s, s: s, s: o}",
320                 "type", "ChannelDialplan",
321                 "timestamp", ast_json_timeval(*tv, NULL),
322                 "dialplan_app", new_snapshot->appl,
323                 "dialplan_app_data", new_snapshot->data,
324                 "channel", ast_channel_snapshot_to_json(new_snapshot));
325 }
326
327 static struct ast_json *channel_callerid(
328         struct ast_channel_snapshot *old_snapshot,
329         struct ast_channel_snapshot *new_snapshot,
330         const struct timeval *tv)
331 {
332         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
333
334         /* No NewCallerid event on cache clear or first event */
335         if (!old_snapshot || !new_snapshot) {
336                 return NULL;
337         }
338
339         if (ast_channel_snapshot_caller_id_equal(old_snapshot, new_snapshot)) {
340                 return NULL;
341         }
342
343         return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
344                 "type", "ChannelCallerId",
345                 "timestamp", ast_json_timeval(*tv, NULL),
346                 "caller_presentation", new_snapshot->caller_pres,
347                 "caller_presentation_txt", ast_describe_caller_presentation(
348                         new_snapshot->caller_pres),
349                 "channel", ast_channel_snapshot_to_json(new_snapshot));
350 }
351
352 channel_snapshot_monitor channel_monitors[] = {
353         channel_state,
354         channel_dialplan,
355         channel_callerid
356 };
357
358 static int app_send_cb(void *obj, void *arg, int flags)
359 {
360         struct app *app = obj;
361         struct ast_json *msg = arg;
362
363         app_send(app, msg);
364         return 0;
365 }
366
367 static void sub_channel_snapshot_handler(void *data,
368                 struct stasis_subscription *sub,
369                 struct stasis_topic *topic,
370                 struct stasis_message *message)
371 {
372         RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
373         struct stasis_cache_update *update = stasis_message_data(message);
374         struct ast_channel_snapshot *new_snapshot = stasis_message_data(update->new_snapshot);
375         struct ast_channel_snapshot *old_snapshot = stasis_message_data(update->old_snapshot);
376         /* Pull timestamp from the new snapshot, or from the update message
377          * when there isn't one. */
378         const struct timeval *tv = update->new_snapshot ? stasis_message_timestamp(update->new_snapshot) : stasis_message_timestamp(message);
379         int i;
380
381         watching_apps = get_apps_watching_channel(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid);
382         if (!watching_apps) {
383                 return;
384         }
385
386         for (i = 0; i < ARRAY_LEN(channel_monitors); ++i) {
387                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
388
389                 msg = channel_monitors[i](old_snapshot, new_snapshot, tv);
390                 if (msg) {
391                         ao2_callback(watching_apps, OBJ_NODATA, app_send_cb, msg);
392                 }
393         }
394 }
395
396 static void distribute_message(struct ao2_container *apps, struct ast_json *msg)
397 {
398         ao2_callback(apps, OBJ_NODATA, app_send_cb, msg);
399 }
400
401 static void sub_channel_blob_handler(void *data,
402                 struct stasis_subscription *sub,
403                 struct stasis_topic *topic,
404                 struct stasis_message *message)
405 {
406         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
407         RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
408         struct ast_channel_blob *obj = stasis_message_data(message);
409
410         if (!obj->snapshot) {
411                 return;
412         }
413
414         msg = stasis_message_to_json(message);
415         if (!msg) {
416                 return;
417         }
418
419         watching_apps = get_apps_watching_channel(obj->snapshot->uniqueid);
420         if (!watching_apps) {
421                 return;
422         }
423
424         distribute_message(watching_apps, msg);
425 }
426
427 /*!
428  * \brief In addition to running ao2_cleanup(), this function also removes the
429  * object from the app_controls container.
430  */
431 static void control_unlink(struct stasis_app_control *control)
432 {
433         if (!control) {
434                 return;
435         }
436
437         ao2_unlink_flags(app_controls, control,
438                 OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA);
439         ao2_cleanup(control);
440 }
441
442 struct ast_bridge *stasis_app_bridge_create(const char *type)
443 {
444         struct ast_bridge *bridge;
445         int capabilities, flags = AST_BRIDGE_FLAG_MERGE_INHIBIT_FROM | AST_BRIDGE_FLAG_MERGE_INHIBIT_TO
446                 | AST_BRIDGE_FLAG_SWAP_INHIBIT_FROM | AST_BRIDGE_FLAG_SWAP_INHIBIT_TO
447                 | AST_BRIDGE_FLAG_TRANSFER_PROHIBITED;
448
449         if (ast_strlen_zero(type) || !strcmp(type, "mixing")) {
450                 capabilities = AST_BRIDGE_CAPABILITY_1TO1MIX |
451                         AST_BRIDGE_CAPABILITY_MULTIMIX |
452                         AST_BRIDGE_CAPABILITY_NATIVE;
453                 flags |= AST_BRIDGE_FLAG_SMART;
454         } else if (!strcmp(type, "holding")) {
455                 capabilities = AST_BRIDGE_CAPABILITY_HOLDING;
456         } else {
457                 return NULL;
458         }
459
460         bridge = ast_bridge_base_new(capabilities, flags);
461         if (bridge) {
462                 ao2_link(app_bridges, bridge);
463         }
464         return bridge;
465 }
466
467 void stasis_app_bridge_destroy(const char *bridge_id)
468 {
469         struct ast_bridge *bridge = stasis_app_bridge_find_by_id(bridge_id);
470         if (!bridge) {
471                 return;
472         }
473         ao2_unlink(app_bridges, bridge);
474         ast_bridge_destroy(bridge, 0);
475 }
476
477 int app_send_start_msg(struct app *app, struct ast_channel *chan,
478         int argc, char *argv[])
479 {
480         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
481         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
482
483         struct ast_json *json_args;
484         int i;
485
486         ast_assert(chan != NULL);
487
488         /* Set channel info */
489         snapshot = ast_channel_snapshot_create(chan);
490         if (!snapshot) {
491                 return -1;
492         }
493
494         msg = ast_json_pack("{s: s, s: [], s: o}",
495                 "type", "StasisStart",
496                 "args",
497                 "channel", ast_channel_snapshot_to_json(snapshot));
498         if (!msg) {
499                 return -1;
500         }
501
502         /* Append arguments to args array */
503         json_args = ast_json_object_get(msg, "args");
504         ast_assert(json_args != NULL);
505         for (i = 0; i < argc; ++i) {
506                 int r = ast_json_array_append(json_args,
507                                               ast_json_string_create(argv[i]));
508                 if (r != 0) {
509                         ast_log(LOG_ERROR, "Error appending start message\n");
510                         return -1;
511                 }
512         }
513
514         app_send(app, msg);
515         return 0;
516 }
517
518 int app_send_end_msg(struct app *app, struct ast_channel *chan)
519 {
520         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
521         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
522
523         ast_assert(chan != NULL);
524
525         /* Set channel info */
526         snapshot = ast_channel_snapshot_create(chan);
527         if (snapshot == NULL) {
528                 return -1;
529         }
530
531         msg = ast_json_pack("{s: s, s: o}",
532                 "type", "StasisEnd",
533                 "channel", ast_channel_snapshot_to_json(snapshot));
534         if (!msg) {
535                 return -1;
536         }
537
538         app_send(app, msg);
539         return 0;
540 }
541
542 void stasis_app_control_execute_until_exhausted(struct ast_channel *chan, struct stasis_app_control *control)
543 {
544         while (!control_is_done(control)) {
545                 int command_count = control_dispatch_all(control, chan);
546                 if (command_count == 0 || ast_channel_fdno(chan) == -1) {
547                         break;
548                 }
549         }
550 }
551
552 /*! /brief Stasis dialplan application callback */
553 int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
554                     char *argv[])
555 {
556         SCOPED_MODULE_USE(ast_module_info->self);
557
558         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
559         RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
560         int res = 0;
561
562         ast_assert(chan != NULL);
563
564         app = ao2_find(apps_registry, app_name, OBJ_KEY);
565         if (!app) {
566                 ast_log(LOG_ERROR,
567                         "Stasis app '%s' not registered\n", app_name);
568                 return -1;
569         }
570         if (!app_is_active(app)) {
571                 ast_log(LOG_ERROR,
572                         "Stasis app '%s' not active\n", app_name);
573                 return -1;
574         }
575
576         control = control_create(chan);
577         if (!control) {
578                 ast_log(LOG_ERROR, "Allocated failed\n");
579                 return -1;
580         }
581         ao2_link(app_controls, control);
582
583         res = app_send_start_msg(app, chan, argc, argv);
584         if (res != 0) {
585                 ast_log(LOG_ERROR,
586                         "Error sending start message to '%s'\n", app_name);
587                 return res;
588         }
589
590         if (app_add_channel(app, chan)) {
591                 ast_log(LOG_ERROR, "Error adding listener for channel %s to app %s\n", ast_channel_name(chan), app_name);
592                 return -1;
593         }
594
595         while (!control_is_done(control)) {
596                 RAII_VAR(struct ast_frame *, f, NULL, ast_frame_dtor);
597                 int r;
598                 int command_count;
599
600                 /* Check to see if a bridge absorbed our hangup frame */
601                 if (ast_check_hangup_locked(chan)) {
602                         break;
603                 }
604
605                 if (stasis_app_get_bridge(control)) {
606                         /* Bridge is handling channel frames */
607                         control_wait(control);
608                         control_dispatch_all(control, chan);
609                         continue;
610                 }
611
612                 r = ast_waitfor(chan, MAX_WAIT_MS);
613
614                 if (r < 0) {
615                         ast_debug(3, "%s: Poll error\n",
616                                   ast_channel_uniqueid(chan));
617                         break;
618                 }
619
620                 command_count = control_dispatch_all(control, chan);
621
622                 if (command_count > 0 && ast_channel_fdno(chan) == -1) {
623                         /* Command drained the channel; wait for next frame */
624                         continue;
625                 }
626
627                 if (r == 0) {
628                         /* Timeout */
629                         continue;
630                 }
631
632                 f = ast_read(chan);
633                 if (!f) {
634                         /* Continue on in the dialplan */
635                         ast_debug(3, "%s: Hangup (no more frames)\n",
636                                 ast_channel_uniqueid(chan));
637                         break;
638                 }
639
640                 if (f->frametype == AST_FRAME_CONTROL) {
641                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
642                                 /* Continue on in the dialplan */
643                                 ast_debug(3, "%s: Hangup\n",
644                                         ast_channel_uniqueid(chan));
645                                 break;
646                         }
647                 }
648         }
649
650         app_remove_channel(app, chan);
651         res = app_send_end_msg(app, chan);
652         if (res != 0) {
653                 ast_log(LOG_ERROR,
654                         "Error sending end message to %s\n", app_name);
655                 return res;
656         }
657
658         return res;
659 }
660
661 int stasis_app_send(const char *app_name, struct ast_json *message)
662 {
663         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
664
665         app = ao2_find(apps_registry, app_name, OBJ_KEY);
666
667         if (!app) {
668                 /* XXX We can do a better job handling late binding, queueing up
669                  * the call for a few seconds to wait for the app to register.
670                  */
671                 ast_log(LOG_WARNING,
672                         "Stasis app '%s' not registered\n", app_name);
673                 return -1;
674         }
675
676         app_send(app, message);
677         return 0;
678 }
679
680 static int cleanup_cb(void *obj, void *arg, int flags)
681 {
682         struct app *app = obj;
683
684         if (!app_is_finished(app)) {
685                 return 0;
686         }
687
688         ast_verb(1, "Cleaning up application '%s'\n", app_name(app));
689
690         return CMP_MATCH;
691
692 }
693
694 /*!
695  * \brief Clean up any old apps that we don't need any more.
696  */
697 static void cleanup(void)
698 {
699         ao2_callback(apps_registry, OBJ_MULTIPLE | OBJ_NODATA | OBJ_UNLINK,
700                 cleanup_cb, NULL);
701 }
702
703 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
704 {
705         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
706
707         SCOPED_LOCK(apps_lock, apps_registry, ao2_lock, ao2_unlock);
708
709         app = ao2_find(apps_registry, app_name, OBJ_KEY | OBJ_NOLOCK);
710
711         if (app) {
712                 app_update(app, handler, data);
713         } else {
714                 app = app_create(app_name, handler, data);
715                 if (app) {
716                         ao2_link_flags(apps_registry, app, OBJ_NOLOCK);
717                 } else {
718                         return -1;
719                 }
720         }
721
722         /* We lazily clean up the apps_registry, because it's good enough to
723          * prevent memory leaks, and we're lazy.
724          */
725         cleanup();
726         return 0;
727 }
728
729 void stasis_app_unregister(const char *app_name)
730 {
731         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
732
733         if (!app_name) {
734                 return;
735         }
736
737         app = ao2_find(apps_registry, app_name, OBJ_KEY);
738         if (!app) {
739                 ast_log(LOG_ERROR,
740                         "Stasis app '%s' not registered\n", app_name);
741                 return;
742         }
743
744         app_deactivate(app);
745
746         /* There's a decent chance that app is ready for cleanup. Go ahead
747          * and clean up, just in case
748          */
749         cleanup();
750 }
751
752 void stasis_app_ref(void)
753 {
754         ast_module_ref(ast_module_info->self);
755 }
756
757 void stasis_app_unref(void)
758 {
759         ast_module_unref(ast_module_info->self);
760 }
761
762 /*! \brief Callback to check whether an app is watching a given bridge */
763 static int app_watching_bridge_cb(void *obj, void *arg, int flags)
764 {
765         struct app *app = obj;
766         char *uniqueid = arg;
767
768         return app_is_watching_bridge(app, uniqueid) ? CMP_MATCH : 0;
769 }
770
771 /*! \brief Get a container full of apps that are interested in the specified bridge */
772 static struct ao2_container *get_apps_watching_bridge(const char *uniqueid)
773 {
774         struct ao2_container *watching_apps;
775         char *uniqueid_dup;
776         RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy);
777         ast_assert(uniqueid != NULL);
778
779         uniqueid_dup = ast_strdupa(uniqueid);
780
781         watching_apps_iter = ao2_callback(apps_registry, OBJ_MULTIPLE, app_watching_bridge_cb, uniqueid_dup);
782         watching_apps = watching_apps_iter->c;
783
784         if (!ao2_container_count(watching_apps)) {
785                 return NULL;
786         }
787
788         ao2_ref(watching_apps, +1);
789         return watching_apps_iter->c;
790 }
791
792 /*! Callback used to remove an app's interest in a bridge */
793 static int remove_bridge_cb(void *obj, void *arg, int flags)
794 {
795         app_remove_bridge(obj, arg);
796         return 0;
797 }
798
799 static struct ast_json *simple_bridge_event(
800         const char *type,
801         struct ast_bridge_snapshot *snapshot,
802         const struct timeval *tv)
803 {
804         return ast_json_pack("{s: s, s: o, s: o}",
805                 "type", type,
806                 "timestamp", ast_json_timeval(*tv, NULL),
807                 "bridge", ast_bridge_snapshot_to_json(snapshot));
808 }
809
810 static struct ast_json *simple_bridge_channel_event(
811         const char *type,
812         struct ast_bridge_snapshot *bridge_snapshot,
813         struct ast_channel_snapshot *channel_snapshot,
814         const struct timeval *tv)
815 {
816         return ast_json_pack("{s: s, s: o, s: o, s: o}",
817                 "type", type,
818                 "timestamp", ast_json_timeval(*tv, NULL),
819                 "bridge", ast_bridge_snapshot_to_json(bridge_snapshot),
820                 "channel", ast_channel_snapshot_to_json(channel_snapshot));
821 }
822
823 static void sub_bridge_snapshot_handler(void *data,
824                 struct stasis_subscription *sub,
825                 struct stasis_topic *topic,
826                 struct stasis_message *message)
827 {
828         RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
829         struct stasis_cache_update *update = stasis_message_data(message);
830         struct ast_bridge_snapshot *new_snapshot = stasis_message_data(update->new_snapshot);
831         struct ast_bridge_snapshot *old_snapshot = stasis_message_data(update->old_snapshot);
832         const struct timeval *tv = update->new_snapshot ? stasis_message_timestamp(update->new_snapshot) : stasis_message_timestamp(message);
833
834         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
835
836         watching_apps = get_apps_watching_bridge(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid);
837         if (!watching_apps || !ao2_container_count(watching_apps)) {
838                 return;
839         }
840
841         if (!new_snapshot) {
842                 RAII_VAR(char *, bridge_id, ast_strdup(old_snapshot->uniqueid), ast_free);
843
844                 /* The bridge has gone away. Create the message, make sure no apps are
845                  * watching this bridge anymore, and destroy the bridge's control
846                  * structure */
847                 msg = simple_bridge_event("BridgeDestroyed", old_snapshot, tv);
848                 ao2_callback(watching_apps, OBJ_NODATA, remove_bridge_cb, bridge_id);
849                 stasis_app_bridge_destroy(old_snapshot->uniqueid);
850         } else if (!old_snapshot) {
851                 msg = simple_bridge_event("BridgeCreated", old_snapshot, tv);
852         }
853
854         if (!msg) {
855                 return;
856         }
857
858         distribute_message(watching_apps, msg);
859 }
860
861 /*! \brief Callback used to merge two containers of applications */
862 static int list_merge_cb(void *obj, void *arg, int flags)
863 {
864         /* remove any current entries for this app */
865         ao2_find(arg, obj, OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE);
866         /* relink as the only entry */
867         ao2_link(arg, obj);
868         return 0;
869 }
870
871 /*! \brief Merge container src into container dst without modifying src */
872 static void update_apps_list(struct ao2_container *dst, struct ao2_container *src)
873 {
874         ao2_callback(src, OBJ_NODATA, list_merge_cb, dst);
875 }
876
877 /*! \brief Callback for adding to an app's bridges of interest */
878 static int app_add_bridge_cb(void *obj, void *arg, int flags)
879 {
880         app_add_bridge(obj, arg);
881         return 0;
882 }
883
884 /*! \brief Add interest in the given bridge to all apps in the container */
885 static void update_bridge_interest(struct ao2_container *apps, const char *bridge_id)
886 {
887         RAII_VAR(char *, bridge_id_dup, ast_strdup(bridge_id), ast_free);
888         ao2_callback(apps, OBJ_NODATA, app_add_bridge_cb, bridge_id_dup);
889 }
890
891 static void sub_bridge_merge_handler(void *data,
892                 struct stasis_subscription *sub,
893                 struct stasis_topic *topic,
894                 struct stasis_message *message)
895 {
896         RAII_VAR(struct ao2_container *, watching_apps_to, NULL, ao2_cleanup);
897         RAII_VAR(struct ao2_container *, watching_apps_from, NULL, ao2_cleanup);
898         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
899         struct ast_bridge_merge_message *merge = stasis_message_data(message);
900         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
901         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
902         const struct timeval *tv = stasis_message_timestamp(message);
903
904         watching_apps_to = get_apps_watching_bridge(merge->to->uniqueid);
905         if (watching_apps_to) {
906                 update_apps_list(watching_apps_all, watching_apps_to);
907         }
908
909         watching_apps_from = get_apps_watching_bridge(merge->from->uniqueid);
910         if (watching_apps_from) {
911                 update_bridge_interest(watching_apps_from, merge->to->uniqueid);
912                 update_apps_list(watching_apps_all, watching_apps_from);
913         }
914
915         if (!ao2_container_count(watching_apps_all)) {
916                 return;
917         }
918
919         msg = ast_json_pack("{s: s, s: o, s: o, s: o}",
920                 "type", "BridgeMerged",
921                 "timestamp", ast_json_timeval(*tv, NULL),
922                 "bridge", ast_bridge_snapshot_to_json(merge->to),
923                 "bridge_from", ast_bridge_snapshot_to_json(merge->from));
924
925         if (!msg) {
926                 return;
927         }
928
929         distribute_message(watching_apps_all, msg);
930 }
931
932 static void sub_bridge_enter_handler(void *data,
933                 struct stasis_subscription *sub,
934                 struct stasis_topic *topic,
935                 struct stasis_message *message)
936 {
937         RAII_VAR(struct ao2_container *, watching_apps_channel, NULL, ao2_cleanup);
938         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
939         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
940         struct ast_bridge_blob *obj = stasis_message_data(message);
941         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
942
943         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
944         if (watching_apps_bridge) {
945                 update_apps_list(watching_apps_all, watching_apps_bridge);
946         }
947
948         watching_apps_channel = get_apps_watching_channel(obj->channel->uniqueid);
949         if (watching_apps_channel) {
950                 update_bridge_interest(watching_apps_channel, obj->bridge->uniqueid);
951                 update_apps_list(watching_apps_all, watching_apps_channel);
952         }
953
954         if (!ao2_container_count(watching_apps_all)) {
955                 return;
956         }
957
958         msg = simple_bridge_channel_event("ChannelEnteredBridge", obj->bridge,
959                 obj->channel, stasis_message_timestamp(message));
960
961         distribute_message(watching_apps_all, msg);
962 }
963
964 static void sub_bridge_leave_handler(void *data,
965                 struct stasis_subscription *sub,
966                 struct stasis_topic *topic,
967                 struct stasis_message *message)
968 {
969         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
970         struct ast_bridge_blob *obj = stasis_message_data(message);
971         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
972
973         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
974         if (!watching_apps_bridge) {
975                 return;
976         }
977
978         msg = simple_bridge_channel_event("ChannelLeftBridge", obj->bridge,
979                 obj->channel, stasis_message_timestamp(message));
980
981         distribute_message(watching_apps_bridge, msg);
982 }
983
984 static int load_module(void)
985 {
986         int r = 0;
987
988         apps_registry =
989                 ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
990         if (apps_registry == NULL) {
991                 return AST_MODULE_LOAD_FAILURE;
992         }
993
994         app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
995                                              control_hash, control_compare);
996         if (app_controls == NULL) {
997                 return AST_MODULE_LOAD_FAILURE;
998         }
999
1000         app_bridges = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
1001                                              bridges_hash, bridges_compare);
1002         if (app_bridges == NULL) {
1003                 return AST_MODULE_LOAD_FAILURE;
1004         }
1005
1006         channel_router = stasis_message_router_create(ast_channel_topic_all_cached());
1007         if (!channel_router) {
1008                 return AST_MODULE_LOAD_FAILURE;
1009         }
1010
1011         r |= stasis_message_router_add(channel_router, stasis_cache_update_type(), sub_channel_snapshot_handler, NULL);
1012         /* TODO: This could be handled a lot better. Instead of subscribing to
1013          * the one caching topic and filtering out messages by channel id, we
1014          * should have individual caching topics per-channel, with a shared
1015          * back-end cache. That would simplify a lot of what's going on right
1016          * here.
1017          */
1018         r |= stasis_message_router_add(channel_router, ast_channel_user_event_type(), sub_channel_blob_handler, NULL);
1019         r |= stasis_message_router_add(channel_router, ast_channel_varset_type(), sub_channel_blob_handler, NULL);
1020         r |= stasis_message_router_add(channel_router, ast_channel_dtmf_end_type(), sub_channel_blob_handler, NULL);
1021         r |= stasis_message_router_add(channel_router, ast_channel_hangup_request_type(), sub_channel_blob_handler, NULL);
1022         if (r) {
1023                 return AST_MODULE_LOAD_FAILURE;
1024         }
1025
1026         bridge_router = stasis_message_router_create(ast_bridge_topic_all_cached());
1027         if (!bridge_router) {
1028                 return AST_MODULE_LOAD_FAILURE;
1029         }
1030
1031         r |= stasis_message_router_add(bridge_router, stasis_cache_update_type(), sub_bridge_snapshot_handler, NULL);
1032         r |= stasis_message_router_add(bridge_router, ast_bridge_merge_message_type(), sub_bridge_merge_handler, NULL);
1033         r |= stasis_message_router_add(bridge_router, ast_channel_entered_bridge_type(), sub_bridge_enter_handler, NULL);
1034         r |= stasis_message_router_add(bridge_router, ast_channel_left_bridge_type(), sub_bridge_leave_handler, NULL);
1035         if (r) {
1036                 return AST_MODULE_LOAD_FAILURE;
1037         }
1038
1039         return AST_MODULE_LOAD_SUCCESS;
1040 }
1041
1042 static int unload_module(void)
1043 {
1044         int r = 0;
1045
1046         stasis_message_router_unsubscribe_and_join(channel_router);
1047         channel_router = NULL;
1048
1049         stasis_message_router_unsubscribe_and_join(bridge_router);
1050         bridge_router = NULL;
1051
1052         ao2_cleanup(apps_registry);
1053         apps_registry = NULL;
1054
1055         ao2_cleanup(app_controls);
1056         app_controls = NULL;
1057
1058         ao2_cleanup(app_bridges);
1059         app_bridges = NULL;
1060
1061         return r;
1062 }
1063
1064 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application support",
1065         .load = load_module,
1066         .unload = unload_module,
1067         );