ARI: allow other operations to happen while bridged
[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 = 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         if (!app_is_active(app)) {
568                 ast_log(LOG_ERROR,
569                         "Stasis app '%s' not active\n", app_name);
570                 return -1;
571         }
572
573         control = control_create(chan);
574         if (!control) {
575                 ast_log(LOG_ERROR, "Allocated failed\n");
576                 return -1;
577         }
578         ao2_link(app_controls, control);
579
580         res = app_send_start_msg(app, chan, argc, argv);
581         if (res != 0) {
582                 ast_log(LOG_ERROR,
583                         "Error sending start message to '%s'\n", app_name);
584                 return res;
585         }
586
587         if (app_add_channel(app, chan)) {
588                 ast_log(LOG_ERROR, "Error adding listener for channel %s to app %s\n", ast_channel_name(chan), app_name);
589                 return -1;
590         }
591
592         while (!control_is_done(control)) {
593                 RAII_VAR(struct ast_frame *, f, NULL, ast_frame_dtor);
594                 int r;
595                 int command_count;
596
597                 if (stasis_app_get_bridge(control)) {
598                         /* Bridge is handling channel frames */
599                         control_wait(control);
600                         control_dispatch_all(control, chan);
601                         continue;
602                 }
603
604                 r = ast_waitfor(chan, MAX_WAIT_MS);
605
606                 if (r < 0) {
607                         ast_debug(3, "%s: Poll error\n",
608                                   ast_channel_uniqueid(chan));
609                         break;
610                 }
611
612                 command_count = control_dispatch_all(control, chan);
613
614                 if (command_count > 0 && ast_channel_fdno(chan) == -1) {
615                         /* Command drained the channel; wait for next frame */
616                         continue;
617                 }
618
619                 if (r == 0) {
620                         /* Timeout */
621                         continue;
622                 }
623
624                 f = ast_read(chan);
625                 if (!f) {
626                         ast_debug(3,
627                                 "%s: No more frames. Must be done, I guess.\n",
628                                 ast_channel_uniqueid(chan));
629                         break;
630                 }
631
632                 switch (f->frametype) {
633                 case AST_FRAME_CONTROL:
634                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
635                                 /* Continue on in the dialplan */
636                                 ast_debug(3, "%s: Hangup\n",
637                                         ast_channel_uniqueid(chan));
638                                 control_continue(control);
639                         }
640                         break;
641                 default:
642                         /* Not handled; discard */
643                         break;
644                 }
645         }
646
647         app_remove_channel(app, chan);
648         res = app_send_end_msg(app, chan);
649         if (res != 0) {
650                 ast_log(LOG_ERROR,
651                         "Error sending end message to %s\n", app_name);
652                 return res;
653         }
654
655         return res;
656 }
657
658 int stasis_app_send(const char *app_name, struct ast_json *message)
659 {
660         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
661
662         app = ao2_find(apps_registry, app_name, OBJ_KEY);
663
664         if (!app) {
665                 /* XXX We can do a better job handling late binding, queueing up
666                  * the call for a few seconds to wait for the app to register.
667                  */
668                 ast_log(LOG_WARNING,
669                         "Stasis app '%s' not registered\n", app_name);
670                 return -1;
671         }
672
673         app_send(app, message);
674         return 0;
675 }
676
677 static int cleanup_cb(void *obj, void *arg, int flags)
678 {
679         struct app *app = obj;
680
681         if (!app_is_finished(app)) {
682                 return 0;
683         }
684
685         ast_verb(1, "Cleaning up application '%s'\n", app_name(app));
686
687         return CMP_MATCH;
688
689 }
690
691 /*!
692  * \brief Clean up any old apps that we don't need any more.
693  */
694 static void cleanup(void)
695 {
696         ao2_callback(apps_registry, OBJ_MULTIPLE | OBJ_NODATA | OBJ_UNLINK,
697                 cleanup_cb, NULL);
698 }
699
700 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
701 {
702         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
703
704         SCOPED_LOCK(apps_lock, apps_registry, ao2_lock, ao2_unlock);
705
706         app = ao2_find(apps_registry, app_name, OBJ_KEY | OBJ_NOLOCK);
707
708         if (app) {
709                 app_update(app, handler, data);
710         } else {
711                 app = app_create(app_name, handler, data);
712                 if (app) {
713                         ao2_link_flags(apps_registry, app, OBJ_NOLOCK);
714                 } else {
715                         return -1;
716                 }
717         }
718
719         /* We lazily clean up the apps_registry, because it's good enough to
720          * prevent memory leaks, and we're lazy.
721          */
722         cleanup();
723         return 0;
724 }
725
726 void stasis_app_unregister(const char *app_name)
727 {
728         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
729
730         if (!app_name) {
731                 return;
732         }
733
734         app = ao2_find(apps_registry, app_name, OBJ_KEY);
735         if (!app) {
736                 ast_log(LOG_ERROR,
737                         "Stasis app '%s' not registered\n", app_name);
738                 return;
739         }
740
741         app_deactivate(app);
742
743         /* There's a decent chance that app is ready for cleanup. Go ahead
744          * and clean up, just in case
745          */
746         cleanup();
747 }
748
749 void stasis_app_ref(void)
750 {
751         ast_module_ref(ast_module_info->self);
752 }
753
754 void stasis_app_unref(void)
755 {
756         ast_module_unref(ast_module_info->self);
757 }
758
759 /*! \brief Callback to check whether an app is watching a given bridge */
760 static int app_watching_bridge_cb(void *obj, void *arg, int flags)
761 {
762         struct app *app = obj;
763         char *uniqueid = arg;
764
765         return app_is_watching_bridge(app, uniqueid) ? CMP_MATCH : 0;
766 }
767
768 /*! \brief Get a container full of apps that are interested in the specified bridge */
769 static struct ao2_container *get_apps_watching_bridge(const char *uniqueid)
770 {
771         struct ao2_container *watching_apps;
772         char *uniqueid_dup;
773         RAII_VAR(struct ao2_iterator *,watching_apps_iter, NULL, ao2_iterator_destroy);
774         ast_assert(uniqueid != NULL);
775
776         uniqueid_dup = ast_strdupa(uniqueid);
777
778         watching_apps_iter = ao2_callback(apps_registry, OBJ_MULTIPLE, app_watching_bridge_cb, uniqueid_dup);
779         watching_apps = watching_apps_iter->c;
780
781         if (!ao2_container_count(watching_apps)) {
782                 return NULL;
783         }
784
785         ao2_ref(watching_apps, +1);
786         return watching_apps_iter->c;
787 }
788
789 /*! Callback used to remove an app's interest in a bridge */
790 static int remove_bridge_cb(void *obj, void *arg, int flags)
791 {
792         app_remove_bridge(obj, arg);
793         return 0;
794 }
795
796 static struct ast_json *simple_bridge_event(
797         const char *type,
798         struct ast_bridge_snapshot *snapshot,
799         const struct timeval *tv)
800 {
801         return ast_json_pack("{s: s, s: o, s: o}",
802                 "type", type,
803                 "timestamp", ast_json_timeval(*tv, NULL),
804                 "bridge", ast_bridge_snapshot_to_json(snapshot));
805 }
806
807 static struct ast_json *simple_bridge_channel_event(
808         const char *type,
809         struct ast_bridge_snapshot *bridge_snapshot,
810         struct ast_channel_snapshot *channel_snapshot,
811         const struct timeval *tv)
812 {
813         return ast_json_pack("{s: s, s: o, s: o, s: o}",
814                 "type", type,
815                 "timestamp", ast_json_timeval(*tv, NULL),
816                 "bridge", ast_bridge_snapshot_to_json(bridge_snapshot),
817                 "channel", ast_channel_snapshot_to_json(channel_snapshot));
818 }
819
820 static void sub_bridge_snapshot_handler(void *data,
821                 struct stasis_subscription *sub,
822                 struct stasis_topic *topic,
823                 struct stasis_message *message)
824 {
825         RAII_VAR(struct ao2_container *, watching_apps, NULL, ao2_cleanup);
826         struct stasis_cache_update *update = stasis_message_data(message);
827         struct ast_bridge_snapshot *new_snapshot = stasis_message_data(update->new_snapshot);
828         struct ast_bridge_snapshot *old_snapshot = stasis_message_data(update->old_snapshot);
829         const struct timeval *tv = update->new_snapshot ? stasis_message_timestamp(update->new_snapshot) : stasis_message_timestamp(message);
830
831         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
832
833         watching_apps = get_apps_watching_bridge(new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid);
834         if (!watching_apps || !ao2_container_count(watching_apps)) {
835                 return;
836         }
837
838         if (!new_snapshot) {
839                 RAII_VAR(char *, bridge_id, ast_strdup(old_snapshot->uniqueid), ast_free);
840
841                 /* The bridge has gone away. Create the message, make sure no apps are
842                  * watching this bridge anymore, and destroy the bridge's control
843                  * structure */
844                 msg = simple_bridge_event("BridgeDestroyed", old_snapshot, tv);
845                 ao2_callback(watching_apps, OBJ_NODATA, remove_bridge_cb, bridge_id);
846                 stasis_app_bridge_destroy(old_snapshot->uniqueid);
847         } else if (!old_snapshot) {
848                 msg = simple_bridge_event("BridgeCreated", old_snapshot, tv);
849         }
850
851         if (!msg) {
852                 return;
853         }
854
855         distribute_message(watching_apps, msg);
856 }
857
858 /*! \brief Callback used to merge two containers of applications */
859 static int list_merge_cb(void *obj, void *arg, int flags)
860 {
861         /* remove any current entries for this app */
862         ao2_find(arg, obj, OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE);
863         /* relink as the only entry */
864         ao2_link(arg, obj);
865         return 0;
866 }
867
868 /*! \brief Merge container src into container dst without modifying src */
869 static void update_apps_list(struct ao2_container *dst, struct ao2_container *src)
870 {
871         ao2_callback(src, OBJ_NODATA, list_merge_cb, dst);
872 }
873
874 /*! \brief Callback for adding to an app's bridges of interest */
875 static int app_add_bridge_cb(void *obj, void *arg, int flags)
876 {
877         app_add_bridge(obj, arg);
878         return 0;
879 }
880
881 /*! \brief Add interest in the given bridge to all apps in the container */
882 static void update_bridge_interest(struct ao2_container *apps, const char *bridge_id)
883 {
884         RAII_VAR(char *, bridge_id_dup, ast_strdup(bridge_id), ast_free);
885         ao2_callback(apps, OBJ_NODATA, app_add_bridge_cb, bridge_id_dup);
886 }
887
888 static void sub_bridge_merge_handler(void *data,
889                 struct stasis_subscription *sub,
890                 struct stasis_topic *topic,
891                 struct stasis_message *message)
892 {
893         RAII_VAR(struct ao2_container *, watching_apps_to, NULL, ao2_cleanup);
894         RAII_VAR(struct ao2_container *, watching_apps_from, NULL, ao2_cleanup);
895         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
896         struct ast_bridge_merge_message *merge = stasis_message_data(message);
897         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
898         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
899         const struct timeval *tv = stasis_message_timestamp(message);
900
901         watching_apps_to = get_apps_watching_bridge(merge->to->uniqueid);
902         if (watching_apps_to) {
903                 update_apps_list(watching_apps_all, watching_apps_to);
904         }
905
906         watching_apps_from = get_apps_watching_bridge(merge->from->uniqueid);
907         if (watching_apps_from) {
908                 update_bridge_interest(watching_apps_from, merge->to->uniqueid);
909                 update_apps_list(watching_apps_all, watching_apps_from);
910         }
911
912         if (!ao2_container_count(watching_apps_all)) {
913                 return;
914         }
915
916         msg = ast_json_pack("{s: s, s: o, s: o, s: o}",
917                 "type", "BridgeMerged",
918                 "timestamp", ast_json_timeval(*tv, NULL),
919                 "bridge", ast_bridge_snapshot_to_json(merge->to),
920                 "bridge_from", ast_bridge_snapshot_to_json(merge->from));
921
922         if (!msg) {
923                 return;
924         }
925
926         distribute_message(watching_apps_all, msg);
927 }
928
929 static void sub_bridge_enter_handler(void *data,
930                 struct stasis_subscription *sub,
931                 struct stasis_topic *topic,
932                 struct stasis_message *message)
933 {
934         RAII_VAR(struct ao2_container *, watching_apps_channel, NULL, ao2_cleanup);
935         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
936         RAII_VAR(struct ao2_container *, watching_apps_all, ao2_container_alloc(1, NULL, NULL), ao2_cleanup);
937         struct ast_bridge_blob *obj = stasis_message_data(message);
938         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
939
940         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
941         if (watching_apps_bridge) {
942                 update_apps_list(watching_apps_all, watching_apps_bridge);
943         }
944
945         watching_apps_channel = get_apps_watching_channel(obj->channel->uniqueid);
946         if (watching_apps_channel) {
947                 update_bridge_interest(watching_apps_channel, obj->bridge->uniqueid);
948                 update_apps_list(watching_apps_all, watching_apps_channel);
949         }
950
951         if (!ao2_container_count(watching_apps_all)) {
952                 return;
953         }
954
955         msg = simple_bridge_channel_event("ChannelEnteredBridge", obj->bridge,
956                 obj->channel, stasis_message_timestamp(message));
957
958         distribute_message(watching_apps_all, msg);
959 }
960
961 static void sub_bridge_leave_handler(void *data,
962                 struct stasis_subscription *sub,
963                 struct stasis_topic *topic,
964                 struct stasis_message *message)
965 {
966         RAII_VAR(struct ao2_container *, watching_apps_bridge, NULL, ao2_cleanup);
967         struct ast_bridge_blob *obj = stasis_message_data(message);
968         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
969
970         watching_apps_bridge = get_apps_watching_bridge(obj->bridge->uniqueid);
971         if (!watching_apps_bridge) {
972                 return;
973         }
974
975         msg = simple_bridge_channel_event("ChannelLeftBridge", obj->bridge,
976                 obj->channel, stasis_message_timestamp(message));
977
978         distribute_message(watching_apps_bridge, msg);
979 }
980
981 static int load_module(void)
982 {
983         int r = 0;
984
985         apps_registry =
986                 ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
987         if (apps_registry == NULL) {
988                 return AST_MODULE_LOAD_FAILURE;
989         }
990
991         app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
992                                              control_hash, control_compare);
993         if (app_controls == NULL) {
994                 return AST_MODULE_LOAD_FAILURE;
995         }
996
997         app_bridges = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
998                                              bridges_hash, bridges_compare);
999         if (app_bridges == NULL) {
1000                 return AST_MODULE_LOAD_FAILURE;
1001         }
1002
1003         channel_router = stasis_message_router_create(ast_channel_topic_all_cached());
1004         if (!channel_router) {
1005                 return AST_MODULE_LOAD_FAILURE;
1006         }
1007
1008         r |= stasis_message_router_add(channel_router, stasis_cache_update_type(), sub_channel_snapshot_handler, NULL);
1009         /* TODO: This could be handled a lot better. Instead of subscribing to
1010          * the one caching topic and filtering out messages by channel id, we
1011          * should have individual caching topics per-channel, with a shared
1012          * back-end cache. That would simplify a lot of what's going on right
1013          * here.
1014          */
1015         r |= stasis_message_router_add(channel_router, ast_channel_user_event_type(), sub_channel_blob_handler, NULL);
1016         r |= stasis_message_router_add(channel_router, ast_channel_varset_type(), sub_channel_blob_handler, NULL);
1017         r |= stasis_message_router_add(channel_router, ast_channel_dtmf_end_type(), sub_channel_blob_handler, NULL);
1018         r |= stasis_message_router_add(channel_router, ast_channel_hangup_request_type(), sub_channel_blob_handler, NULL);
1019         if (r) {
1020                 return AST_MODULE_LOAD_FAILURE;
1021         }
1022
1023         bridge_router = stasis_message_router_create(ast_bridge_topic_all_cached());
1024         if (!bridge_router) {
1025                 return AST_MODULE_LOAD_FAILURE;
1026         }
1027
1028         r |= stasis_message_router_add(bridge_router, stasis_cache_update_type(), sub_bridge_snapshot_handler, NULL);
1029         r |= stasis_message_router_add(bridge_router, ast_bridge_merge_message_type(), sub_bridge_merge_handler, NULL);
1030         r |= stasis_message_router_add(bridge_router, ast_channel_entered_bridge_type(), sub_bridge_enter_handler, NULL);
1031         r |= stasis_message_router_add(bridge_router, ast_channel_left_bridge_type(), sub_bridge_leave_handler, NULL);
1032         if (r) {
1033                 return AST_MODULE_LOAD_FAILURE;
1034         }
1035
1036         return AST_MODULE_LOAD_SUCCESS;
1037 }
1038
1039 static int unload_module(void)
1040 {
1041         int r = 0;
1042
1043         stasis_message_router_unsubscribe_and_join(channel_router);
1044         channel_router = NULL;
1045
1046         stasis_message_router_unsubscribe_and_join(bridge_router);
1047         bridge_router = NULL;
1048
1049         ao2_cleanup(apps_registry);
1050         apps_registry = NULL;
1051
1052         ao2_cleanup(app_controls);
1053         app_controls = NULL;
1054
1055         ao2_cleanup(app_bridges);
1056         app_bridges = NULL;
1057
1058         return r;
1059 }
1060
1061 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application support",
1062         .load = load_module,
1063         .unload = unload_module,
1064         );