stasis: add functions embarrassingly missing from r400522
[asterisk/asterisk.git] / res / stasis / app.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 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
26 #include "asterisk.h"
27
28 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
29
30 #include "app.h"
31
32 #include "asterisk/callerid.h"
33 #include "asterisk/stasis_app.h"
34 #include "asterisk/stasis_bridges.h"
35 #include "asterisk/stasis_channels.h"
36 #include "asterisk/stasis_endpoints.h"
37 #include "asterisk/stasis_message_router.h"
38
39 struct app {
40         /*! Aggregation topic for this application. */
41         struct stasis_topic *topic;
42         /*! Router for handling messages forwarded to \a topic. */
43         struct stasis_message_router *router;
44         /*! Subscription to watch for bridge merge messages */
45         struct stasis_subscription *bridge_merge_sub;
46         /*! Container of the channel forwards to this app's topic. */
47         struct ao2_container *forwards;
48         /*! Callback function for this application. */
49         stasis_app_cb handler;
50         /*! Opaque data to hand to callback function. */
51         void *data;
52         /*! Name of the Stasis application */
53         char name[];
54 };
55
56 enum forward_type {
57         FORWARD_CHANNEL,
58         FORWARD_BRIDGE,
59         FORWARD_ENDPOINT,
60 };
61
62 /*! Subscription info for a particular channel/bridge. */
63 struct app_forwards {
64         /*! Count of number of times this channel/bridge has been subscribed */
65         int interested;
66
67         /*! Forward for the regular topic */
68         struct stasis_forward *topic_forward;
69         /*! Forward for the caching topic */
70         struct stasis_forward *topic_cached_forward;
71
72         /* Type of object being forwarded */
73         enum forward_type forward_type;
74         /*! Unique id of the object being forwarded */
75         char id[];
76 };
77
78 static void forwards_dtor(void *obj)
79 {
80 #ifdef AST_DEVMODE
81         struct app_forwards *forwards = obj;
82 #endif /* AST_DEVMODE */
83
84         ast_assert(forwards->topic_forward == NULL);
85         ast_assert(forwards->topic_cached_forward == NULL);
86 }
87
88 static void forwards_unsubscribe(struct app_forwards *forwards)
89 {
90         stasis_forward_cancel(forwards->topic_forward);
91         forwards->topic_forward = NULL;
92         stasis_forward_cancel(forwards->topic_cached_forward);
93         forwards->topic_cached_forward = NULL;
94 }
95
96 static struct app_forwards *forwards_create(struct app *app,
97         const char *id)
98 {
99         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
100
101         if (!app || ast_strlen_zero(id)) {
102                 return NULL;
103         }
104
105         forwards = ao2_alloc(sizeof(*forwards) + strlen(id) + 1, forwards_dtor);
106         if (!forwards) {
107                 return NULL;
108         }
109
110         strcpy(forwards->id, id);
111
112         ao2_ref(forwards, +1);
113         return forwards;
114 }
115
116 /*! Forward a channel's topics to an app */
117 static struct app_forwards *forwards_create_channel(struct app *app,
118         struct ast_channel *chan)
119 {
120         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
121
122         if (!app || !chan) {
123                 return NULL;
124         }
125
126         forwards = forwards_create(app, ast_channel_uniqueid(chan));
127         if (!forwards) {
128                 return NULL;
129         }
130
131         forwards->forward_type = FORWARD_CHANNEL;
132         forwards->topic_forward = stasis_forward_all(ast_channel_topic(chan),
133                 app->topic);
134         if (!forwards->topic_forward) {
135                 return NULL;
136         }
137
138         forwards->topic_cached_forward = stasis_forward_all(
139                 ast_channel_topic_cached(chan), app->topic);
140         if (!forwards->topic_cached_forward) {
141                 /* Half-subscribed is a bad thing */
142                 stasis_forward_cancel(forwards->topic_forward);
143                 forwards->topic_forward = NULL;
144                 return NULL;
145         }
146
147         ao2_ref(forwards, +1);
148         return forwards;
149 }
150
151 /*! Forward a bridge's topics to an app */
152 static struct app_forwards *forwards_create_bridge(struct app *app,
153         struct ast_bridge *bridge)
154 {
155         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
156
157         if (!app || !bridge) {
158                 return NULL;
159         }
160
161         forwards = forwards_create(app, bridge->uniqueid);
162         if (!forwards) {
163                 return NULL;
164         }
165
166         forwards->forward_type = FORWARD_BRIDGE;
167         forwards->topic_forward = stasis_forward_all(ast_bridge_topic(bridge),
168                 app->topic);
169         if (!forwards->topic_forward) {
170                 return NULL;
171         }
172
173         forwards->topic_cached_forward = stasis_forward_all(
174                 ast_bridge_topic_cached(bridge), app->topic);
175         if (!forwards->topic_cached_forward) {
176                 /* Half-subscribed is a bad thing */
177                 stasis_forward_cancel(forwards->topic_forward);
178                 forwards->topic_forward = NULL;
179                 return NULL;
180         }
181
182         ao2_ref(forwards, +1);
183         return forwards;
184 }
185
186 /*! Forward a endpoint's topics to an app */
187 static struct app_forwards *forwards_create_endpoint(struct app *app,
188         struct ast_endpoint *endpoint)
189 {
190         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
191
192         if (!app || !endpoint) {
193                 return NULL;
194         }
195
196         forwards = forwards_create(app, ast_endpoint_get_id(endpoint));
197         if (!forwards) {
198                 return NULL;
199         }
200
201         forwards->forward_type = FORWARD_ENDPOINT;
202         forwards->topic_forward = stasis_forward_all(ast_endpoint_topic(endpoint),
203                 app->topic);
204         if (!forwards->topic_forward) {
205                 return NULL;
206         }
207
208         forwards->topic_cached_forward = stasis_forward_all(
209                 ast_endpoint_topic_cached(endpoint), app->topic);
210         if (!forwards->topic_cached_forward) {
211                 /* Half-subscribed is a bad thing */
212                 stasis_forward_cancel(forwards->topic_forward);
213                 forwards->topic_forward = NULL;
214                 return NULL;
215         }
216
217         ao2_ref(forwards, +1);
218         return forwards;
219 }
220
221 static int forwards_sort(const void *obj_left, const void *obj_right, int flags)
222 {
223         const struct app_forwards *object_left = obj_left;
224         const struct app_forwards *object_right = obj_right;
225         const char *right_key = obj_right;
226         int cmp;
227
228         switch (flags & (OBJ_POINTER | OBJ_KEY | OBJ_PARTIAL_KEY)) {
229         case OBJ_POINTER:
230                 right_key = object_right->id;
231                 /* Fall through */
232         case OBJ_KEY:
233                 cmp = strcmp(object_left->id, right_key);
234                 break;
235         case OBJ_PARTIAL_KEY:
236                 /*
237                  * We could also use a partial key struct containing a length
238                  * so strlen() does not get called for every comparison instead.
239                  */
240                 cmp = strncmp(object_left->id, right_key, strlen(right_key));
241                 break;
242         default:
243                 /* Sort can only work on something with a full or partial key. */
244                 ast_assert(0);
245                 cmp = 0;
246                 break;
247         }
248         return cmp;
249 }
250
251 static void app_dtor(void *obj)
252 {
253         struct app *app = obj;
254
255         ast_verb(1, "Destroying Stasis app %s\n", app->name);
256
257         ast_assert(app->router == NULL);
258         ast_assert(app->bridge_merge_sub == NULL);
259
260         ao2_cleanup(app->topic);
261         app->topic = NULL;
262         ao2_cleanup(app->forwards);
263         app->forwards = NULL;
264         ao2_cleanup(app->data);
265         app->data = NULL;
266 }
267
268 static void sub_default_handler(void *data, struct stasis_subscription *sub,
269         struct stasis_message *message)
270 {
271         struct app *app = data;
272         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
273
274         if (stasis_subscription_final_message(sub, message)) {
275                 ao2_cleanup(app);
276         }
277
278         /* By default, send any message that has a JSON representation */
279         json = stasis_message_to_json(message);
280         if (!json) {
281                 return;
282         }
283
284         app_send(app, json);
285 }
286
287 /*! \brief Typedef for callbacks that get called on channel snapshot updates */
288 typedef struct ast_json *(*channel_snapshot_monitor)(
289         struct ast_channel_snapshot *old_snapshot,
290         struct ast_channel_snapshot *new_snapshot,
291         const struct timeval *tv);
292
293 static struct ast_json *simple_channel_event(
294         const char *type,
295         struct ast_channel_snapshot *snapshot,
296         const struct timeval *tv)
297 {
298         return ast_json_pack("{s: s, s: o, s: o}",
299                 "type", type,
300                 "timestamp", ast_json_timeval(*tv, NULL),
301                 "channel", ast_channel_snapshot_to_json(snapshot));
302 }
303
304 static struct ast_json *channel_created_event(
305         struct ast_channel_snapshot *snapshot,
306         const struct timeval *tv)
307 {
308         return simple_channel_event("ChannelCreated", snapshot, tv);
309 }
310
311 static struct ast_json *channel_destroyed_event(
312         struct ast_channel_snapshot *snapshot,
313         const struct timeval *tv)
314 {
315         return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
316                 "type", "ChannelDestroyed",
317                 "timestamp", ast_json_timeval(*tv, NULL),
318                 "cause", snapshot->hangupcause,
319                 "cause_txt", ast_cause2str(snapshot->hangupcause),
320                 "channel", ast_channel_snapshot_to_json(snapshot));
321 }
322
323 static struct ast_json *channel_state_change_event(
324         struct ast_channel_snapshot *snapshot,
325         const struct timeval *tv)
326 {
327         return simple_channel_event("ChannelStateChange", snapshot, tv);
328 }
329
330 /*! \brief Handle channel state changes */
331 static struct ast_json *channel_state(
332         struct ast_channel_snapshot *old_snapshot,
333         struct ast_channel_snapshot *new_snapshot,
334         const struct timeval *tv)
335 {
336         struct ast_channel_snapshot *snapshot = new_snapshot ?
337                 new_snapshot : old_snapshot;
338
339         if (!old_snapshot) {
340                 return channel_created_event(snapshot, tv);
341         } else if (!new_snapshot) {
342                 return channel_destroyed_event(snapshot, tv);
343         } else if (old_snapshot->state != new_snapshot->state) {
344                 return channel_state_change_event(snapshot, tv);
345         }
346
347         return NULL;
348 }
349
350 static struct ast_json *channel_dialplan(
351         struct ast_channel_snapshot *old_snapshot,
352         struct ast_channel_snapshot *new_snapshot,
353         const struct timeval *tv)
354 {
355         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
356
357         /* No Newexten event on cache clear or first event */
358         if (!old_snapshot || !new_snapshot) {
359                 return NULL;
360         }
361
362         /* Empty application is not valid for a Newexten event */
363         if (ast_strlen_zero(new_snapshot->appl)) {
364                 return NULL;
365         }
366
367         if (ast_channel_snapshot_cep_equal(old_snapshot, new_snapshot)) {
368                 return NULL;
369         }
370
371         return ast_json_pack("{s: s, s: o, s: s, s: s, s: o}",
372                 "type", "ChannelDialplan",
373                 "timestamp", ast_json_timeval(*tv, NULL),
374                 "dialplan_app", new_snapshot->appl,
375                 "dialplan_app_data", new_snapshot->data,
376                 "channel", ast_channel_snapshot_to_json(new_snapshot));
377 }
378
379 static struct ast_json *channel_callerid(
380         struct ast_channel_snapshot *old_snapshot,
381         struct ast_channel_snapshot *new_snapshot,
382         const struct timeval *tv)
383 {
384         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
385
386         /* No NewCallerid event on cache clear or first event */
387         if (!old_snapshot || !new_snapshot) {
388                 return NULL;
389         }
390
391         if (ast_channel_snapshot_caller_id_equal(old_snapshot, new_snapshot)) {
392                 return NULL;
393         }
394
395         return ast_json_pack("{s: s, s: o, s: i, s: s, s: o}",
396                 "type", "ChannelCallerId",
397                 "timestamp", ast_json_timeval(*tv, NULL),
398                 "caller_presentation", new_snapshot->caller_pres,
399                 "caller_presentation_txt", ast_describe_caller_presentation(
400                         new_snapshot->caller_pres),
401                 "channel", ast_channel_snapshot_to_json(new_snapshot));
402 }
403
404 static channel_snapshot_monitor channel_monitors[] = {
405         channel_state,
406         channel_dialplan,
407         channel_callerid
408 };
409
410 static void sub_channel_update_handler(void *data,
411         struct stasis_subscription *sub,
412         struct stasis_message *message)
413 {
414         struct app *app = data;
415         struct stasis_cache_update *update;
416         struct ast_channel_snapshot *new_snapshot;
417         struct ast_channel_snapshot *old_snapshot;
418         const struct timeval *tv;
419         int i;
420
421         ast_assert(stasis_message_type(message) == stasis_cache_update_type());
422
423         update = stasis_message_data(message);
424
425         ast_assert(update->type == ast_channel_snapshot_type());
426
427         new_snapshot = stasis_message_data(update->new_snapshot);
428         old_snapshot = stasis_message_data(update->old_snapshot);
429
430         /* Pull timestamp from the new snapshot, or from the update message
431          * when there isn't one. */
432         tv = update->new_snapshot ?
433                 stasis_message_timestamp(update->new_snapshot) :
434                 stasis_message_timestamp(message);
435
436         for (i = 0; i < ARRAY_LEN(channel_monitors); ++i) {
437                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
438
439                 msg = channel_monitors[i](old_snapshot, new_snapshot, tv);
440                 if (msg) {
441                         app_send(app, msg);
442                 }
443         }
444 }
445
446 static struct ast_json *simple_endpoint_event(
447         const char *type,
448         struct ast_endpoint_snapshot *snapshot,
449         const struct timeval *tv)
450 {
451         return ast_json_pack("{s: s, s: o, s: o}",
452                 "type", type,
453                 "timestamp", ast_json_timeval(*tv, NULL),
454                 "endpoint", ast_endpoint_snapshot_to_json(snapshot));
455 }
456
457 static void sub_endpoint_update_handler(void *data,
458         struct stasis_subscription *sub,
459         struct stasis_message *message)
460 {
461         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
462         struct app *app = data;
463         struct stasis_cache_update *update;
464         struct ast_endpoint_snapshot *new_snapshot;
465         const struct timeval *tv;
466
467         ast_assert(stasis_message_type(message) == stasis_cache_update_type());
468
469         update = stasis_message_data(message);
470
471         ast_assert(update->type == ast_endpoint_snapshot_type());
472
473         new_snapshot = stasis_message_data(update->new_snapshot);
474         tv = update->new_snapshot ?
475                 stasis_message_timestamp(update->new_snapshot) :
476                 stasis_message_timestamp(message);
477
478         json = simple_endpoint_event("EndpointStateChange", new_snapshot, tv);
479
480         if (!json) {
481                 return;
482         }
483
484         app_send(app, json);
485 }
486
487 static struct ast_json *simple_bridge_event(
488         const char *type,
489         struct ast_bridge_snapshot *snapshot,
490         const struct timeval *tv)
491 {
492         return ast_json_pack("{s: s, s: o, s: o}",
493                 "type", type,
494                 "timestamp", ast_json_timeval(*tv, NULL),
495                 "bridge", ast_bridge_snapshot_to_json(snapshot));
496 }
497
498 static void sub_bridge_update_handler(void *data,
499         struct stasis_subscription *sub,
500         struct stasis_message *message)
501 {
502         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
503         struct app *app = data;
504         struct stasis_cache_update *update;
505         struct ast_bridge_snapshot *new_snapshot;
506         struct ast_bridge_snapshot *old_snapshot;
507         const struct timeval *tv;
508
509         ast_assert(stasis_message_type(message) == stasis_cache_update_type());
510
511         update = stasis_message_data(message);
512
513         ast_assert(update->type == ast_bridge_snapshot_type());
514
515         new_snapshot = stasis_message_data(update->new_snapshot);
516         old_snapshot = stasis_message_data(update->old_snapshot);
517         tv = update->new_snapshot ?
518                 stasis_message_timestamp(update->new_snapshot) :
519                 stasis_message_timestamp(message);
520
521         if (!new_snapshot) {
522                 json = simple_bridge_event("BridgeDestroyed", old_snapshot, tv);
523         } else if (!old_snapshot) {
524                 json = simple_bridge_event("BridgeCreated", old_snapshot, tv);
525         }
526
527         if (!json) {
528                 return;
529         }
530
531         app_send(app, json);
532 }
533
534 static void bridge_merge_handler(void *data, struct stasis_subscription *sub,
535         struct stasis_message *message)
536 {
537         struct app *app = data;
538         struct ast_bridge_merge_message *merge;
539         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
540
541         if (stasis_subscription_final_message(sub, message)) {
542                 ao2_cleanup(app);
543         }
544
545         if (stasis_message_type(message) != ast_bridge_merge_message_type()) {
546                 return;
547         }
548
549         merge = stasis_message_data(message);
550
551         /* Find out if we're subscribed to either bridge */
552         forwards = ao2_find(app->forwards, merge->from->uniqueid,
553                 OBJ_SEARCH_KEY);
554         if (!forwards) {
555                 forwards = ao2_find(app->forwards, merge->to->uniqueid,
556                         OBJ_SEARCH_KEY);
557         }
558
559         if (!forwards) {
560                 return;
561         }
562
563         /* Forward the message to the app */
564         stasis_publish(app->topic, message);
565 }
566
567 struct app *app_create(const char *name, stasis_app_cb handler, void *data)
568 {
569         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
570         size_t size;
571         int res = 0;
572
573         ast_assert(name != NULL);
574         ast_assert(handler != NULL);
575
576         ast_verb(1, "Creating Stasis app '%s'\n", name);
577
578         size = sizeof(*app) + strlen(name) + 1;
579         app = ao2_alloc_options(size, app_dtor, AO2_ALLOC_OPT_LOCK_MUTEX);
580
581         if (!app) {
582                 return NULL;
583         }
584
585         app->forwards = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_MUTEX,
586                 AO2_CONTAINER_ALLOC_OPT_DUPS_OBJ_REJECT,
587                 forwards_sort, NULL);
588         if (!app->forwards) {
589                 return NULL;
590         }
591
592         app->topic = stasis_topic_create(name);
593         if (!app->topic) {
594                 return NULL;
595         }
596
597         app->bridge_merge_sub = stasis_subscribe(ast_bridge_topic_all(),
598                 bridge_merge_handler, app);
599         if (!app->bridge_merge_sub) {
600                 return NULL;
601         }
602         /* Subscription holds a reference */
603         ao2_ref(app, +1);
604
605         app->router = stasis_message_router_create(app->topic);
606         if (!app->router) {
607                 return NULL;
608         }
609
610         res |= stasis_message_router_add_cache_update(app->router,
611                 ast_bridge_snapshot_type(), sub_bridge_update_handler, app);
612
613         res |= stasis_message_router_add_cache_update(app->router,
614                 ast_channel_snapshot_type(), sub_channel_update_handler, app);
615
616         res |= stasis_message_router_add_cache_update(app->router,
617                 ast_endpoint_snapshot_type(), sub_endpoint_update_handler, app);
618
619         res |= stasis_message_router_set_default(app->router,
620                 sub_default_handler, app);
621
622         if (res != 0) {
623                 return NULL;
624         }
625         /* Router holds a reference */
626         ao2_ref(app, +1);
627
628         strncpy(app->name, name, size - sizeof(*app));
629         app->handler = handler;
630         ao2_ref(data, +1);
631         app->data = data;
632
633         ao2_ref(app, +1);
634         return app;
635 }
636
637 /*!
638  * \brief Send a message to the given application.
639  * \param app App to send the message to.
640  * \param message Message to send.
641  */
642 void app_send(struct app *app, struct ast_json *message)
643 {
644         stasis_app_cb handler;
645         RAII_VAR(void *, data, NULL, ao2_cleanup);
646
647         /* Copy off mutable state with lock held */
648         {
649                 SCOPED_AO2LOCK(lock, app);
650                 handler = app->handler;
651                 if (app->data) {
652                         ao2_ref(app->data, +1);
653                         data = app->data;
654                 }
655                 /* Name is immutable; no need to copy */
656         }
657
658         if (!handler) {
659                 ast_verb(3,
660                         "Inactive Stasis app '%s' missed message\n", app->name);
661                 return;
662         }
663
664         handler(data, app->name, message);
665 }
666
667 void app_deactivate(struct app *app)
668 {
669         SCOPED_AO2LOCK(lock, app);
670         ast_verb(1, "Deactivating Stasis app '%s'\n", app->name);
671         app->handler = NULL;
672         ao2_cleanup(app->data);
673         app->data = NULL;
674 }
675
676 void app_shutdown(struct app *app)
677 {
678         SCOPED_AO2LOCK(lock, app);
679
680         ast_assert(app_is_finished(app));
681
682         stasis_message_router_unsubscribe(app->router);
683         app->router = NULL;
684         stasis_unsubscribe(app->bridge_merge_sub);
685         app->bridge_merge_sub = NULL;
686 }
687
688 int app_is_active(struct app *app)
689 {
690         SCOPED_AO2LOCK(lock, app);
691         return app->handler != NULL;
692 }
693
694 int app_is_finished(struct app *app)
695 {
696         SCOPED_AO2LOCK(lock, app);
697
698         return app->handler == NULL && ao2_container_count(app->forwards) == 0;
699 }
700
701 void app_update(struct app *app, stasis_app_cb handler, void *data)
702 {
703         SCOPED_AO2LOCK(lock, app);
704
705         if (app->handler) {
706                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
707
708                 ast_verb(1, "Replacing Stasis app '%s'\n", app->name);
709
710                 msg = ast_json_pack("{s: s, s: s}",
711                         "type", "ApplicationReplaced",
712                         "application", app->name);
713                 if (msg) {
714                         app_send(app, msg);
715                 }
716         } else {
717                 ast_verb(1, "Activating Stasis app '%s'\n", app->name);
718         }
719
720         app->handler = handler;
721         ao2_cleanup(app->data);
722         if (data) {
723                 ao2_ref(data, +1);
724         }
725         app->data = data;
726 }
727
728 const char *app_name(const struct app *app)
729 {
730         return app->name;
731 }
732
733 struct ast_json *app_to_json(const struct app *app)
734 {
735         RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
736         struct ast_json *channels;
737         struct ast_json *bridges;
738         struct ast_json *endpoints;
739         struct ao2_iterator i;
740         void *obj;
741
742         json = ast_json_pack("{s: s, s: [], s: [], s: []}",
743                 "name", app->name,
744                 "channel_ids", "bridge_ids", "endpoint_ids");
745         channels = ast_json_object_get(json, "channel_ids");
746         bridges = ast_json_object_get(json, "bridge_ids");
747         endpoints = ast_json_object_get(json, "endpoint_ids");
748
749         i = ao2_iterator_init(app->forwards, 0);
750         while ((obj = ao2_iterator_next(&i))) {
751                 RAII_VAR(struct app_forwards *, forwards, obj, ao2_cleanup);
752                 RAII_VAR(struct ast_json *, id, NULL, ast_json_unref);
753                 int append_res = -1;
754
755                 id = ast_json_string_create(forwards->id);
756
757                 switch (forwards->forward_type) {
758                 case FORWARD_CHANNEL:
759                         append_res = ast_json_array_append(channels,
760                                 ast_json_ref(id));
761                         break;
762                 case FORWARD_BRIDGE:
763                         append_res = ast_json_array_append(bridges,
764                                 ast_json_ref(id));
765                         break;
766                 case FORWARD_ENDPOINT:
767                         append_res = ast_json_array_append(endpoints,
768                                 ast_json_ref(id));
769                         break;
770                 }
771
772                 if (append_res != 0) {
773                         ast_log(LOG_ERROR, "Error building response\n");
774                         ao2_iterator_destroy(&i);
775                         return NULL;
776                 }
777         }
778         ao2_iterator_destroy(&i);
779
780         return ast_json_ref(json);
781 }
782
783 int app_subscribe_channel(struct app *app, struct ast_channel *chan)
784 {
785         int res;
786
787         if (!app || !chan) {
788                 return -1;
789         } else {
790                 RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
791                 SCOPED_AO2LOCK(lock, app->forwards);
792
793                 forwards = ao2_find(app->forwards, ast_channel_uniqueid(chan),
794                         OBJ_SEARCH_KEY | OBJ_NOLOCK);
795                 if (!forwards) {
796                         /* Forwards not found, create one */
797                         forwards = forwards_create_channel(app, chan);
798                         if (!forwards) {
799                                 return -1;
800                         }
801
802                         res = ao2_link_flags(app->forwards, forwards,
803                                 OBJ_NOLOCK);
804                         if (!res) {
805                                 return -1;
806                         }
807                 }
808
809                 ++forwards->interested;
810                 return 0;
811         }
812 }
813
814 static int unsubscribe(struct app *app, const char *kind, const char *id)
815 {
816         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
817         SCOPED_AO2LOCK(lock, app->forwards);
818
819         forwards = ao2_find(app->forwards, id, OBJ_SEARCH_KEY | OBJ_NOLOCK);
820         if (!forwards) {
821                 ast_log(LOG_WARNING,
822                         "App '%s' not subscribed to %s '%s'\n",
823                         app->name, kind, id);
824                 return -1;
825         }
826
827         if (--forwards->interested == 0) {
828                 /* No one is interested any more; unsubscribe */
829                 forwards_unsubscribe(forwards);
830                 ao2_find(app->forwards, forwards,
831                         OBJ_POINTER | OBJ_NOLOCK | OBJ_UNLINK |
832                         OBJ_NODATA);
833         }
834
835         return 0;
836 }
837
838 int app_unsubscribe_channel(struct app *app, struct ast_channel *chan)
839 {
840         if (!app || !chan) {
841                 return -1;
842         }
843
844         return app_unsubscribe_channel_id(app, ast_channel_uniqueid(chan));
845 }
846
847 int app_unsubscribe_channel_id(struct app *app, const char *channel_id)
848 {
849         if (!app || !channel_id) {
850                 return -1;
851         }
852
853         return unsubscribe(app, "channel", channel_id);
854 }
855
856 int app_is_subscribed_channel_id(struct app *app, const char *channel_id)
857 {
858         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
859         forwards = ao2_find(app->forwards, channel_id, OBJ_SEARCH_KEY);
860         return forwards != NULL;
861 }
862
863 int app_subscribe_bridge(struct app *app, struct ast_bridge *bridge)
864 {
865         if (!app || !bridge) {
866                 return -1;
867         } else {
868                 RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
869                 SCOPED_AO2LOCK(lock, app->forwards);
870
871                 forwards = ao2_find(app->forwards, bridge->uniqueid,
872                         OBJ_SEARCH_KEY | OBJ_NOLOCK);
873
874                 if (!forwards) {
875                         /* Forwards not found, create one */
876                         forwards = forwards_create_bridge(app, bridge);
877                         if (!forwards) {
878                                 return -1;
879                         }
880                         ao2_link_flags(app->forwards, forwards, OBJ_NOLOCK);
881                 }
882
883                 ++forwards->interested;
884                 return 0;
885         }
886 }
887
888 int app_unsubscribe_bridge(struct app *app, struct ast_bridge *bridge)
889 {
890         if (!app || !bridge) {
891                 return -1;
892         }
893
894         return app_unsubscribe_bridge_id(app, bridge->uniqueid);
895 }
896
897 int app_unsubscribe_bridge_id(struct app *app, const char *bridge_id)
898 {
899         if (!app || !bridge_id) {
900                 return -1;
901         }
902
903         return unsubscribe(app, "bridge", bridge_id);
904 }
905
906 int app_is_subscribed_bridge_id(struct app *app, const char *bridge_id)
907 {
908         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
909         forwards = ao2_find(app->forwards, bridge_id, OBJ_SEARCH_KEY);
910         return forwards != NULL;
911 }
912
913 int app_subscribe_endpoint(struct app *app, struct ast_endpoint *endpoint)
914 {
915         if (!app || !endpoint) {
916                 return -1;
917         } else {
918                 RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
919                 SCOPED_AO2LOCK(lock, app->forwards);
920
921                 forwards = ao2_find(app->forwards, ast_endpoint_get_id(endpoint),
922                         OBJ_SEARCH_KEY | OBJ_NOLOCK);
923
924                 if (!forwards) {
925                         /* Forwards not found, create one */
926                         forwards = forwards_create_endpoint(app, endpoint);
927                         if (!forwards) {
928                                 return -1;
929                         }
930                         ao2_link_flags(app->forwards, forwards, OBJ_NOLOCK);
931                 }
932
933                 ++forwards->interested;
934                 return 0;
935         }
936 }
937
938 int app_unsubscribe_endpoint_id(struct app *app, const char *endpoint_id)
939 {
940         if (!app || !endpoint_id) {
941                 return -1;
942         }
943
944         return unsubscribe(app, "endpoint", endpoint_id);
945 }
946
947 int app_is_subscribed_endpoint_id(struct app *app, const char *endpoint_id)
948 {
949         RAII_VAR(struct app_forwards *, forwards, NULL, ao2_cleanup);
950         forwards = ao2_find(app->forwards, endpoint_id, OBJ_SEARCH_KEY);
951         return forwards != NULL;
952 }