a294db915b1a04fcf1693f1054956d0f20fadec6
[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_bridging.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 = 0;
446         if (ast_strlen_zero(type) || !strcmp(type, "mixing")) {
447                 capabilities = AST_BRIDGE_CAPABILITY_1TO1MIX |
448                         AST_BRIDGE_CAPABILITY_MULTIMIX |
449                         AST_BRIDGE_CAPABILITY_NATIVE;
450                 flags = AST_BRIDGE_FLAG_SMART;
451         } else if (!strcmp(type, "holding")) {
452                 capabilities = AST_BRIDGE_CAPABILITY_HOLDING;
453         } else {
454                 return NULL;
455         }
456
457         bridge = ast_bridge_base_new(capabilities, flags);
458         if (bridge) {
459                 ao2_link(app_bridges, bridge);
460         }
461         return bridge;
462 }
463
464 void stasis_app_bridge_destroy(const char *bridge_id)
465 {
466         struct ast_bridge *bridge = stasis_app_bridge_find_by_id(bridge_id);
467         if (!bridge) {
468                 return;
469         }
470         ao2_unlink(app_bridges, bridge);
471         ast_bridge_destroy(bridge);
472 }
473
474 int app_send_start_msg(struct app *app, struct ast_channel *chan,
475         int argc, char *argv[])
476 {
477         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
478         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
479
480         struct ast_json *json_args;
481         int i;
482
483         ast_assert(chan != NULL);
484
485         /* Set channel info */
486         snapshot = ast_channel_snapshot_create(chan);
487         if (!snapshot) {
488                 return -1;
489         }
490
491         msg = ast_json_pack("{s: s, s: [], s: o}",
492                 "type", "StasisStart",
493                 "args",
494                 "channel", ast_channel_snapshot_to_json(snapshot));
495         if (!msg) {
496                 return -1;
497         }
498
499         /* Append arguments to args array */
500         json_args = ast_json_object_get(msg, "args");
501         ast_assert(json_args != NULL);
502         for (i = 0; i < argc; ++i) {
503                 int r = ast_json_array_append(json_args,
504                                               ast_json_string_create(argv[i]));
505                 if (r != 0) {
506                         ast_log(LOG_ERROR, "Error appending start message\n");
507                         return -1;
508                 }
509         }
510
511         app_send(app, msg);
512         return 0;
513 }
514
515 int app_send_end_msg(struct app *app, struct ast_channel *chan)
516 {
517         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
518         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
519
520         ast_assert(chan != NULL);
521
522         /* Set channel info */
523         snapshot = ast_channel_snapshot_create(chan);
524         if (snapshot == NULL) {
525                 return -1;
526         }
527
528         msg = ast_json_pack("{s: s, s: o}",
529                 "type", "StasisEnd",
530                 "channel", ast_channel_snapshot_to_json(snapshot));
531         if (!msg) {
532                 return -1;
533         }
534
535         app_send(app, msg);
536         return 0;
537 }
538
539 void stasis_app_control_execute_until_exhausted(struct ast_channel *chan, struct stasis_app_control *control)
540 {
541         while (!control_is_done(control)) {
542                 int command_count = control_dispatch_all(control, chan);
543                 if (command_count == 0 || ast_channel_fdno(chan) == -1) {
544                         break;
545                 }
546         }
547 }
548
549 /*! /brief Stasis dialplan application callback */
550 int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
551                     char *argv[])
552 {
553         SCOPED_MODULE_USE(ast_module_info->self);
554
555         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
556         RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
557         int res = 0;
558
559         ast_assert(chan != NULL);
560
561         app = ao2_find(apps_registry, app_name, OBJ_KEY);
562         if (!app) {
563                 ast_log(LOG_ERROR,
564                         "Stasis app '%s' not registered\n", app_name);
565                 return -1;
566         }
567
568         control = control_create(chan);
569         if (!control) {
570                 ast_log(LOG_ERROR, "Allocated failed\n");
571                 return -1;
572         }
573         ao2_link(app_controls, control);
574
575         res = app_send_start_msg(app, chan, argc, argv);
576         if (res != 0) {
577                 ast_log(LOG_ERROR,
578                         "Error sending start message to %s\n", app_name);
579                 return res;
580         }
581
582         if (app_add_channel(app, chan)) {
583                 ast_log(LOG_ERROR, "Error adding listener for channel %s to app %s\n", ast_channel_name(chan), app_name);
584                 return -1;
585         }
586
587         while (!control_is_done(control)) {
588                 RAII_VAR(struct ast_frame *, f, NULL, ast_frame_dtor);
589                 int r;
590                 int command_count;
591
592                 r = ast_waitfor(chan, MAX_WAIT_MS);
593
594                 if (r < 0) {
595                         ast_debug(3, "%s: Poll error\n",
596                                   ast_channel_uniqueid(chan));
597                         break;
598                 }
599
600                 command_count = control_dispatch_all(control, chan);
601
602                 if (command_count > 0 && ast_channel_fdno(chan) == -1) {
603                         /* Command drained the channel; wait for next frame */
604                         continue;
605                 }
606
607                 if (r == 0) {
608                         /* Timeout */
609                         continue;
610                 }
611
612                 f = ast_read(chan);
613                 if (!f) {
614                         ast_debug(3,
615                                 "%s: No more frames. Must be done, I guess.\n",
616                                 ast_channel_uniqueid(chan));
617                         break;
618                 }
619
620                 switch (f->frametype) {
621                 case AST_FRAME_CONTROL:
622                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
623                                 /* Continue on in the dialplan */
624                                 ast_debug(3, "%s: Hangup\n",
625                                         ast_channel_uniqueid(chan));
626                                 control_continue(control);
627                         }
628                         break;
629                 default:
630                         /* Not handled; discard */
631                         break;
632                 }
633         }
634
635         app_remove_channel(app, chan);
636         res = app_send_end_msg(app, chan);
637         if (res != 0) {
638                 ast_log(LOG_ERROR,
639                         "Error sending end message to %s\n", app_name);
640                 return res;
641         }
642
643         return res;
644 }
645
646 int stasis_app_send(const char *app_name, struct ast_json *message)
647 {
648         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
649
650         app = ao2_find(apps_registry, app_name, OBJ_KEY);
651
652         if (!app) {
653                 /* XXX We can do a better job handling late binding, queueing up
654                  * the call for a few seconds to wait for the app to register.
655                  */
656                 ast_log(LOG_WARNING,
657                         "Stasis app '%s' not registered\n", app_name);
658                 return -1;
659         }
660
661         app_send(app, message);
662         return 0;
663 }
664
665 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
666 {
667         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
668
669         SCOPED_LOCK(apps_lock, apps_registry, ao2_lock, ao2_unlock);
670
671         app = ao2_find(apps_registry, app_name, OBJ_KEY | OBJ_NOLOCK);
672
673         if (app) {
674                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
675
676                 msg = ast_json_pack("{s: s, s: s}",
677                         "type", "ApplicationReplaced",
678                         "application", app_name);
679                 if (msg) {
680                         app_send(app, msg);
681                 }
682
683                 app_update(app, handler, data);
684         } else {
685                 app = app_create(app_name, handler, data);
686                 if (app) {
687                         ao2_link_flags(apps_registry, app, OBJ_NOLOCK);
688                 } else {
689                         return -1;
690                 }
691         }
692
693         return 0;
694 }
695
696 void stasis_app_unregister(const char *app_name)
697 {
698         if (app_name) {
699                 ao2_cleanup(ao2_find(
700                                 apps_registry, app_name, OBJ_KEY | OBJ_UNLINK));
701         }
702 }
703
704 void stasis_app_ref(void)
705 {
706         ast_module_ref(ast_module_info->self);
707 }
708
709 void stasis_app_unref(void)
710 {
711         ast_module_unref(ast_module_info->self);
712 }
713
714 /*! \brief Callback to check whether an app is watching a given bridge */
715 static int app_watching_bridge_cb(void *obj, void *arg, int flags)
716 {
717         struct app *app = obj;
718         char *uniqueid = arg;
719
720         return app_is_watching_bridge(app, uniqueid) ? CMP_MATCH : 0;
721 }
722
723 /*! \brief Get a container full of apps that are interested in the specified bridge */
724 static struct ao2_container *get_apps_watching_bridge(const char *uniqueid)
725 {
726         struct ao2_container *watching_apps;
727         char *uniqueid_dup;
728         RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy);
729         ast_assert(uniqueid != NULL);
730
731         uniqueid_dup = ast_strdupa(uniqueid);
732
733         watching_apps_iter = ao2_callback(apps_registry, OBJ_MULTIPLE, app_watching_bridge_cb, uniqueid_dup);
734         watching_apps = watching_apps_iter->c;
735
736         if (!ao2_container_count(watching_apps)) {
737                 return NULL;
738         }
739
740         ao2_ref(watching_apps, +1);
741         return watching_apps_iter->c;
742 }
743
744 /*! Callback used to remove an app's interest in a bridge */
745 static int remove_bridge_cb(void *obj, void *arg, int flags)
746 {
747         app_remove_bridge(obj, arg);
748         return 0;
749 }
750
751 static struct ast_json *simple_bridge_event(
752         const char *type,
753         struct ast_bridge_snapshot *snapshot,
754         const struct timeval *tv)
755 {
756         return ast_json_pack("{s: s, s: o, s: o}",
757                 "type", type,
758                 "timestamp", ast_json_timeval(*tv, NULL),
759                 "bridge", ast_bridge_snapshot_to_json(snapshot));
760 }
761
762 static struct ast_json *simple_bridge_channel_event(
763         const char *type,
764         struct ast_bridge_snapshot *bridge_snapshot,
765         struct ast_channel_snapshot *channel_snapshot,
766         const struct timeval *tv)
767 {
768         return ast_json_pack("{s: s, s: o, s: o, s: o}",
769                 "type", type,
770                 "timestamp", ast_json_timeval(*tv, NULL),
771                 "bridge", ast_bridge_snapshot_to_json(bridge_snapshot),
772                 "channel", ast_channel_snapshot_to_json(channel_snapshot));
773 }
774
775 static void sub_bridge_snapshot_handler(void *data,
776                 struct stasis_subscription *sub,
777                 struct stasis_topic *topic,
778                 struct stasis_message *message)
779 {
780         RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
781         struct stasis_cache_update *update = stasis_message_data(message);
782         struct ast_bridge_snapshot *new_snapshot = stasis_message_data(update->new_snapshot);
783         struct ast_bridge_snapshot *old_snapshot = stasis_message_data(update->old_snapshot);
784         const struct timeval *tv = update->new_snapshot ? stasis_message_timestamp(update->new_snapshot) : stasis_message_timestamp(message);
785
786         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
787
788         watching_apps = get_apps_watching_bridge(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid);
789         if (!watching_apps || !ao2_container_count(watching_apps)) {
790                 return;
791         }
792
793         if (!new_snapshot) {
794                 RAII_VAR(char *, bridge_id, ast_strdup(old_snapshot->uniqueid), ast_free);
795
796                 /* The bridge has gone away. Create the message, make sure no apps are
797                  * watching this bridge anymore, and destroy the bridge's control
798                  * structure */
799                 msg = simple_bridge_event("BridgeDestroyed", old_snapshot, tv);
800                 ao2_callback(watching_apps, OBJ_NODATA, remove_bridge_cb, bridge_id);
801                 stasis_app_bridge_destroy(old_snapshot->uniqueid);
802         } else if (!old_snapshot) {
803                 msg = simple_bridge_event("BridgeCreated", old_snapshot, tv);
804         }
805
806         if (!msg) {
807                 return;
808         }
809
810         distribute_message(watching_apps, msg);
811 }
812
813 /*! \brief Callback used to merge two containers of applications */
814 static int list_merge_cb(void *obj, void *arg, int flags)
815 {
816         /* remove any current entries for this app */
817         ao2_find(arg, obj, OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE);
818         /* relink as the only entry */
819         ao2_link(arg, obj);
820         return 0;
821 }
822
823 /*! \brief Merge container src into container dst without modifying src */
824 static void update_apps_list(struct ao2_container *dst, struct ao2_container *src)
825 {
826         ao2_callback(src, OBJ_NODATA, list_merge_cb, dst);
827 }
828
829 /*! \brief Callback for adding to an app's bridges of interest */
830 static int app_add_bridge_cb(void *obj, void *arg, int flags)
831 {
832         app_add_bridge(obj, arg);
833         return 0;
834 }
835
836 /*! \brief Add interest in the given bridge to all apps in the container */
837 static void update_bridge_interest(struct ao2_container *apps, const char *bridge_id)
838 {
839         RAII_VAR(char *, bridge_id_dup, ast_strdup(bridge_id), ast_free);
840         ao2_callback(apps, OBJ_NODATA, app_add_bridge_cb, bridge_id_dup);
841 }
842
843 static void sub_bridge_merge_handler(void *data,
844                 struct stasis_subscription *sub,
845                 struct stasis_topic *topic,
846                 struct stasis_message *message)
847 {
848         RAII_VAR(struct ao2_container *, watching_apps_to, NULL, ao2_cleanup);
849         RAII_VAR(struct ao2_container *, watching_apps_from, NULL, ao2_cleanup);
850         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
851         struct ast_bridge_merge_message *merge = stasis_message_data(message);
852         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
853         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
854         const struct timeval *tv = stasis_message_timestamp(message);
855
856         watching_apps_to = get_apps_watching_bridge(merge->to->uniqueid);
857         if (watching_apps_to) {
858                 update_apps_list(watching_apps_all, watching_apps_to);
859         }
860
861         watching_apps_from = get_apps_watching_bridge(merge->from->uniqueid);
862         if (watching_apps_from) {
863                 update_bridge_interest(watching_apps_from, merge->to->uniqueid);
864                 update_apps_list(watching_apps_all, watching_apps_from);
865         }
866
867         if (!ao2_container_count(watching_apps_all)) {
868                 return;
869         }
870
871         msg = ast_json_pack("{s: s, s: o, s: o, s: o}",
872                 "type", "BridgeMerged",
873                 "timestamp", ast_json_timeval(*tv, NULL),
874                 "bridge", ast_bridge_snapshot_to_json(merge->to),
875                 "bridge_from", ast_bridge_snapshot_to_json(merge->from));
876
877         if (!msg) {
878                 return;
879         }
880
881         distribute_message(watching_apps_all, msg);
882 }
883
884 static void sub_bridge_enter_handler(void *data,
885                 struct stasis_subscription *sub,
886                 struct stasis_topic *topic,
887                 struct stasis_message *message)
888 {
889         RAII_VAR(struct ao2_container *, watching_apps_channel, NULL, ao2_cleanup);
890         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
891         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
892         struct ast_bridge_blob *obj = stasis_message_data(message);
893         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
894
895         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
896         if (watching_apps_bridge) {
897                 update_apps_list(watching_apps_all, watching_apps_bridge);
898         }
899
900         watching_apps_channel = get_apps_watching_channel(obj->channel->uniqueid);
901         if (watching_apps_channel) {
902                 update_bridge_interest(watching_apps_channel, obj->bridge->uniqueid);
903                 update_apps_list(watching_apps_all, watching_apps_channel);
904         }
905
906         if (!ao2_container_count(watching_apps_all)) {
907                 return;
908         }
909
910         msg = simple_bridge_channel_event("ChannelEnteredBridge", obj->bridge,
911                 obj->channel, stasis_message_timestamp(message));
912
913         distribute_message(watching_apps_all, msg);
914 }
915
916 static void sub_bridge_leave_handler(void *data,
917                 struct stasis_subscription *sub,
918                 struct stasis_topic *topic,
919                 struct stasis_message *message)
920 {
921         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
922         struct ast_bridge_blob *obj = stasis_message_data(message);
923         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
924
925         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
926         if (!watching_apps_bridge) {
927                 return;
928         }
929
930         msg = simple_bridge_channel_event("ChannelLeftBridge", obj->bridge,
931                 obj->channel, stasis_message_timestamp(message));
932
933         distribute_message(watching_apps_bridge, msg);
934 }
935
936 static int load_module(void)
937 {
938         int r = 0;
939
940         apps_registry =
941                 ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
942         if (apps_registry == NULL) {
943                 return AST_MODULE_LOAD_FAILURE;
944         }
945
946         app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
947                                              control_hash, control_compare);
948         if (app_controls == NULL) {
949                 return AST_MODULE_LOAD_FAILURE;
950         }
951
952         app_bridges = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
953                                              bridges_hash, bridges_compare);
954         if (app_bridges == NULL) {
955                 return AST_MODULE_LOAD_FAILURE;
956         }
957
958         channel_router = stasis_message_router_create(stasis_caching_get_topic(ast_channel_topic_all_cached()));
959         if (!channel_router) {
960                 return AST_MODULE_LOAD_FAILURE;
961         }
962
963         r |= stasis_message_router_add(channel_router, stasis_cache_update_type(), sub_channel_snapshot_handler, NULL);
964         /* TODO: This could be handled a lot better. Instead of subscribing to
965          * the one caching topic and filtering out messages by channel id, we
966          * should have individual caching topics per-channel, with a shared
967          * back-end cache. That would simplify a lot of what's going on right
968          * here.
969          */
970         r |= stasis_message_router_add(channel_router, ast_channel_user_event_type(), sub_channel_blob_handler, NULL);
971         r |= stasis_message_router_add(channel_router, ast_channel_varset_type(), sub_channel_blob_handler, NULL);
972         r |= stasis_message_router_add(channel_router, ast_channel_dtmf_end_type(), sub_channel_blob_handler, NULL);
973         r |= stasis_message_router_add(channel_router, ast_channel_hangup_request_type(), sub_channel_blob_handler, NULL);
974         if (r) {
975                 return AST_MODULE_LOAD_FAILURE;
976         }
977
978         bridge_router = stasis_message_router_create(stasis_caching_get_topic(ast_bridge_topic_all_cached()));
979         if (!bridge_router) {
980                 return AST_MODULE_LOAD_FAILURE;
981         }
982
983         r |= stasis_message_router_add(bridge_router, stasis_cache_update_type(), sub_bridge_snapshot_handler, NULL);
984         r |= stasis_message_router_add(bridge_router, ast_bridge_merge_message_type(), sub_bridge_merge_handler, NULL);
985         r |= stasis_message_router_add(bridge_router, ast_channel_entered_bridge_type(), sub_bridge_enter_handler, NULL);
986         r |= stasis_message_router_add(bridge_router, ast_channel_left_bridge_type(), sub_bridge_leave_handler, NULL);
987         if (r) {
988                 return AST_MODULE_LOAD_FAILURE;
989         }
990
991         return AST_MODULE_LOAD_SUCCESS;
992 }
993
994 static int unload_module(void)
995 {
996         int r = 0;
997
998         stasis_message_router_unsubscribe_and_join(channel_router);
999         channel_router = NULL;
1000
1001         stasis_message_router_unsubscribe_and_join(bridge_router);
1002         bridge_router = NULL;
1003
1004         ao2_cleanup(apps_registry);
1005         apps_registry = NULL;
1006
1007         ao2_cleanup(app_controls);
1008         app_controls = NULL;
1009
1010         ao2_cleanup(app_bridges);
1011         app_bridges = NULL;
1012
1013         return r;
1014 }
1015
1016 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application support",
1017         .load = load_module,
1018         .unload = unload_module,
1019         );