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