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