ARI: Channels added to Stasis application during WebSocket creation ...
[asterisk/asterisk.git] / res / ari / resource_events.c
index e666f2e..f1342b7 100644 (file)
 
 ASTERISK_REGISTER_FILE()
 
+#include "resource_events.h"
 #include "asterisk/astobj2.h"
 #include "asterisk/stasis_app.h"
-#include "resource_events.h"
+#include "asterisk/vector.h"
 
-/*! Number of buckets for the Stasis application hash table. Remember to keep it
- *  a prime number!
- */
+/*! Number of buckets for the event session registry. Remember to keep it a prime number! */
+#define EVENT_SESSION_NUM_BUCKETS 23
+
+/*! Number of buckets for a websocket apps container. Remember to keep it a prime number! */
 #define APPS_NUM_BUCKETS 7
 
-/*! \brief A connection to the event WebSocket */
+/*! Initial size of a message queue. */
+#define MESSAGES_INIT_SIZE 23
+
+
+/*! \brief A wrapper for the /ref ast_ari_websocket_session. */
 struct event_session {
-       struct ast_ari_websocket_session *ws_session;
-       struct ao2_container *websocket_apps;
+       struct ast_ari_websocket_session *ws_session;   /*!< Handle to the websocket session. */
+       struct ao2_container *websocket_apps;           /*!< List of Stasis apps registered to
+                                                            the websocket session. */
+       AST_VECTOR(, struct ast_json *) message_queue;  /*!< Container for holding delayed messages. */
+       char session_id[];                              /*!< The id for the websocket session. */
 };
 
+/*! \brief \ref event_session error types. */
+enum event_session_error_type {
+       ERROR_TYPE_STASIS_REGISTRATION = 1,  /*!< Stasis failed to register the application. */
+       ERROR_TYPE_OOM = 2,                  /*!< Insufficient memory to create the event
+                                                 session. */
+       ERROR_TYPE_MISSING_APP_PARAM = 3,    /*!< HTTP request was missing an [app] parameter. */
+       ERROR_TYPE_INVALID_APP_PARAM = 4,    /*!< HTTP request contained an invalid [app]
+                                                 parameter. */
+};
+
+/*! \brief Local registry for created \ref event_session objects. */
+static struct ao2_container *event_session_registry;
+
 /*!
- * \brief Explicitly shutdown a session.
+ * \brief Callback handler for Stasis application messages.
  *
- * An explicit shutdown is necessary, since stasis-app has a reference to this
- * session. We also need to be sure to null out the \c ws_session field, since
- * the websocket is about to go away.
+ * \internal
  *
- * \param session Session info struct.
+ * \param data      Void pointer to the event session (\ref event_session).
+ * \param app_name  Name of the Stasis application that dispatched the message.
+ * \param message   The dispatched message.
  */
