Merge "chan_sip+native_bridge_rtp: no directmedia for ptime other than default ptime."
[asterisk/asterisk.git] / res / ari / resource_events.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 /api-docs/events.{format} implementation- WebSocket resource
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 #include "resource_events.h"
33 #include "asterisk/astobj2.h"
34 #include "asterisk/http_websocket.h"
35 #include "asterisk/stasis_app.h"
36 #include "asterisk/vector.h"
37
38 /*! Number of buckets for the event session registry. Remember to keep it a prime number! */
39 #define EVENT_SESSION_NUM_BUCKETS 23
40
41 /*! Number of buckets for a websocket apps container. Remember to keep it a prime number! */
42 #define APPS_NUM_BUCKETS 7
43
44 /*! Initial size of a message queue. */
45 #define MESSAGES_INIT_SIZE 23
46
47
48 /*! \brief A wrapper for the /ref ast_ari_websocket_session. */
49 struct event_session {
50         struct ast_ari_websocket_session *ws_session;   /*!< Handle to the websocket session. */
51         struct ao2_container *websocket_apps;           /*!< List of Stasis apps registered to
52                                                              the websocket session. */
53         AST_VECTOR(, struct ast_json *) message_queue;  /*!< Container for holding delayed messages. */
54         char session_id[];                              /*!< The id for the websocket session. */
55 };
56
57 /*! \brief \ref event_session error types. */
58 enum event_session_error_type {
59         ERROR_TYPE_STASIS_REGISTRATION = 1,  /*!< Stasis failed to register the application. */
60         ERROR_TYPE_OOM = 2,                  /*!< Insufficient memory to create the event
61                                                   session. */
62         ERROR_TYPE_MISSING_APP_PARAM = 3,    /*!< HTTP request was missing an [app] parameter. */
63         ERROR_TYPE_INVALID_APP_PARAM = 4,    /*!< HTTP request contained an invalid [app]
64                                                   parameter. */
65 };
66
67 /*! \brief Local registry for created \ref event_session objects. */
68 static struct ao2_container *event_session_registry;
69
70 /*!
71  * \brief Callback handler for Stasis application messages.
72  *
73  * \internal
74  *
75  * \param data      Void pointer to the event session (\ref event_session).
76  * \param app_name  Name of the Stasis application that dispatched the message.
77  * \param message   The dispatched message.
78  */
79 static void stasis_app_message_handler(
80                 void *data, const char *app_name, struct ast_json *message)
81 {
82         struct event_session *session = data;
83         const char *msg_type, *msg_application;
84         int app_debug_enabled;
85
86         ast_assert(session != NULL);
87
88         /*
89          * We need to get the debug flag before lockinf session
90          * to help prevent a deadlock with the apps_registry container.
91          */
92         app_debug_enabled = stasis_app_get_debug_by_name(app_name);
93
94         ao2_lock(session);
95
96         msg_type = S_OR(ast_json_string_get(ast_json_object_get(message, "type")), "");
97         msg_application = S_OR(
98                 ast_json_string_get(ast_json_object_get(message, "application")), "");
99
100         /* If we've been replaced, remove the application from our local
101            websocket_apps container */
102         if (strcmp(msg_type, "ApplicationReplaced") == 0 &&
103                 strcmp(msg_application, app_name) == 0) {
104                 ao2_find(session->websocket_apps, msg_application,
105                         OBJ_UNLINK | OBJ_NODATA);
106         }
107
108         /* Now, we need to determine our state to see how we will handle the message */
109         if (ast_json_object_set(message, "application", ast_json_string_create(app_name))) {
110                 /* We failed to add an application element to our json message */
111                 ast_log(LOG_WARNING,
112                         "Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
113                         msg_type,
114                         msg_application);
115         } else if (!session->ws_session) {
116                 /* If the websocket is NULL, the message goes to the queue */
117                 if (!AST_VECTOR_APPEND(&session->message_queue, message)) {
118                         ast_json_ref(message);
119                 }
120                 ast_log(LOG_WARNING,
121                                 "Queued '%s' message for Stasis app '%s'; websocket is not ready\n",
122                                 msg_type,
123                                 msg_application);
124         } else if (stasis_app_event_allowed(app_name, message)) {
125                 if (app_debug_enabled) {
126                         char *str = ast_json_dump_string_format(message, ast_ari_json_format());
127
128                         ast_verbose("<--- Sending ARI event to %s --->\n%s\n",
129                                 ast_sockaddr_stringify(ast_ari_websocket_session_get_remote_addr(session->ws_session)),
130                                 str);
131                         ast_json_free(str);
132                 }
133
134                 /* We are ready to publish the message */
135                 ast_ari_websocket_session_write(session->ws_session, message);
136         }
137
138         ao2_unlock(session);
139 }
140
141 /*!
142  * \brief AO2 comparison function for \ref event_session objects.
143  *
144  * \internal
145  *
146  * \param obj    Void pointer to the \ref event_session container.
147  * \param arg    Void pointer to the \ref event_session object.
148  * \param flags  The \ref search_flags to use when creating the hash key.
149  *
150  * \retval 0          The objects are not equal.
151  * \retval CMP_MATCH  The objects are equal.
152  */
153 static int event_session_compare(void *obj, void *arg, int flags)
154 {
155         const struct event_session *object_left = obj;
156         const struct event_session *object_right = arg;
157         const char *right_key = arg;
158         int cmp = 0;
159
160         switch (flags & OBJ_SEARCH_MASK) {
161         case OBJ_SEARCH_OBJECT:
162                 right_key = object_right->session_id;
163                 /* Fall through */
164         case OBJ_SEARCH_KEY:
165                 cmp = strcmp(object_left->session_id, right_key);
166                 break;
167         case OBJ_SEARCH_PARTIAL_KEY:
168                 cmp = strncmp(object_left->session_id, right_key, strlen(right_key));
169                 break;
170         default:
171                 break;
172         }
173
174         return cmp ? 0 : CMP_MATCH;
175 }
176
177 /*!
178  * \brief AO2 hash function for \ref event_session objects.
179  *
180  * \details Computes hash value for the given \ref event_session, with respect to the
181  *          provided search flags.
182  *
183  * \internal
184  *
185  * \param obj    Void pointer to the \ref event_session object.
186  * \param flags  The \ref search_flags to use when creating the hash key.
187  *
188  * \retval > 0  on success
189  * \retval   0  on failure
190  */
191 static int event_session_hash(const void *obj, const int flags)
192 {
193         const struct event_session *session;
194         const char *key;
195
196         switch (flags & OBJ_SEARCH_MASK) {
197         case OBJ_SEARCH_KEY:
198                 key = obj;
199                 break;
200         case OBJ_SEARCH_OBJECT:
201                 session = obj;
202                 key = session->session_id;
203                 break;
204         default:
205                 /* Hash can only work on something with a full key. */
206                 ast_assert(0);
207                 return 0;
208         }
209         return ast_str_hash(key);
210 }
211
212 /*!
213  * \brief Explicitly shutdown a session.
214  *
215  * \details An explicit shutdown is necessary, since the \ref stasis_app has a reference
216  *          to this session. We also need to be sure to null out the \c ws_session field,
217  *          since the websocket is about to go away.
218  *
219  * \internal
220  *
221  * \param session  Event session object (\ref event_session).
222  */
223 static void event_session_shutdown(struct event_session *session)
224 {
225         struct ao2_iterator i;
226         char *app;
227         int j;
228         SCOPED_AO2LOCK(lock, session);
229
230         /* Clean up the websocket_apps container */
231         if (session->websocket_apps) {
232                 i = ao2_iterator_init(session->websocket_apps, 0);
233                 while ((app = ao2_iterator_next(&i))) {
234                         stasis_app_unregister(app);
235                         ao2_cleanup(app);
236                 }
237                 ao2_iterator_destroy(&i);
238                 ao2_cleanup(session->websocket_apps);
239                 session->websocket_apps = NULL;
240         }
241
242         /* Clean up the message_queue container */
243         for (j = 0; j < AST_VECTOR_SIZE(&session->message_queue); j++) {
244                 struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, j);
245                 ast_json_unref(msg);
246         }
247         AST_VECTOR_FREE(&session->message_queue);
248
249         /* Remove the handle to the underlying websocket session */
250         session->ws_session = NULL;
251 }
252
253 /*!
254  * \brief Updates the websocket session for an \ref event_session.
255  *
256  * \details The websocket for the given \ref event_session will be updated to the value
257  *          of the \c ws_session argument.
258  *
259  *          If the value of the \c ws_session is not \c NULL and there are messages in the
260  *          event session's \c message_queue, the messages are dispatched and removed from
261  *          the queue.
262  *
263  * \internal
264  *
265  * \param session     The event session object to update (\ref event_session).
266  * \param ws_session  Handle to the underlying websocket session
267  *                    (\ref ast_ari_websocket_session).
268  */
269 static void event_session_update_websocket(
270                 struct event_session *session, struct ast_ari_websocket_session *ws_session)
271 {
272         int i;
273
274         ast_assert(session != NULL);
275
276         ao2_lock(session);
277
278         session->ws_session = ws_session;
279
280         for (i = 0; i < AST_VECTOR_SIZE(&session->message_queue); i++) {
281                 struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, i);
282                 ast_ari_websocket_session_write(session->ws_session, msg);
283                 ast_json_unref(msg);
284         }
285
286         AST_VECTOR_RESET(&session->message_queue, AST_VECTOR_ELEM_CLEANUP_NOOP);
287         ao2_unlock(session);
288 }
289
290 /*!
291  * \brief Processes cleanup actions for a \ref event_session object.
292  *
293  * \internal
294  *
295  * \param session  The event session object to cleanup (\ref event_session).
296  */
297 static void event_session_cleanup(struct event_session *session)
298 {
299         if (!session) {
300                 return;
301         }
302
303         event_session_shutdown(session);
304         if (event_session_registry) {
305                 ao2_unlink(event_session_registry, session);
306         }
307 }
308
309 /*!
310  * \brief Event session object destructor (\ref event_session).
311  *
312  * \internal
313  *
314  * \param obj  Void pointer to the \ref event_session object.
315  */
316 static void event_session_dtor(void *obj)
317 {
318 #ifdef AST_DEVMODE /* Avoid unused variable warning */
319         struct event_session *session = obj;
320 #endif
321
322         /* event_session_shutdown should have been called before now */
323         ast_assert(session->ws_session == NULL);
324         ast_assert(session->websocket_apps == NULL);
325         ast_assert(AST_VECTOR_SIZE(&session->message_queue) == 0);
326 }
327
328 /*!
329  * \brief Handles \ref event_session error processing.
330  *
331  * \internal
332  *
333  * \param session  The \ref event_session object.
334  * \param error    The \ref event_session_error_type to handle.
335  * \param ser      HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
336  *
337  * \retval  -1  Always returns -1.
338  */
339 static int event_session_allocation_error_handler(
340                 struct event_session *session, enum event_session_error_type error,
341                 struct ast_tcptls_session_instance *ser)
342 {
343         /* Notify the client */
344         switch (error) {
345         case ERROR_TYPE_STASIS_REGISTRATION:
346                 ast_http_error(ser, 500, "Internal Server Error",
347                         "Stasis registration failed");
348                 break;
349
350         case ERROR_TYPE_OOM:
351                 ast_http_error(ser, 500, "Internal Server Error",
352                         "Allocation failed");
353                 break;
354
355         case ERROR_TYPE_MISSING_APP_PARAM:
356                 ast_http_error(ser, 400, "Bad Request",
357                         "HTTP request is missing param: [app]");
358                 break;
359
360         case ERROR_TYPE_INVALID_APP_PARAM:
361                 ast_http_error(ser, 400, "Bad Request",
362                         "Invalid application provided in param [app].");
363                 break;
364
365         default:
366                 break;
367         }
368
369         /* Cleanup the session */
370         event_session_cleanup(session);
371         return -1;
372 }
373
374 /*!
375  * \brief Creates an \ref event_session object and registers its apps with Stasis.
376  *
377  * \internal
378  *
379  * \param ser         HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
380  * \param args        The Stasis [app] parameters as parsed from the HTTP request
381  *                    (\ref ast_ari_events_event_websocket_args).
382  * \param session_id  The id for the websocket session that will be created for this
383  *                    event session.
384  *
385  * \retval  0  on success
386  * \retval -1  on failure
387  */
388 static int event_session_alloc(struct ast_tcptls_session_instance *ser,
389                 struct ast_ari_events_event_websocket_args *args, const char *session_id)
390 {
391         RAII_VAR(struct event_session *, session, NULL, ao2_cleanup);
392         int (* register_handler)(const char *, stasis_app_cb handler, void *data);
393         size_t size, i;
394
395         /* The request must have at least one [app] parameter */
396         if (args->app_count == 0) {
397                 return event_session_allocation_error_handler(
398                         session, ERROR_TYPE_MISSING_APP_PARAM, ser);
399         }
400
401         size = sizeof(*session) + strlen(session_id) + 1;
402
403         /* Instantiate the event session */
404         session = ao2_alloc(size, event_session_dtor);
405         if (!session) {
406                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
407         }
408
409         strncpy(session->session_id, session_id, size - sizeof(*session));
410
411         /* Instantiate the hash table for Stasis apps */
412         session->websocket_apps =
413                 ast_str_container_alloc(APPS_NUM_BUCKETS);
414
415         if (!session->websocket_apps) {
416                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
417         }
418
419         /* Instantiate the message queue */
420         if (AST_VECTOR_INIT(&session->message_queue, MESSAGES_INIT_SIZE)) {
421                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
422         }
423
424         /* Register the apps with Stasis */
425         if (args->subscribe_all) {
426                 register_handler = &stasis_app_register_all;
427         } else {
428                 register_handler = &stasis_app_register;
429         }
430
431         for (i = 0; i < args->app_count; ++i) {
432                 const char *app = args->app[i];
433
434                 if (ast_strlen_zero(app)) {
435                         return event_session_allocation_error_handler(
436                                 session, ERROR_TYPE_INVALID_APP_PARAM, ser);
437                 }
438
439                 if (ast_str_container_add(session->websocket_apps, app)) {
440                         return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
441                 }
442
443                 if (register_handler(app, stasis_app_message_handler, session)) {
444                         ast_log(LOG_WARNING, "Stasis registration failed for application: '%s'\n", app);
445                         return event_session_allocation_error_handler(
446                                 session, ERROR_TYPE_STASIS_REGISTRATION, ser);
447                 }
448         }
449
450         /* Add the event session to the local registry */
451         if (!ao2_link(event_session_registry, session)) {
452                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
453         }
454
455         return 0;
456 }
457
458 static int event_session_shutdown_cb(void *session, void *arg, int flags)
459 {
460         event_session_cleanup(session);
461
462         return 0;
463 }
464
465 void ast_ari_websocket_events_event_websocket_dtor(void)
466 {
467         ao2_callback(event_session_registry, OBJ_MULTIPLE | OBJ_NODATA, event_session_shutdown_cb, NULL);
468
469         ao2_cleanup(event_session_registry);
470         event_session_registry = NULL;
471 }
472
473 int ast_ari_websocket_events_event_websocket_init(void)
474 {
475         /* Try to instantiate the registry */
476         event_session_registry = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
477                 EVENT_SESSION_NUM_BUCKETS, event_session_hash, NULL, event_session_compare);
478         if (!event_session_registry) {
479                 /* This is bad, bad. */
480                 ast_log(LOG_WARNING,
481                             "Failed to allocate the local registry for websocket applications\n");
482                 return -1;
483         }
484
485         return 0;
486 }
487
488 int ast_ari_websocket_events_event_websocket_attempted(
489                 struct ast_tcptls_session_instance *ser, struct ast_variable *headers,
490                 struct ast_ari_events_event_websocket_args *args, const char *session_id)
491 {
492         ast_debug(3, "/events WebSocket attempted\n");
493
494         /* Create the event session */
495         return event_session_alloc(ser, args, session_id);
496 }
497
498 void ast_ari_websocket_events_event_websocket_established(
499                 struct ast_ari_websocket_session *ws_session, struct ast_variable *headers,
500                 struct ast_ari_events_event_websocket_args *args)
501 {
502         struct event_session *session;
503
504         struct ast_json *msg;
505         const char *session_id;
506
507         ast_debug(3, "/events WebSocket established\n");
508
509         ast_assert(ws_session != NULL);
510
511         session_id = ast_ari_websocket_session_id(ws_session);
512
513         /* Find the event_session and update its websocket  */
514         session = ao2_find(event_session_registry, session_id, OBJ_SEARCH_KEY);
515         if (session) {
516                 ao2_unlink(event_session_registry, session);
517                 event_session_update_websocket(session, ws_session);
518         } else {
519                 ast_log(LOG_WARNING,
520                         "Failed to locate an event session for the provided websocket session\n");
521         }
522
523         /* We don't process any input, but we'll consume it waiting for EOF */
524         while ((msg = ast_ari_websocket_session_read(ws_session))) {
525                 ast_json_unref(msg);
526         }
527
528         event_session_cleanup(session);
529         ao2_ref(session, -1);
530 }
531
532 void ast_ari_events_user_event(struct ast_variable *headers,
533         struct ast_ari_events_user_event_args *args,
534         struct ast_ari_response *response)
535 {
536         enum stasis_app_user_event_res res;
537         struct ast_json *json_variables = NULL;
538
539         if (args->variables) {
540                 ast_ari_events_user_event_parse_body(args->variables, args);
541                 json_variables = ast_json_object_get(args->variables, "variables");
542         }
543
544         if (ast_strlen_zero(args->application)) {
545                 ast_ari_response_error(response, 400, "Bad Request",
546                         "Missing parameter application");
547                 return;
548         }
549
550         res = stasis_app_user_event(args->application,
551                 args->event_name,
552                 args->source, args->source_count,
553                 json_variables);
554
555         switch (res) {
556         case STASIS_APP_USER_OK:
557                 ast_ari_response_no_content(response);
558                 break;
559
560         case STASIS_APP_USER_APP_NOT_FOUND:
561                 ast_ari_response_error(response, 404, "Not Found",
562                         "Application not found");
563                 break;
564
565         case STASIS_APP_USER_EVENT_SOURCE_NOT_FOUND:
566                 ast_ari_response_error(response, 422, "Unprocessable Entity",
567                         "Event source was not found");
568                 break;
569
570         case STASIS_APP_USER_EVENT_SOURCE_BAD_SCHEME:
571                 ast_ari_response_error(response, 400, "Bad Request",
572                         "Invalid event source URI scheme");
573                 break;
574
575         case STASIS_APP_USER_USEREVENT_INVALID:
576                 ast_ari_response_error(response, 400, "Bad Request",
577                         "Invalid userevnet data");
578                 break;
579
580         case STASIS_APP_USER_INTERNAL_ERROR:
581         default:
582                 ast_ari_response_error(response, 500, "Internal Server Error",
583                         "Error processing request");
584         }
585 }