ARI: Channels added to Stasis application during WebSocket creation ...
authorAshley Sanders <asanders@digium.com>
Fri, 31 Jul 2015 16:27:23 +0000 (11:27 -0500)
committerAshley Sanders <asanders@digium.com>
Fri, 31 Jul 2015 16:28:10 +0000 (11:28 -0500)
Prior to ASTERISK-24988, the WebSocket handshake was resolved before Stasis
applications were registered. This was done such that the WebSocket would be
ready when an application is registered. However, by creating the WebSocket
first, the client had the ability to make requests for the Stasis application
it thought had been created with the initial handshake request. The inevitable
conclusion of this scenario was the cart being put before the horse.

ASTERISK-24988 resolved half of the problem by ensuring that the applications
were created and registered with Stasis prior to completing the handshake
with the client. While this meant that Stasis was ready when the client
received the green-light from Asterisk, it also meant that the WebSocket was
not yet ready for Stasis to dispatch messages.

This patch introduces a message queuing mechanism for delaying messages from
Stasis applications while the WebSocket is being constructed. When the ARI
event processor receives the message from the WebSocket that it is being
created, the event processor instantiates an event session which contains a
message queue. It then tries to create and register the requested applications
with Stasis. Messages that are dispatched from Stasis between this point and
the point at which the event processor is notified the WebSocket is ready, are
stashed in the queue. Once the WebSocket has been built, the queue's messages
are dispatched in the order in which they were originally received and the
queue is concurrently cleared.

ASTERISK-25181 #close
Reported By: Matt Jordan

Change-Id: Iafef7b85a2e0bf78c114db4c87ffc3d16d671a17

include/asterisk/ari.h
include/asterisk/astobj2.h
include/asterisk/http_websocket.h
include/asterisk/vector.h
res/ari/ari_websockets.c
res/ari/resource_events.c
res/ari/resource_events.h
res/res_ari_events.c
res/res_http_websocket.c
rest-api-templates/ari_resource.h.mustache
rest-api-templates/res_ari_resource.c.mustache

index c3df46a..c9f47a6 100644 (file)
@@ -188,6 +188,16 @@ int ast_ari_websocket_session_write(struct ast_ari_websocket_session *session,
        struct ast_json *message);
 
 /*!
+ * \brief Get the Session ID for an ARI WebSocket.
+ *
+ * \param session Session to query.
+ * \return Session ID.
+ * \return \c NULL on error.
+ */
+const char *ast_ari_websocket_session_id(
+       const struct ast_ari_websocket_session *session);
+
+/*!
  * \brief The stock message to return when out of memory.
  *
  * The refcount is NOT bumped on this object, so ast_json_ref() if you want to
index 4a7aeee..c28dd23 100644 (file)
@@ -1642,7 +1642,7 @@ void *__ao2_unlink(struct ao2_container *c, void *obj, int flags,
  * The use of flags argument is the follow:
  *
  *      OBJ_UNLINK              unlinks the object found
- *      OBJ_NODATA              on match, do return an object
+ *      OBJ_NODATA              on match, do not return an object
  *                              Callbacks use OBJ_NODATA as a default
  *                              functions such as find() do
  *      OBJ_MULTIPLE            return multiple matches
index 5adc089..23492ff 100644 (file)
@@ -77,13 +77,14 @@ struct ast_websocket;
  * \param ser The TCP/TLS session
  * \param parameters Parameters extracted from the request URI
  * \param headers Headers included in the request
+ * \param session_id The id of the current session.
  *
  * \retval 0 The session should be accepted
  * \retval -1 The session should be rejected. Note that the caller must send an error
  * response using \ref ast_http_error.
  * \since 13.5.0
  */