-static void session_shutdown(struct event_session *session)
+static void stasis_app_message_handler(
+               void *data, const char *app_name, struct ast_json *message)
 {
-        struct ao2_iterator i;
-       char *app;
-       SCOPED_AO2LOCK(lock, session);
+       struct event_session *session = data;
+       const char *msg_type, *msg_application;
+
+       ast_assert(session != NULL);
+
+       ao2_lock(session);
+
+       msg_type = S_OR(ast_json_string_get(ast_json_object_get(message, "type")), "");
+       msg_application = S_OR(
+               ast_json_string_get(ast_json_object_get(message, "application")), "");
 
-       i = ao2_iterator_init(session->websocket_apps, 0);
-       while ((app = ao2_iterator_next(&i))) {
-               stasis_app_unregister(app);
-               ao2_cleanup(app);
+       /* If we've been replaced, remove the application from our local
+          websocket_apps container */
+       if (strcmp(msg_type, "ApplicationReplaced") == 0 &&
+               strcmp(msg_application, app_name) == 0) {
+               ao2_find(session->websocket_apps, msg_application,
+                       OBJ_UNLINK | OBJ_NODATA);
        }
-       ao2_iterator_destroy(&i);
-       ao2_cleanup(session->websocket_apps);
 
-       session->websocket_apps = NULL;
-       session->ws_session = NULL;
+       /* Now, we need to determine our state to see how we will handle the message */
+       if (ast_json_object_set(message, "application", ast_json_string_create(app_name))) {
+               /* We failed to add an application element to our json message */
+               ast_log(LOG_WARNING,
+                       "Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
+                       msg_type,
+                       msg_application);
+       } else if (!session->ws_session) {
+                       /* If the websocket is NULL, the message goes to the queue */
+                       AST_VECTOR_APPEND(&session->message_queue, message);
+                       ast_log(LOG_WARNING,
+                               "Queued '%s' message for Stasis app '%s'; websocket is not ready\n",
+                               msg_type,
+                               msg_application);
+       } else {
+               /* We are ready to publish the message */
+               ast_ari_websocket_session_write(session->ws_session, message);
+       }
+
+       ao2_unlock(session);
 }
 
-static void session_dtor(void *obj)
+/*!
+ * \brief AO2 comparison function for \ref event_session objects.
+ *
+ * \internal
+ *
+ * \param obj    Void pointer to the \ref event_session container.
+ * \param arg    Void pointer to the \ref event_session object.
+ * \param flags  The \ref search_flags to use when creating the hash key.
+ *
+ * \retval 0          The objects are not equal.
+ * \retval CMP_MATCH  The objects are equal.
+ */
+static int event_session_compare(void *obj, void *arg, int flags)
 {
-#ifdef AST_DEVMODE /* Avoid unused variable warning */
-       struct event_session *session = obj;
-#endif
+       const struct event_session *object_left = obj;
+       const struct event_session *object_right = arg;
+       const char *right_key = arg;
+       int cmp = 0;
+
+       switch (flags & OBJ_SEARCH_MASK) {
+       case OBJ_SEARCH_OBJECT:
+               right_key = object_right->session_id;
+               /* Fall through */
+       case OBJ_SEARCH_KEY:
+               cmp = strcmp(object_left->session_id, right_key);
+               break;
+       case OBJ_SEARCH_PARTIAL_KEY:
+               cmp = strncmp(object_left->session_id, right_key, strlen(right_key));
+               break;
+       default:
+               break;
+       }
 
-       /* session_shutdown should have been called before */
-       ast_assert(session->ws_session == NULL);
-       ast_assert(session->websocket_apps == NULL);
+       return cmp ? 0 : CMP_MATCH;
+}
+
+/*!
+ * \brief AO2 hash function for \ref event_session objects.
+ *
+ * \details Computes hash value for the given \ref event_session, with respect to the
+ *          provided search flags.
+ *
+ * \internal
+ *
+ * \param obj    Void pointer to the \ref event_session object.
+ * \param flags  The \ref search_flags to use when creating the hash key.
+ *
+ * \retval > 0  on success
+ * \retval   0  on failure
+ */
+static int event_session_hash(const void *obj, const int flags)
+{
+       const struct event_session *session;
+       const char *key;
+
+       switch (flags & OBJ_SEARCH_MASK) {
+       case OBJ_SEARCH_KEY:
+               key = obj;
+               break;
+       case OBJ_SEARCH_OBJECT:
+               session = obj;
+               key = session->session_id;
+               break;
+       default:
+               /* Hash can only work on something with a full key. */
+               ast_assert(0);
+               return 0;
+       }
+       return ast_str_hash(key);
 }
 
-static void session_cleanup(struct event_session *session)
+/*!
+ * \brief Explicitly shutdown a session.
+ *
+ * \details An explicit shutdown is necessary, since the \ref stasis_app has a reference
+ *          to this session. We also need to be sure to null out the \c ws_session field,
+ *          since the websocket is about to go away.
+ *
+ * \internal
+ *
+ * \param session  Event session object (\ref event_session).
+ */
+static void event_session_shutdown(struct event_session *session)
 {
-       session_shutdown(session);
-       ao2_cleanup(session);
+       struct ao2_iterator i;
+       char *app;
+       int j;
+       SCOPED_AO2LOCK(lock, session);
+
+       /* Clean up the websocket_apps container */
+       if (session->websocket_apps) {
+               i = ao2_iterator_init(session->websocket_apps, 0);
+               while ((app = ao2_iterator_next(&i))) {
+                       stasis_app_unregister(app);
+                       ao2_cleanup(app);
+               }
+               ao2_iterator_destroy(&i);
+               ao2_cleanup(session->websocket_apps);
+               session->websocket_apps = NULL;
+       }
+
+       /* Clean up the message_queue container */
+       for (j = 0; j < AST_VECTOR_SIZE(&session->message_queue); j++) {
+               struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, j);
+               ast_json_unref(msg);
+       }
+       AST_VECTOR_FREE(&session->message_queue);
+
+       /* Remove the handle to the underlying websocket session */
+       session->ws_session = NULL;
 }
 
