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