-typedef int (*ast_websocket_pre_callback)(struct ast_tcptls_session_instance *ser, struct ast_variable *parameters, struct ast_variable *headers);
+typedef int (*ast_websocket_pre_callback)(struct ast_tcptls_session_instance *ser, struct ast_variable *parameters, struct ast_variable *headers, const char *session_id);
 
 /*!
  * \brief Callback for when a new connection for a sub-protocol is established
@@ -360,6 +361,13 @@ AST_OPTIONAL_API(int, ast_websocket_is_secure, (struct ast_websocket *session),
 AST_OPTIONAL_API(int, ast_websocket_set_nonblock, (struct ast_websocket *session), { errno = ENOSYS; return -1;});
 
 /*!
+ * \brief Get the session ID for a WebSocket session.
+ *
+ * \retval session id
+ */
+AST_OPTIONAL_API(const char *, ast_websocket_session_id, (struct ast_websocket *session), { errno = ENOSYS; return NULL;});
+
+/*!
  * \brief Result code for a websocket client.
  */
 enum ast_websocket_result {
index 0a13c56..be90912 100644 (file)
 /*!
  * \brief Deallocates this vector.
  *
- * If any code to free the elements of this vector need to be run, that should
+ * If any code to free the elements of this vector needs to be run, that should
  * be done prior to this call.
  *
  * \param vec Vector to deallocate.
index ac5b578..9a4f2c9 100644 (file)
@@ -153,7 +153,7 @@ struct ast_json *ast_ari_websocket_session_read(
        "{"                                             \
        "  \"error\": \"InvalidMessage\","              \
        "  \"message\": \"Message validation failed\""  \
-       "}" 
+       "}"
 
 int ast_ari_websocket_session_write(struct ast_ari_websocket_session *session,
        struct ast_json *message)
@@ -196,3 +196,9 @@ void ari_handle_websocket(struct ast_websocket_server *ws_server,
        ast_websocket_uri_cb(ser, &fake_urih, uri, method, get_params,
                headers);
 }
+
+const char *ast_ari_websocket_session_id(
+       const struct ast_ari_websocket_session *session)
+{
+       return ast_websocket_session_id(session->ws_session);
+}
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");
        }
 }
-
index 2b63181..bc763eb 100644 (file)
@@ -52,14 +52,24 @@ struct ast_ari_events_event_websocket_args {
 /*!
  * \brief WebSocket connection for events.
  *
+ * \retval  0 success
+ * \retval -1 error
+ */
+int ast_ari_websocket_events_event_websocket_init(void);
+
+/*!
+ * \brief WebSocket connection for events.
+ *
  * \param ser HTTP TCP/TLS Server Session
  * \param headers HTTP headers
  * \param args Swagger parameters
+ * \param session_id The id of the current session.
  *
  * \retval 0 success
  * \retval non-zero error
  */
-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);
+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);
 
 /*!
  * \brief WebSocket connection for events.
@@ -67,8 +77,10 @@ int ast_ari_websocket_events_event_websocket_attempted(struct ast_tcptls_session
  * \param session ARI WebSocket.
  * \param headers HTTP headers.
  * \param args Swagger parameters.
+ * \param session_id The id of the current session.
  */