-static struct event_session *session_create(
-       struct ast_ari_websocket_session *ws_session)
+/*!
+ * \brief Updates the websocket session for an \ref event_session.
+ *
+ * \details The websocket for the given \ref event_session will be updated to the value
+ *          of the \c ws_session argument.
+ *
+ *          If the value of the \c ws_session is not \c NULL and there are messages in the
+ *          event session's \c message_queue, the messages are dispatched and removed from
+ *          the queue.
+ *
+ * \internal
+ *
+ * \param session     The event session object to update (\ref event_session).
+ * \param ws_session  Handle to the underlying websocket session
+ *                    (\ref ast_ari_websocket_session).
+ */
+static void event_session_update_websocket(
+               struct event_session *session, struct ast_ari_websocket_session *ws_session)
 {
-       RAII_VAR(struct event_session *, session, NULL, ao2_cleanup);
+       int i;
+
+       ast_assert(session != NULL);
 
-       session = ao2_alloc(sizeof(*session), session_dtor);
+       ao2_lock(session);
 
        session->ws_session = ws_session;
-       session->websocket_apps =
-               ast_str_container_alloc(APPS_NUM_BUCKETS);
 
-       if (!session->websocket_apps) {
-               return NULL;
+       for (i = 0; i < AST_VECTOR_SIZE(&session->message_queue); i++) {
+               struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, i);
+               ast_ari_websocket_session_write(session->ws_session, msg);
+               ast_json_unref(msg);
        }
 
-       ao2_ref(session, +1);
-       return session;
+       AST_VECTOR_RESET(&session->message_queue, AST_VECTOR_ELEM_CLEANUP_NOOP);
+       ao2_unlock(session);
 }
 
 /*!
- * \brief Callback handler for Stasis application messages.
+ * \brief Processes cleanup actions for a \ref event_session object.
+ *
+ * \internal
+ *
+ * \param session  The event session object to cleanup (\ref event_session).
  */
-static void app_handler(void *data, const char *app_name,
-                       struct ast_json *message)
+static void event_session_cleanup(struct event_session *session)
 {
-       struct event_session *session = data;
-       int res;
-       const char *msg_type = S_OR(
-               ast_json_string_get(ast_json_object_get(message, "type")),
-               "");
-       const char *msg_application = S_OR(
-               ast_json_string_get(ast_json_object_get(message, "application")),
-               "");
-
        if (!session) {
                return;
        }
-       /* Determine if we've been replaced */
-       if (strcmp(msg_type, "ApplicationReplaced") == 0 &&
-               strcmp(msg_application, app_name) == 0) {
-               ao2_find(session->websocket_apps, msg_application,
-                       OBJ_UNLINK | OBJ_NODATA);
-       }
 
-       res = ast_json_object_set(message, "application",
-                                 ast_json_string_create(app_name));
-       if(res != 0) {
-               return;
-       }
+       event_session_shutdown(session);
+       ao2_unlink(event_session_registry, session);
+}
 
-       ao2_lock(session);
-       if (session->ws_session) {
-               ast_ari_websocket_session_write(session->ws_session, message);
-       }
-       ao2_unlock(session);
+/*!
+ * \brief Event session object destructor (\ref event_session).
+ *
+ * \internal
+ *
+ * \param obj  Void pointer to the \ref event_session object.
+ */
+static void event_session_dtor(void *obj)
+{
+#ifdef AST_DEVMODE /* Avoid unused variable warning */
+       struct event_session *session = obj;
+
+#endif
+
+       /* event_session_shutdown should have been called before now */
+       ast_assert(session->ws_session == NULL);
+       ast_assert(session->websocket_apps == NULL);
+       ast_assert(AST_VECTOR_SIZE(&session->message_queue) == 0);
 }
 
 /*!
- * \brief Register for all of the apps given.
- * \param session Session info struct.
- * \param app_name Name of application to register.
+ * \brief Handles \ref event_session error processing.
+ *
+ * \internal
+ *
+ * \param session  The \ref event_session object.
+ * \param error    The \ref event_session_error_type to handle.
+ * \param ser      HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
+ *
+ * \retval  -1  Always returns -1.
  */
