Moved core logic from app_stasis to res_stasis
[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
26 /*** MODULEINFO
27         <support_level>core</support_level>
28  ***/
29
30 #include "asterisk.h"
31
32 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
33
34 #include "asterisk/astobj2.h"
35 #include "asterisk/channel.h"
36 #include "asterisk/lock.h"
37 #include "asterisk/module.h"
38 #include "asterisk/stasis.h"
39 #include "asterisk/stasis_app.h"
40 #include "asterisk/stasis_channels.h"
41 #include "asterisk/strings.h"
42
43 /*!
44  * \brief Number of buckets for the Stasis application hash table.  Remember to
45  * keep it a prime number!
46  */
47 #define APPS_NUM_BUCKETS 127
48
49 /*!
50  * \brief Number of buckets for the Stasis application hash table.  Remember to
51  * keep it a prime number!
52  */
53 #define CONTROLS_NUM_BUCKETS 127
54
55 /*!
56  * \brief Stasis application container. Please call apps_registry() instead of
57  * directly accessing.
58  */
59 struct ao2_container *__apps_registry;
60
61 struct ao2_container *__app_controls;
62
63 /*! Ref-counting accessor for the stasis applications container */
64 static struct ao2_container *apps_registry(void)
65 {
66         ao2_ref(__apps_registry, +1);
67         return __apps_registry;
68 }
69
70 static struct ao2_container *app_controls(void)
71 {
72         ao2_ref(__app_controls, +1);
73         return __app_controls;
74 }
75
76 struct app {
77         /*! Callback function for this application. */
78         stasis_app_cb handler;
79         /*! Opaque data to hand to callback function. */
80         void *data;
81         /*! Name of the Stasis application */
82         char name[];
83 };
84
85 static void app_dtor(void *obj)
86 {
87         struct app *app = obj;
88
89         ao2_cleanup(app->data);
90         app->data = NULL;
91 }
92
93 /*! Constructor for \ref app. */
94 static struct app *app_create(const char *name, stasis_app_cb handler, void *data)
95 {
96         struct app *app;
97         size_t size;
98
99         ast_assert(name != NULL);
100         ast_assert(handler != NULL);
101
102         size = sizeof(*app) + strlen(name) + 1;
103         app = ao2_alloc_options(size, app_dtor, AO2_ALLOC_OPT_LOCK_MUTEX);
104
105         if (!app) {
106                 return NULL;
107         }
108
109         strncpy(app->name, name, size - sizeof(*app));
110         app->handler = handler;
111         ao2_ref(data, +1);
112         app->data = data;
113
114         return app;
115 }
116
117 /*! AO2 hash function for \ref app */
118 static int app_hash(const void *obj, const int flags)
119 {
120         const struct app *app = obj;
121         const char *name = flags & OBJ_KEY ? obj : app->name;
122
123         return ast_str_hash(name);
124 }
125
126 /*! AO2 comparison function for \ref app */
127 static int app_compare(void *lhs, void *rhs, int flags)
128 {
129         const struct app *lhs_app = lhs;
130         const struct app *rhs_app = rhs;
131         const char *rhs_name = flags & OBJ_KEY ? rhs : rhs_app->name;
132
133         if (strcmp(lhs_app->name, rhs_name) == 0) {
134                 return CMP_MATCH | CMP_STOP;
135         } else {
136                 return 0;
137         }
138 }
139
140 /*!
141  * \brief Send a message to the given application.
142  * \param app App to send the message to.
143  * \param message Message to send.
144  */
145 static void app_send(struct app *app, struct ast_json *message)
146 {
147         app->handler(app->data, app->name, message);
148 }
149
150 struct stasis_app_control {
151         /*!
152          * When set, /c app_stasis should exit and continue in the dialplan.
153          */
154         int continue_to_dialplan:1;
155         /*! Uniqueid of the associated channel */
156         char channel_id[];
157 };
158
159 static struct stasis_app_control *control_create(const char *uniqueid)
160 {
161         struct stasis_app_control *control;
162         size_t size;
163
164         size = sizeof(*control) + strlen(uniqueid) + 1;
165         control = ao2_alloc(size, NULL);
166         if (!control) {
167                 return NULL;
168         }
169
170         strncpy(control->channel_id, uniqueid, size - sizeof(*control));
171
172         return control;
173 }
174
175 /*! AO2 hash function for \ref stasis_app_control */
176 static int control_hash(const void *obj, const int flags)
177 {
178         const struct stasis_app_control *control = obj;
179         const char *id = flags & OBJ_KEY ? obj : control->channel_id;
180
181         return ast_str_hash(id);
182 }
183
184 /*! AO2 comparison function for \ref stasis_app_control */
185 static int control_compare(void *lhs, void *rhs, int flags)
186 {
187         const struct stasis_app_control *lhs_control = lhs;
188         const struct stasis_app_control *rhs_control = rhs;
189         const char *rhs_name =
190                 flags & OBJ_KEY ? rhs : rhs_control->channel_id;
191
192         if (strcmp(lhs_control->channel_id, rhs_name) == 0) {
193                 return CMP_MATCH | CMP_STOP;
194         } else {
195                 return 0;
196         }
197 }
198
199 struct stasis_app_control *stasis_app_control_find_by_channel(
200         const struct ast_channel *chan)
201 {
202         RAII_VAR(struct ao2_container *, controls, NULL, ao2_cleanup);
203         if (chan == NULL) {
204                 return NULL;
205         }
206
207         controls = app_controls();
208         return ao2_find(controls, ast_channel_uniqueid(chan), OBJ_KEY);
209 }
210
211 /*!
212  * \brief Test the \c continue_to_dialplan bit for the given \a app.
213  *
214  * The bit is also reset for the next call.
215  *
216  * \param app Application to check the \c continue_to_dialplan bit.
217  * \return Zero to remain in \c Stasis
218  * \return Non-zero to continue in the dialplan
219  */
220 static int control_continue_test_and_reset(struct stasis_app_control *control)
221 {
222         int r;
223         SCOPED_AO2LOCK(lock, control);
224
225         r = control->continue_to_dialplan;
226         control->continue_to_dialplan = 0;
227         return r;
228 }
229
230 void stasis_app_control_continue(struct stasis_app_control *control)
231 {
232         SCOPED_AO2LOCK(lock, control);
233         control->continue_to_dialplan = 1;
234 }
235
236 static struct ast_json *app_event_create(
237         const char *event_name,
238         const struct ast_channel_snapshot *snapshot,
239         const struct ast_json *extra_info)
240 {
241         RAII_VAR(struct ast_json *, message, NULL, ast_json_unref);
242         RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
243
244         if (extra_info) {
245                 event = ast_json_deep_copy(extra_info);
246         } else {
247                 event = ast_json_object_create();
248         }
249
250         if (snapshot) {
251                 int ret;
252
253                 /* Mustn't already have a channel field */
254                 ast_assert(ast_json_object_get(event, "channel") == NULL);
255
256                 ret = ast_json_object_set(
257                         event,
258                         "channel", ast_channel_snapshot_to_json(snapshot));
259                 if (ret != 0) {
260                         return NULL;
261                 }
262         }
263
264         message = ast_json_pack("{s: o}", event_name, ast_json_ref(event));
265
266         return ast_json_ref(message);
267 }
268
269 static int send_start_msg(struct app *app, struct ast_channel *chan,
270                           int argc, char *argv[])
271 {
272         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
273         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
274
275         struct ast_json *json_args;
276         int i;
277
278         ast_assert(chan != NULL);
279
280         /* Set channel info */
281         snapshot = ast_channel_snapshot_create(chan);
282         if (!snapshot) {
283                 return -1;
284         }
285
286         msg = ast_json_pack("{s: {s: [], s: o}}",
287                             "stasis-start",
288                             "args",
289                             "channel", ast_channel_snapshot_to_json(snapshot));
290
291         if (!msg) {
292                 return -1;
293         }
294
295         /* Append arguments to args array */
296         json_args = ast_json_object_get(
297                 ast_json_object_get(msg, "stasis-start"),
298                 "args");
299         ast_assert(json_args != NULL);
300         for (i = 0; i < argc; ++i) {
301                 int r = ast_json_array_append(json_args,
302                                               ast_json_string_create(argv[i]));
303                 if (r != 0) {
304                         ast_log(LOG_ERROR, "Error appending start message\n");
305                         return -1;
306                 }
307         }
308
309         app_send(app, msg);
310         return 0;
311 }
312
313 static int send_end_msg(struct app *app, struct ast_channel *chan)
314 {
315         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
316         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
317
318         ast_assert(chan != NULL);
319
320         /* Set channel info */
321         snapshot = ast_channel_snapshot_create(chan);
322         if (snapshot == NULL) {
323                 return -1;
324         }
325         msg = app_event_create("stasis-end", snapshot, NULL);
326         if (!msg) {
327                 return -1;
328         }
329
330         app_send(app, msg);
331         return 0;
332 }
333
334 static void dtmf_handler(struct app *app, struct ast_channel_blob *obj)
335 {
336         RAII_VAR(struct ast_json *, extra, NULL, ast_json_unref);
337         RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
338         const char *direction;
339
340         /* To simplify events, we'll only generate on receive */
341         direction = ast_json_string_get(
342                 ast_json_object_get(obj->blob, "direction"));
343
344         if (strcmp("Received", direction) != 0) {
345                 return;
346         }
347
348         extra = ast_json_pack(
349                 "{s: o}",
350                 "digit", ast_json_ref(ast_json_object_get(obj->blob, "digit")));
351         if (!extra) {
352                 return;
353         }
354
355         msg = app_event_create("dtmf-received", obj->snapshot, extra);
356         if (!msg) {
357                 return;
358         }
359
360         app_send(app, msg);
361 }
362
363 static void blob_handler(struct app *app, struct ast_channel_blob *blob)
364 {
365         /* To simplify events, we'll only generate on DTMF end */
366         if (strcmp(ast_channel_blob_json_type(blob), "dtmf_end") == 0) {
367                 dtmf_handler(app, blob);
368         }
369 }
370
371 static void sub_handler(void *data, struct stasis_subscription *sub,
372                         struct stasis_topic *topic,
373                         struct stasis_message *message)
374 {
375         struct app *app = data;
376         if (ast_channel_snapshot_type() == stasis_message_type(message)) {
377                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
378                 struct ast_channel_snapshot *snapshot =
379                         stasis_message_data(message);
380
381                 msg = app_event_create("channel-state-change", snapshot, NULL);
382                 if (!msg) {
383                         return;
384                 }
385                 app_send(app, msg);
386         } else if (ast_channel_blob_type() == stasis_message_type(message)) {
387                 struct ast_channel_blob *blob = stasis_message_data(message);
388                 blob_handler(app, blob);
389         }
390         if (stasis_subscription_final_message(sub, message)) {
391                 ao2_cleanup(data);
392         }
393 }
394
395 /*!
396  * \brief In addition to running ao2_cleanup(), this function also removes the
397  * object from the app_controls() container.
398  */
399 static void control_unlink(struct stasis_app_control *control)
400 {
401         RAII_VAR(struct ao2_container *, controls, NULL, ao2_cleanup);
402
403         if (!control) {
404                 return;
405         }
406
407         controls = app_controls();
408         ao2_unlink_flags(controls, control,
409                          OBJ_POINTER | OBJ_UNLINK | OBJ_NODATA);
410         ao2_cleanup(control);
411 }
412
413 /*! /brief Stasis dialplan application callback */
414 int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
415                     char *argv[])
416 {
417         RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
418         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
419         RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
420         RAII_VAR(struct stasis_subscription *, subscription, NULL,
421                  stasis_unsubscribe);
422         int res = 0;
423         int hungup = 0;
424
425         ast_assert(chan != NULL);
426
427         app = ao2_find(apps, app_name, OBJ_KEY);
428         if (!app) {
429                 ast_log(LOG_ERROR,
430                         "Stasis app '%s' not registered\n", app_name);
431                 return -1;
432         }
433
434         {
435                 RAII_VAR(struct ao2_container *, controls, NULL, ao2_cleanup);
436
437                 controls = app_controls();
438                 control = control_create(ast_channel_uniqueid(chan));
439                 if (!control) {
440                         ast_log(LOG_ERROR, "Allocated failed\n");
441                         return -1;
442                 }
443                 ao2_link(controls, control);
444         }
445
446         subscription =
447                 stasis_subscribe(ast_channel_topic(chan), sub_handler, app);
448         if (subscription == NULL) {
449                 ast_log(LOG_ERROR, "Error subscribing app %s to channel %s\n",
450                         app_name, ast_channel_name(chan));
451                 return -1;
452         }
453         ao2_ref(app, +1); /* subscription now has a reference */
454
455         res = send_start_msg(app, chan, argc, argv);
456         if (res != 0) {
457                 ast_log(LOG_ERROR, "Error sending start message to %s\n", app_name);
458                 return res;
459         }
460
461         while (!hungup && !control_continue_test_and_reset(control) && ast_waitfor(chan, -1) > -1) {
462                 RAII_VAR(struct ast_frame *, f, ast_read(chan), ast_frame_dtor);
463                 if (!f) {
464                         ast_debug(3, "%s: No more frames. Must be done, I guess.\n", ast_channel_uniqueid(chan));
465                         break;
466                 }
467
468                 switch (f->frametype) {
469                 case AST_FRAME_CONTROL:
470                         if (f->subclass.integer == AST_CONTROL_HANGUP) {
471                                 ast_debug(3, "%s: Received hangup\n",
472                                           ast_channel_uniqueid(chan));
473                                 hungup = 1;
474                         }
475                         break;
476                 default:
477                         /* Not handled; discard */
478                         break;
479                 }
480         }
481
482         res = send_end_msg(app, chan);
483         if (res != 0) {
484                 ast_log(LOG_ERROR,
485                         "Error sending end message to %s\n", app_name);
486                 return res;
487         }
488
489         return res;
490 }
491
492 int stasis_app_send(const char *app_name, struct ast_json *message)
493 {
494         RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
495         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
496
497         app = ao2_find(apps, app_name, OBJ_KEY);
498
499         if (!app) {
500                 /* XXX We can do a better job handling late binding, queueing up
501                  * the call for a few seconds to wait for the app to register.
502                  */
503                 ast_log(LOG_WARNING,
504                         "Stasis app '%s' not registered\n", app_name);
505                 return -1;
506         }
507
508         app_send(app, message);
509         return 0;
510 }
511
512 int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
513 {
514         RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
515         RAII_VAR(struct app *, app, NULL, ao2_cleanup);
516
517         SCOPED_LOCK(apps_lock, apps, ao2_lock, ao2_unlock);
518
519         app = ao2_find(apps, app_name, OBJ_KEY | OBJ_NOLOCK);
520
521         if (app) {
522                 RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
523                 SCOPED_LOCK(app_lock, app, ao2_lock, ao2_unlock);
524
525                 msg = app_event_create("application-replaced", NULL, NULL);
526                 app->handler(app->data, app_name, msg);
527
528                 app->handler = handler;
529                 ao2_cleanup(app->data);
530                 ao2_ref(data, +1);
531                 app->data = data;
532         } else {
533                 app = app_create(app_name, handler, data);
534                 if (app) {
535                         ao2_link_flags(apps, app, OBJ_NOLOCK);
536                 } else {
537                         return -1;
538                 }
539         }
540
541         return 0;
542 }
543
544 void stasis_app_unregister(const char *app_name)
545 {
546         RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
547
548         if (app_name) {
549                 apps = apps_registry();
550                 ao2_cleanup(ao2_find(apps, app_name, OBJ_KEY | OBJ_UNLINK));
551         }
552 }
553
554 static int load_module(void)
555 {
556         int r = 0;
557
558         __apps_registry =
559                 ao2_container_alloc(APPS_NUM_BUCKETS, app_hash, app_compare);
560         if (__apps_registry == NULL) {
561                 return AST_MODULE_LOAD_FAILURE;
562         }
563
564         __app_controls = ao2_container_alloc(CONTROLS_NUM_BUCKETS,
565                                              control_hash, control_compare);
566         if (__app_controls == NULL) {
567                 return AST_MODULE_LOAD_FAILURE;
568         }
569
570         return r;
571 }
572
573 static int unload_module(void)
574 {
575         int r = 0;
576
577         ao2_cleanup(__apps_registry);
578         __apps_registry = NULL;
579
580         ao2_cleanup(__app_controls);
581         __app_controls = NULL;
582
583         return r;
584 }
585
586 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS,
587                 "Stasis application support",
588                 .load = load_module,
589                 .unload = unload_module);