-void ast_ari_websocket_events_event_websocket_established(struct ast_ari_websocket_session *session, struct ast_variable *headers, struct ast_ari_events_event_websocket_args *args);
+void ast_ari_websocket_events_event_websocket_established(struct ast_ari_websocket_session *session,
+       struct ast_variable *headers, struct ast_ari_events_event_websocket_args *args);
 /*! Argument struct for ast_ari_events_user_event() */
 struct ast_ari_events_user_event_args {
        /*! Event name */
index 4542339..bf33aea 100644 (file)
@@ -53,7 +53,8 @@ ASTERISK_REGISTER_FILE()
 
 #define MAX_VALS 128
 
-static int ast_ari_events_event_websocket_ws_attempted_cb(struct ast_tcptls_session_instance *ser, struct ast_variable *get_params, struct ast_variable *headers)
+static int ast_ari_events_event_websocket_ws_attempted_cb(struct ast_tcptls_session_instance *ser,
+       struct ast_variable *get_params, struct ast_variable *headers, const char *session_id)
 {
        struct ast_ari_events_event_websocket_args args = {};
        int res = 0;
@@ -113,7 +114,7 @@ static int ast_ari_events_event_websocket_ws_attempted_cb(struct ast_tcptls_sess
                {}
        }
 
-       res = ast_ari_websocket_events_event_websocket_attempted(ser, headers, &args);
+       res = ast_ari_websocket_events_event_websocket_attempted(ser, headers, &args, session_id);
 
 fin: __attribute__((unused))
        if (!response) {
@@ -433,6 +434,10 @@ static int load_module(void)
        int res = 0;
        struct ast_websocket_protocol *protocol;
 
+       if (ast_ari_websocket_events_event_websocket_init() == -1) {
+               return AST_MODULE_LOAD_FAILURE;
+       }
+
        events.ws_server = ast_websocket_server_create();
        if (!events.ws_server) {
                return AST_MODULE_LOAD_FAILURE;
index ecae039..3fe774a 100644 (file)
@@ -38,6 +38,7 @@ ASTERISK_REGISTER_FILE()
 #include "asterisk/file.h"
 #include "asterisk/unaligned.h"
 #include "asterisk/uri.h"
+#include "asterisk/uuid.h"
 
 #define AST_API_MODULE
 #include "asterisk/http_websocket.h"
@@ -74,18 +75,19 @@ ASTERISK_REGISTER_FILE()
 
 /*! \brief Structure definition for session */
 struct ast_websocket {
-       FILE *f;                          /*!< Pointer to the file instance used for writing and reading */
-       int fd;                           /*!< File descriptor for the session, only used for polling */
-       struct ast_sockaddr address;      /*!< Address of the remote client */
-       enum ast_websocket_opcode opcode; /*!< Cached opcode for multi-frame messages */
-       size_t payload_len;               /*!< Length of the payload */
-       char *payload;                    /*!< Pointer to the payload */
-       size_t reconstruct;               /*!< Number of bytes before a reconstructed payload will be returned and a new one started */
-       int timeout;                      /*!< The timeout for operations on the socket */
-       unsigned int secure:1;            /*!< Bit to indicate that the transport is secure */
-       unsigned int closing:1;           /*!< Bit to indicate that the session is in the process of being closed */
-       unsigned int close_sent:1;        /*!< Bit to indicate that the session close opcode has been sent and no further data will be sent */
-       struct websocket_client *client;  /*!< Client object when connected as a client websocket */
+       FILE *f;                           /*!< Pointer to the file instance used for writing and reading */
+       int fd;                            /*!< File descriptor for the session, only used for polling */
+       struct ast_sockaddr address;       /*!< Address of the remote client */
+       enum ast_websocket_opcode opcode;  /*!< Cached opcode for multi-frame messages */
+       size_t payload_len;                /*!< Length of the payload */
+       char *payload;                     /*!< Pointer to the payload */
+       size_t reconstruct;                /*!< Number of bytes before a reconstructed payload will be returned and a new one started */
+       int timeout;                       /*!< The timeout for operations on the socket */
+       unsigned int secure:1;             /*!< Bit to indicate that the transport is secure */
+       unsigned int closing:1;            /*!< Bit to indicate that the session is in the process of being closed */
+       unsigned int close_sent:1;         /*!< Bit to indicate that the session close opcode has been sent and no further data will be sent */
+       struct websocket_client *client;   /*!< Client object when connected as a client websocket */
+       char session_id[AST_UUID_STR_LEN]; /*!< The identifier for the websocket session */
 };
 
 /*! \brief Hashing function for protocols */
@@ -414,6 +416,12 @@ int AST_OPTIONAL_API_NAME(ast_websocket_set_timeout)(struct ast_websocket *sessi
        return 0;
 }
 
+const char * AST_OPTIONAL_API_NAME(ast_websocket_session_id)(struct ast_websocket *session)
+{
+       return session->session_id;
+}
+
+
 /* MAINTENANCE WARNING on ast_websocket_read()!
  *
  * We have to keep in mind during this function that the fact that session->fd seems ready
@@ -764,7 +772,7 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
                        return 0;
                }
 
-               if (!(session = ao2_alloc(sizeof(*session), session_destroy_fn))) {
+               if (!(session = ao2_alloc(sizeof(*session) + AST_UUID_STR_LEN + 1, session_destroy_fn))) {
                        ast_log(LOG_WARNING, "WebSocket connection from '%s' could not be accepted\n",
                                ast_sockaddr_stringify(&ser->remote_address));
                        websocket_bad_request(ser);
@@ -773,8 +781,16 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
                }
                session->timeout =  AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
 
+               /* Generate the session id */
+               if (!ast_uuid_generate_str(session->session_id, sizeof(session->session_id))) {
+                       ast_log(LOG_WARNING, "WebSocket connection from '%s' could not be accepted - failed to generate a session id\n",
+                               ast_sockaddr_stringify(&ser->remote_address));
+                       ast_http_error(ser, 500, "Internal Server Error", "Allocation failed");
+                       return 0;
+               }
+
                if (protocol_handler->session_attempted
-                   && protocol_handler->session_attempted(ser, get_vars, headers)) {
+                   && protocol_handler->session_attempted(ser, get_vars, headers, session->session_id)) {
                        ast_debug(3, "WebSocket connection from '%s' rejected by protocol handler '%s'\n",
                                ast_sockaddr_stringify(&ser->remote_address), protocol_handler->name);
                        ao2_ref(protocol_handler, -1);
index d3f40b6..f28e832 100644 (file)
@@ -97,14 +97,28 @@ void ast_ari_{{c_name}}_{{c_nickname}}(struct ast_variable *headers, struct ast_
  * {{{notes}}}
 {{/notes}}
  *
+ * \retval  0 success
+ * \retval -1 error
+ */
+int ast_ari_websocket_{{c_name}}_{{c_nickname}}_init(void);
+
+/*!
+ * \brief {{summary}}
+{{#notes}}
+ *
+ * {{{notes}}}
+{{/notes}}
+ *
  * \param ser HTTP TCP/TLS Server Session
  * \param headers HTTP headers
  * \param args Swagger parameters
+ * \param session_id The id of the current session.
  *
  * \retval 0 success
  * \retval non-zero error
  */
-int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_session_instance *ser, struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
+int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_session_instance *ser,
+       struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args, const char *session_id);
 
 /*!
  * \brief {{summary}}
@@ -116,8 +130,10 @@ int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_sess
  * \param session ARI WebSocket.
  * \param headers HTTP headers.
  * \param args Swagger parameters.
+ * \param session_id The id of the current session.
  */
-void ast_ari_websocket_{{c_name}}_{{c_nickname}}_established(struct ast_ari_websocket_session *session, struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
+void ast_ari_websocket_{{c_name}}_{{c_nickname}}_established(struct ast_ari_websocket_session *session,
+       struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
 {{/is_websocket}}
 {{/operations}}
 {{/apis}}
index 7fe360e..36ca035 100644 (file)
@@ -137,7 +137,8 @@ fin: __attribute__((unused))
 }
 {{/is_req}}
 {{#is_websocket}}
-static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_session_instance *ser, struct ast_variable *get_params, struct ast_variable *headers)
+static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_session_instance *ser,
+       struct ast_variable *get_params, struct ast_variable *headers, const char *session_id)
 {
        struct ast_ari_{{c_name}}_{{c_nickname}}_args args = {};
 {{#has_parameters}}
@@ -156,7 +157,7 @@ static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_s
 
 {{> param_parsing}}
 
-       res = ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(ser, headers, &args);
+       res = ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(ser, headers, &args, session_id);
 
 fin: __attribute__((unused))
        if (!response) {
@@ -255,6 +256,10 @@ static int load_module(void)
 {{#has_websocket}}
        struct ast_websocket_protocol *protocol;
 
+       if (ast_ari_websocket_{{c_name}}_{{c_nickname}}_init() == -1) {
+               return AST_MODULE_LOAD_FAILURE;
+       }
+
        {{full_name}}.ws_server = ast_websocket_server_create();
        if (!{{full_name}}.ws_server) {
                return AST_MODULE_LOAD_FAILURE;