-static int session_register_app(struct event_session *session,
-                                const char *app_name)
+static int event_session_allocation_error_handler(
+               struct event_session *session, enum event_session_error_type error,
+               struct ast_tcptls_session_instance *ser)
 {
-       SCOPED_AO2LOCK(lock, session);
+       /* Notify the client */
+       switch (error) {
+       case ERROR_TYPE_STASIS_REGISTRATION:
+               ast_http_error(ser, 500, "Internal Server Error",
+                       "Stasis registration failed");
+               break;
 
-       ast_assert(session->ws_session != NULL);
-       ast_assert(session->websocket_apps != NULL);
+       case ERROR_TYPE_OOM:
+               ast_http_error(ser, 500, "Internal Server Error",
+                       "Allocation failed");
+               break;
 
-       if (ast_strlen_zero(app_name)) {
-               return -1;
-       }
+       case ERROR_TYPE_MISSING_APP_PARAM:
+               ast_http_error(ser, 400, "Bad Request",
+                       "HTTP request is missing param: [app]");
+               break;
 
-       if (ast_str_container_add(session->websocket_apps, app_name)) {
-               ast_ari_websocket_session_write(session->ws_session,
-                       ast_ari_oom_json());
-               return -1;
-       }
+       case ERROR_TYPE_INVALID_APP_PARAM:
+               ast_http_error(ser, 400, "Bad Request",
+                       "Invalid application provided in param [app].");
+               break;
 
-       stasis_app_register(app_name, app_handler, session);
+       default:
+               break;
+       }
 
-       return 0;
+       /* Cleanup the session */
+       event_session_cleanup(session);
+       return -1;
 }
 
