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