ARI: Add the ability to subscribe to all events
[asterisk/asterisk.git] / res / stasis / messaging.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2014, Digium, Inc.
5  *
6  * Matt Jordan <mjordan@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 /*!
20  * \file
21  *
22  * \brief Stasis out-of-call text message support
23  *
24  * \author Matt Jordan <mjordan@digium.com>
25  */
26
27 #include "asterisk.h"
28
29 ASTERISK_REGISTER_FILE()
30
31 #include "asterisk/message.h"
32 #include "asterisk/endpoints.h"
33 #include "asterisk/astobj2.h"
34 #include "asterisk/vector.h"
35 #include "asterisk/lock.h"
36 #include "asterisk/utils.h"
37 #include "asterisk/test.h"
38 #include "messaging.h"
39
40 /*!
41  * \brief Subscription to all technologies
42  */
43 #define TECH_WILDCARD "__AST_ALL_TECH"
44
45 /*!
46  * \brief Number of buckets for the \ref endpoint_subscriptions container
47  */
48 #define ENDPOINTS_NUM_BUCKETS 127
49
50 /*! \brief Storage object for an application */
51 struct application_tuple {
52         /*! ao2 ref counted private object to pass to the callback */
53         void *pvt;
54         /*! The callback to call when this application has a message */
55         message_received_cb callback;
56         /*! The name (key) of the application */
57         char app_name[];
58 };
59
60 /*! \brief A subscription to some endpoint or technology */
61 struct message_subscription {
62         /*! The applications that have subscribed to this endpoint or tech */
63         AST_VECTOR(, struct application_tuple *) applications;
64         /*! The name of this endpoint or tech */
65         char token[];
66 };
67
68 /*! \brief The subscriptions to endpoints */
69 static struct ao2_container *endpoint_subscriptions;
70
71 /*!
72  * \brief The subscriptions to technologies
73  *
74  * \note These are stored separately from standard endpoints, given how
75  * relatively few of them there are.
76  */
77 static AST_VECTOR(,struct message_subscription *) tech_subscriptions;
78
79 /*! \brief RWLock for \c tech_subscriptions */
80 static ast_rwlock_t tech_subscriptions_lock;
81
82 /*! \internal \brief Destructor for \c application_tuple */
83 static void application_tuple_dtor(void *obj)
84 {
85         struct application_tuple *tuple = obj;
86
87         ao2_cleanup(tuple->pvt);
88 }
89
90 /*! \internal \brief Constructor for \c application_tuple */
91 static struct application_tuple *application_tuple_alloc(const char *app_name, message_received_cb callback, void *pvt)
92 {
93         struct application_tuple *tuple;
94         size_t size = sizeof(*tuple) + strlen(app_name) + 1;
95
96         ast_assert(callback != NULL);
97
98         tuple = ao2_alloc_options(size, application_tuple_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
99         if (!tuple) {
100                 return NULL;
101         }
102
103         strcpy(tuple->app_name, app_name); /* Safe */
104         tuple->pvt = ao2_bump(pvt);
105         tuple->callback = callback;
106
107         return tuple;
108 }
109
110 /*! \internal \brief Destructor for \ref message_subscription */
111 static void message_subscription_dtor(void *obj)
112 {
113         struct message_subscription *sub = obj;
114         int i;
115
116         for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
117                 struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
118
119                 ao2_cleanup(tuple);
120         }
121         AST_VECTOR_FREE(&sub->applications);
122 }
123
124 /*! \internal \brief Constructor for \ref message_subscription */
125 static struct message_subscription *message_subscription_alloc(const char *token)
126 {
127         struct message_subscription *sub;
128         size_t size = sizeof(*sub) + strlen(token) + 1;
129
130         sub = ao2_alloc_options(size, message_subscription_dtor, AO2_ALLOC_OPT_LOCK_RWLOCK);
131         if (!sub) {
132                 return NULL;
133         }
134         strcpy(sub->token, token); /* Safe */
135
136         return sub;
137 }
138
139 /*! AO2 hash function for \ref message_subscription */
140 static int message_subscription_hash_cb(const void *obj, const int flags)
141 {
142         const struct message_subscription *sub;
143         const char *key;
144
145         switch (flags & OBJ_SEARCH_MASK) {
146         case OBJ_SEARCH_KEY:
147                 key = obj;
148                 break;
149         case OBJ_SEARCH_OBJECT:
150                 sub = obj;
151                 key = sub->token;
152                 break;
153         default:
154                 /* Hash can only work on something with a full key. */
155                 ast_assert(0);
156                 return 0;
157         }
158         return ast_str_hash(key);
159 }
160
161 /*! AO2 comparison function for \ref message_subscription */
162 static int message_subscription_compare_cb(void *obj, void *arg, int flags)
163 {
164         const struct message_subscription *object_left = obj;
165         const struct message_subscription *object_right = arg;
166         const char *right_key = arg;
167         int cmp;
168
169         switch (flags & OBJ_SEARCH_MASK) {
170         case OBJ_SEARCH_OBJECT:
171                 right_key = object_right->token;
172                 /* Fall through */
173         case OBJ_SEARCH_KEY:
174                 cmp = strcmp(object_left->token, right_key);
175                 break;
176         case OBJ_SEARCH_PARTIAL_KEY:
177                 /*
178                  * We could also use a partial key struct containing a length
179                  * so strlen() does not get called for every comparison instead.
180                  */
181                 cmp = strncmp(object_left->token, right_key, strlen(right_key));
182                 break;
183         default:
184                 /*
185                  * What arg points to is specific to this traversal callback
186                  * and has no special meaning to astobj2.
187                  */
188                 cmp = 0;
189                 break;
190         }
191         if (cmp) {
192                 return 0;
193         }
194         /*
195          * At this point the traversal callback is identical to a sorted
196          * container.
197          */
198         return CMP_MATCH;
199 }
200
201 /*! \internal \brief Convert a \c ast_msg To/From URI to a Stasis endpoint name */
202 static void msg_to_endpoint(const struct ast_msg *msg, char *buf, size_t len)
203 {
204         const char *endpoint = ast_msg_get_endpoint(msg);
205
206         snprintf(buf, len, "%s%s%s", ast_msg_get_tech(msg),
207                 ast_strlen_zero(endpoint) ? "" : "/",
208                 S_OR(endpoint, ""));
209 }
210
211 /*! \internal
212  * \brief Callback from the \c message API that determines if we can handle
213  * this message
214  */
215 static int has_destination_cb(const struct ast_msg *msg)
216 {
217         struct message_subscription *sub;
218         int i;
219         char buf[256];
220
221         msg_to_endpoint(msg, buf, sizeof(buf));
222
223         ast_rwlock_rdlock(&tech_subscriptions_lock);
224         for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
225                 sub = AST_VECTOR_GET(&tech_subscriptions, i);
226
227                 if (!sub) {
228                         continue;
229                 }
230
231                 if (!strcmp(sub->token, TECH_WILDCARD)
232                     || !strncasecmp(sub->token, buf, strlen(sub->token))
233                     || !strncasecmp(sub->token, buf, strlen(sub->token))) {
234                         ast_rwlock_unlock(&tech_subscriptions_lock);
235                         goto match;
236                 }
237
238         }
239         ast_rwlock_unlock(&tech_subscriptions_lock);
240
241         sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
242         if (sub) {
243                 ao2_ref(sub, -1);
244                 goto match;
245         }
246
247         ast_debug(1, "No subscription found for %s\n", buf);
248         return 0;
249
250 match:
251         return 1;
252 }
253
254 static struct ast_json *msg_to_json(struct ast_msg *msg)
255 {
256         struct ast_json *json_obj;
257         struct ast_json *json_vars;
258         struct ast_msg_var_iterator *it_vars;
259         const char *name;
260         const char *value;
261
262         it_vars = ast_msg_var_iterator_init(msg);
263         if (!it_vars) {
264                 return NULL;
265         }
266
267         json_vars = ast_json_array_create();
268         if (!json_vars) {
269                 return NULL;
270         }
271
272         while (ast_msg_var_iterator_next(msg, it_vars, &name, &value)) {
273                 struct ast_json *json_tuple;
274
275                 json_tuple = ast_json_pack("{s: s}", name, value);
276                 if (!json_tuple) {
277                         ast_json_free(json_vars);
278                         return NULL;
279                 }
280
281                 ast_json_array_append(json_vars, json_tuple);
282                 ast_msg_var_unref_current(it_vars);
283         }
284         ast_msg_var_iterator_destroy(it_vars);
285
286         json_obj = ast_json_pack("{s: s, s: s, s: s, s: o}",
287                 "from", ast_msg_get_from(msg),
288                 "to", ast_msg_get_to(msg),
289                 "body", ast_msg_get_body(msg),
290                 "variables", json_vars);
291
292         return json_obj;
293 }
294
295 static int handle_msg_cb(struct ast_msg *msg)
296 {
297         struct message_subscription *sub;
298         int i;
299         char buf[256];
300         const char *endpoint_name;
301         struct ast_json *json_msg;
302
303         msg_to_endpoint(msg, buf, sizeof(buf));
304
305         ast_rwlock_rdlock(&tech_subscriptions_lock);
306         for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
307                 sub = AST_VECTOR_GET(&tech_subscriptions, i);
308
309                 if (!sub) {
310                         continue;
311                 }
312
313                 if (!strcmp(sub->token, TECH_WILDCARD)
314                     || !strncasecmp(sub->token, buf, strlen(sub->token))) {
315                         ast_rwlock_unlock(&tech_subscriptions_lock);
316                         ao2_bump(sub);
317                         endpoint_name = buf;
318                         goto match;
319                 }
320         }
321         ast_rwlock_unlock(&tech_subscriptions_lock);
322
323         sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
324         if (sub) {
325                 endpoint_name = buf;
326                 goto match;
327         }
328
329         return -1;
330
331 match:
332         ast_debug(3, "Dispatching message for %s\n", endpoint_name);
333
334         json_msg = msg_to_json(msg);
335         if (!json_msg) {
336                 ao2_ref(sub, -1);
337                 return -1;
338         }
339
340         for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
341                 struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
342
343                 tuple->callback(endpoint_name, json_msg, tuple->pvt);
344         }
345
346         ast_json_unref(json_msg);
347         ao2_ref(sub, -1);
348         return 0;
349 }
350
351 struct ast_msg_handler ari_msg_handler = {
352         .name = "ari",
353         .handle_msg = handle_msg_cb,
354         .has_destination = has_destination_cb,
355 };
356
357 static int messaging_subscription_cmp(struct message_subscription *sub, const char *key)
358 {
359         return !strcmp(sub->token, key) ? 1 : 0;
360 }
361
362 static int application_tuple_cmp(struct application_tuple *item, const char *key)
363 {
364         return !strcmp(item->app_name, key) ? 1 : 0;
365 }
366
367 static int is_app_subscribed(struct message_subscription *sub, const char *app_name)
368 {
369         int i;
370
371         for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
372                 struct application_tuple *tuple;
373
374                 tuple = AST_VECTOR_GET(&sub->applications, i);
375                 if (tuple && !strcmp(tuple->app_name, app_name)) {
376                         return 1;
377                 }
378         }
379
380         return 0;
381 }
382
383 static struct message_subscription *get_subscription(struct ast_endpoint *endpoint)
384 {
385         struct message_subscription *sub = NULL;
386
387         if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
388                 sub = ao2_find(endpoint_subscriptions, endpoint, OBJ_SEARCH_KEY);
389         } else {
390                 int i;
391
392                 ast_rwlock_rdlock(&tech_subscriptions_lock);
393                 for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
394                         sub = AST_VECTOR_GET(&tech_subscriptions, i);
395
396                         if (sub && !strcmp(sub->token, endpoint ? ast_endpoint_get_tech(endpoint) : TECH_WILDCARD)) {
397                                 ao2_bump(sub);
398                                 break;
399                         }
400                 }
401                 ast_rwlock_unlock(&tech_subscriptions_lock);
402         }
403
404         return sub;
405 }
406
407 void messaging_app_unsubscribe_endpoint(const char *app_name, const char *endpoint_id)
408 {
409         RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
410         RAII_VAR(struct ast_endpoint *, endpoint, NULL, ao2_cleanup);
411
412         endpoint = ast_endpoint_find_by_id(endpoint_id);
413         sub = get_subscription(endpoint);
414         if (!sub) {
415                 return;
416         }
417
418         ao2_lock(sub);
419         if (!is_app_subscribed(sub, app_name)) {
420                 ao2_unlock(sub);
421                 return;
422         }
423
424         AST_VECTOR_REMOVE_CMP_UNORDERED(&sub->applications, app_name, application_tuple_cmp, ao2_cleanup);
425         if (AST_VECTOR_SIZE(&sub->applications) == 0) {
426                 if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
427                         ao2_unlink(endpoint_subscriptions, sub);
428                 } else {
429                         ast_rwlock_wrlock(&tech_subscriptions_lock);
430                         AST_VECTOR_REMOVE_CMP_UNORDERED(&tech_subscriptions, endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD,
431                                 messaging_subscription_cmp, AST_VECTOR_ELEM_CLEANUP_NOOP);
432                         ast_rwlock_unlock(&tech_subscriptions_lock);
433                 }
434         }
435         ao2_unlock(sub);
436         ao2_ref(sub, -1);
437
438         ast_debug(3, "App '%s' unsubscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
439         ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Unsubscribed\r\nAppName: %s\r\nToken: %s\r\n",
440                 app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
441 }
442
443 static struct message_subscription *get_or_create_subscription(struct ast_endpoint *endpoint)
444 {
445         struct message_subscription *sub = get_subscription(endpoint);
446
447         if (sub) {
448                 return sub;
449         }
450
451         sub = message_subscription_alloc(endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD);
452         if (!sub) {
453                 return NULL;
454         }
455
456         if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
457                 ao2_link(endpoint_subscriptions, sub);
458         } else {
459                 ast_rwlock_wrlock(&tech_subscriptions_lock);
460                 AST_VECTOR_APPEND(&tech_subscriptions, ao2_bump(sub));
461                 ast_rwlock_unlock(&tech_subscriptions_lock);
462         }
463
464         return sub;
465 }
466
467 int messaging_app_subscribe_endpoint(const char *app_name, struct ast_endpoint *endpoint, message_received_cb callback, void *pvt)
468 {
469         RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
470         struct application_tuple *tuple;
471
472         sub = get_or_create_subscription(endpoint);
473         if (!sub) {
474                 return -1;
475         }
476
477         ao2_lock(sub);
478         if (is_app_subscribed(sub, app_name)) {
479                 ao2_unlock(sub);
480                 return 0;
481         }
482
483         tuple = application_tuple_alloc(app_name, callback, pvt);
484         if (!tuple) {
485                 ao2_unlock(sub);
486                 return -1;
487         }
488         AST_VECTOR_APPEND(&sub->applications, tuple);
489         ao2_unlock(sub);
490
491         ast_debug(3, "App '%s' subscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
492         ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Subscribed\r\nAppName: %s\r\nToken: %s\r\n",
493                 app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
494
495         return 0;
496 }
497
498
499 int messaging_cleanup(void)
500 {
501         ast_msg_handler_unregister(&ari_msg_handler);
502         ao2_ref(endpoint_subscriptions, -1);
503         AST_VECTOR_FREE(&tech_subscriptions);
504         ast_rwlock_destroy(&tech_subscriptions_lock);\
505
506         return 0;
507 }
508
509 int messaging_init(void)
510 {
511         endpoint_subscriptions = ao2_t_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
512                 ENDPOINTS_NUM_BUCKETS, message_subscription_hash_cb, NULL,
513                 message_subscription_compare_cb, "Endpoint messaging subscription container creation");
514         if (!endpoint_subscriptions) {
515                 return -1;
516         }
517
518         if (AST_VECTOR_INIT(&tech_subscriptions, 4)) {
519                 ao2_ref(endpoint_subscriptions, -1);
520                 return -1;
521         }
522
523         if (ast_rwlock_init(&tech_subscriptions_lock)) {
524                 ao2_ref(endpoint_subscriptions, -1);
525                 AST_VECTOR_FREE(&tech_subscriptions);
526                 return -1;
527         }
528
529         if (ast_msg_handler_register(&ari_msg_handler)) {
530                 ao2_ref(endpoint_subscriptions, -1);
531                 AST_VECTOR_FREE(&tech_subscriptions);
532                 ast_rwlock_destroy(&tech_subscriptions_lock);
533                 return -1;
534         }
535
536         return 0;
537 }