-int ast_ari_websocket_events_event_websocket_attempted(struct ast_tcptls_session_instance *ser,
-       struct ast_variable *headers,
-       struct ast_ari_events_event_websocket_args *args)
+/*!
+ * \brief Creates an \ref event_session object and registers its apps with Stasis.
+ *
+ * \internal
+ *
+ * \param ser         HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
+ * \param args        The Stasis [app] parameters as parsed from the HTTP request
+ *                    (\ref ast_ari_events_event_websocket_args).
+ * \param session_id  The id for the websocket session that will be created for this
+ *                    event session.
+ *
+ * \retval  0  on success
+ * \retval -1  on failure
+ */
+static int event_session_alloc(struct ast_tcptls_session_instance *ser,
+               struct ast_ari_events_event_websocket_args *args, const char *session_id)
 {
-       int res = 0;
-       size_t i, j;
-
-       ast_debug(3, "/events WebSocket attempted\n");
+       RAII_VAR(struct event_session *, session, NULL, ao2_cleanup);
+       size_t size, i;
 
+       /* The request must have at least one [app] parameter */
        if (args->app_count == 0) {
-               ast_http_error(ser, 400, "Bad Request", "Missing param 'app'");
-               return -1;
+               return event_session_allocation_error_handler(
+                       session, ERROR_TYPE_MISSING_APP_PARAM, ser);
+       }
+
+       size = sizeof(*session) + strlen(session_id) + 1;
+
+       /* Instantiate the event session */
+       session = ao2_alloc(size, event_session_dtor);
+       if (!session) {
+               return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
        }
 
+       strncpy(session->session_id, session_id, size - sizeof(*session));
+
+       /* Instantiate the hash table for Stasis apps */
+       session->websocket_apps =
+               ast_str_container_alloc(APPS_NUM_BUCKETS);
+
+       if (!session->websocket_apps) {
+               return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
+       }
+
+       /* Instantiate the message queue */
+       if (AST_VECTOR_INIT(&session->message_queue, MESSAGES_INIT_SIZE)) {
+               return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
+       }
+
+       /* Register the apps with Stasis */
        for (i = 0; i < args->app_count; ++i) {
-               if (ast_strlen_zero(args->app[i])) {
-                       res = -1;
-                       break;
+               const char *app = args->app[i];
+
+               if (ast_strlen_zero(app)) {
+                       return event_session_allocation_error_handler(
+                               session, ERROR_TYPE_INVALID_APP_PARAM, ser);
                }
 
-               res |= stasis_app_register(args->app[i], app_handler, NULL);
-       }
+               if (ast_str_container_add(session->websocket_apps, app)) {
+                       return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
+               }
 
-       if (res) {
-               for (j = 0; j < i; ++j) {
-                       stasis_app_unregister(args->app[j]);
+               if (stasis_app_register(app, stasis_app_message_handler, session)) {
+                       ast_log(LOG_WARNING, "Stasis registration failed for application: '%s'\n", app);
+                       return event_session_allocation_error_handler(
+                               session, ERROR_TYPE_STASIS_REGISTRATION, ser);
                }
-               ast_http_error(ser, 400, "Bad Request", "Invalid application provided in param 'app'.");
        }
 
-       return res;
+       /* Add the event session to the local registry */
+       if (!ao2_link(event_session_registry, session)) {
+               return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
+       }
+
+       return 0;
 }
 
-void ast_ari_websocket_events_event_websocket_established(struct ast_ari_websocket_session *ws_session,
-       struct ast_variable *headers,
-       struct ast_ari_events_event_websocket_args *args)
+int ast_ari_websocket_events_event_websocket_init(void)
 {
-       RAII_VAR(struct event_session *, session, NULL, session_cleanup);
-       struct ast_json *msg;
-       int res;
-       size_t i;
+       /* Try to instantiate the registry */
+       event_session_registry = ao2_container_alloc(EVENT_SESSION_NUM_BUCKETS,
+                                                    event_session_hash,
+                                                    event_session_compare);
+
+       if (!event_session_registry) {
+               /* This is bad, bad. */
+               ast_log(LOG_WARNING,
+                           "Failed to allocate the local registry for websocket applications\n");
+               return -1;
+       }
 
-       ast_debug(3, "/events WebSocket connection\n");
+       return 0;
+}
 
-       session = session_create(ws_session);
-       if (!session) {
-               ast_ari_websocket_session_write(ws_session, ast_ari_oom_json());
-               return;
-       }
+int ast_ari_websocket_events_event_websocket_attempted(
+               struct ast_tcptls_session_instance *ser, struct ast_variable *headers,
+               struct ast_ari_events_event_websocket_args *args, const char *session_id)
+{
+       ast_debug(3, "/events WebSocket attempted\n");
 
-       res = 0;
-       for (i = 0; i < args->app_count; ++i) {
-               if (ast_strlen_zero(args->app[i])) {
-                       continue;
-               }
-               res |= session_register_app(session, args->app[i]);
-       }
+       /* Create the event session */
+       return event_session_alloc(ser, args, session_id);
+}
 
-       if (ao2_container_count(session->websocket_apps) == 0) {
-               RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
+void ast_ari_websocket_events_event_websocket_established(
+               struct ast_ari_websocket_session *ws_session, struct ast_variable *headers,
+               struct ast_ari_events_event_websocket_args *args)
+{
+       RAII_VAR(struct event_session *, session, NULL, event_session_cleanup);
+       struct ast_json *msg;
+       const char *session_id;
 
-               msg = ast_json_pack("{s: s, s: [s]}",
-                       "type", "MissingParams",
-                       "params", "app");
-               if (!msg) {
-                       msg = ast_json_ref(ast_ari_oom_json());
-               }
+       ast_debug(3, "/events WebSocket established\n");
 
-               ast_ari_websocket_session_write(session->ws_session, msg);
-               return;
-       }
+       ast_assert(ws_session != NULL);
 
-       if (res != 0) {
-               ast_ari_websocket_session_write(ws_session, ast_ari_oom_json());
-               return;
+       session_id = ast_ari_websocket_session_id(ws_session);
+
+       /* Find the event_session and update its websocket  */
+       session = ao2_find(event_session_registry, session_id, OBJ_SEARCH_KEY);
+
+       if (session) {
+               ao2_unlink(event_session_registry, session);
+               event_session_update_websocket(session, ws_session);
+       } else {
+               ast_log(LOG_WARNING,
+                       "Failed to locate an event session for the provided websocket session\n");
        }
 
        /* We don't process any input, but we'll consume it waiting for EOF */
@@ -309,4 +538,3 @@ void ast_ari_events_user_event(struct ast_variable *headers,
                        "Error processing request");
        }
 }
-