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