Merge "ARI: Channels added to Stasis application during WebSocket creation ..."
This commit is contained in:
commit
b002e09214
|
@ -187,6 +187,16 @@ struct ast_json *ast_ari_websocket_session_read(
|
|||
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.
|
||||
*
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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
|
||||
|
@ -359,6 +360,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.
|
||||
*/
|
||||
|
|
|
@ -110,7 +110,7 @@
|
|||
/*!
|
||||
* \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.
|
||||
|
|
|
@ -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);
|
||||
}
|
||||
|
|
|
@ -27,226 +27,455 @@
|
|||
|
||||
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 Explicitly shutdown a session.
|
||||
*
|
||||
* 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.
|
||||
*
|
||||
* \param session Session info struct.
|
||||
*/
|
||||
static void session_shutdown(struct event_session *session)
|
||||
{
|
||||
struct ao2_iterator i;
|
||||
char *app;
|
||||
SCOPED_AO2LOCK(lock, 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. */
|
||||
};
|
||||
|
||||
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;
|
||||
session->ws_session = NULL;
|
||||
}
|
||||
|
||||
static void session_dtor(void *obj)
|
||||
{
|
||||
#ifdef AST_DEVMODE /* Avoid unused variable warning */
|
||||
struct event_session *session = obj;
|
||||
#endif
|
||||
|
||||
/* session_shutdown should have been called before */
|
||||
ast_assert(session->ws_session == NULL);
|
||||
ast_assert(session->websocket_apps == NULL);
|
||||
}
|
||||
|
||||
static void session_cleanup(struct event_session *session)
|
||||
{
|
||||
session_shutdown(session);
|
||||
ao2_cleanup(session);
|
||||
}
|
||||
|
||||
static struct event_session *session_create(
|
||||
struct ast_ari_websocket_session *ws_session)
|
||||
{
|
||||
RAII_VAR(struct event_session *, session, NULL, ao2_cleanup);
|
||||
|
||||
session = ao2_alloc(sizeof(*session), session_dtor);
|
||||
|
||||
session->ws_session = ws_session;
|
||||
session->websocket_apps =
|
||||
ast_str_container_alloc(APPS_NUM_BUCKETS);
|
||||
|
||||
if (!session->websocket_apps) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
ao2_ref(session, +1);
|
||||
return session;
|
||||
}
|
||||
/*! \brief Local registry for created \ref event_session objects. */
|
||||
static struct ao2_container *event_session_registry;
|
||||
|
||||
/*!
|
||||
* \brief Callback handler for Stasis application messages.
|
||||
*
|
||||
* \internal
|
||||
*
|
||||
* \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 app_handler(void *data, const char *app_name,
|
||||
struct ast_json *message)
|
||||
static void stasis_app_message_handler(
|
||||
void *data, const char *app_name, struct ast_json *message)
|
||||
{
|
||||
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")),
|
||||
"");
|
||||
const char *msg_type, *msg_application;
|
||||
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
|
||||
/* Determine if we've been replaced */
|
||||
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")), "");
|
||||
|
||||
/* 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);
|
||||
}
|
||||
|
||||
res = ast_json_object_set(message, "application",
|
||||
ast_json_string_create(app_name));
|
||||
if(res != 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
ao2_lock(session);
|
||||
if (session->ws_session) {
|
||||
/* 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);
|
||||
}
|
||||
|
||||
/*!
|
||||
* \brief Register for all of the apps given.
|
||||
* \param session Session info struct.
|
||||
* \param app_name Name of application to register.
|
||||
* \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 session_register_app(struct event_session *session,
|
||||
const char *app_name)
|
||||
static int event_session_compare(void *obj, void *arg, int flags)
|
||||
{
|
||||
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;
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
/*!
|
||||
* \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)
|
||||
{
|
||||
struct ao2_iterator i;
|
||||
char *app;
|
||||
int j;
|
||||
SCOPED_AO2LOCK(lock, session);
|
||||
|
||||
ast_assert(session->ws_session != NULL);
|
||||
ast_assert(session->websocket_apps != NULL);
|
||||
|
||||
if (ast_strlen_zero(app_name)) {
|
||||
return -1;
|
||||
/* 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;
|
||||
}
|
||||
|
||||
if (ast_str_container_add(session->websocket_apps, app_name)) {
|
||||
ast_ari_websocket_session_write(session->ws_session,
|
||||
ast_ari_oom_json());
|
||||
return -1;
|
||||
/* 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;
|
||||
}
|
||||
|
||||
/*!
|
||||
* \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)
|
||||
{
|
||||
int i;
|
||||
|
||||
ast_assert(session != NULL);
|
||||
|
||||
ao2_lock(session);
|
||||
|
||||
session->ws_session = ws_session;
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
stasis_app_register(app_name, app_handler, session);
|
||||
AST_VECTOR_RESET(&session->message_queue, AST_VECTOR_ELEM_CLEANUP_NOOP);
|
||||
ao2_unlock(session);
|
||||
}
|
||||
|
||||
/*!
|
||||
* \brief Processes cleanup actions for a \ref event_session object.
|
||||
*
|
||||
* \internal
|
||||
*
|
||||
* \param session The event session object to cleanup (\ref event_session).
|
||||
*/
|
||||
static void event_session_cleanup(struct event_session *session)
|
||||
{
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
|
||||
event_session_shutdown(session);
|
||||
ao2_unlink(event_session_registry, 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 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 event_session_allocation_error_handler(
|
||||
struct event_session *session, enum event_session_error_type error,
|
||||
struct ast_tcptls_session_instance *ser)
|
||||
{
|
||||
/* Notify the client */
|
||||
switch (error) {
|
||||
case ERROR_TYPE_STASIS_REGISTRATION:
|
||||
ast_http_error(ser, 500, "Internal Server Error",
|
||||
"Stasis registration failed");
|
||||
break;
|
||||
|
||||
case ERROR_TYPE_OOM:
|
||||
ast_http_error(ser, 500, "Internal Server Error",
|
||||
"Allocation failed");
|
||||
break;
|
||||
|
||||
case ERROR_TYPE_MISSING_APP_PARAM:
|
||||
ast_http_error(ser, 400, "Bad Request",
|
||||
"HTTP request is missing param: [app]");
|
||||
break;
|
||||
|
||||
case ERROR_TYPE_INVALID_APP_PARAM:
|
||||
ast_http_error(ser, 400, "Bad Request",
|
||||
"Invalid application provided in param [app].");
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
/* Cleanup the session */
|
||||
event_session_cleanup(session);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/*!
|
||||
* \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)
|
||||
{
|
||||
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) {
|
||||
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) {
|
||||
const char *app = args->app[i];
|
||||
|
||||
if (ast_strlen_zero(app)) {
|
||||
return event_session_allocation_error_handler(
|
||||
session, ERROR_TYPE_INVALID_APP_PARAM, ser);
|
||||
}
|
||||
|
||||
if (ast_str_container_add(session->websocket_apps, app)) {
|
||||
return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/* 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;
|
||||
}
|
||||
|
||||
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_init(void)
|
||||
{
|
||||
int res = 0;
|
||||
size_t i, j;
|
||||
/* Try to instantiate the registry */
|
||||
event_session_registry = ao2_container_alloc(EVENT_SESSION_NUM_BUCKETS,
|
||||
event_session_hash,
|
||||
event_session_compare);
|
||||
|
||||
ast_debug(3, "/events WebSocket attempted\n");
|
||||
|
||||
if (args->app_count == 0) {
|
||||
ast_http_error(ser, 400, "Bad Request", "Missing param 'app'");
|
||||
if (!event_session_registry) {
|
||||
/* This is bad, bad. */
|
||||
ast_log(LOG_WARNING,
|
||||
"Failed to allocate the local registry for websocket applications\n");
|
||||
return -1;
|
||||
}
|
||||
|
||||
for (i = 0; i < args->app_count; ++i) {
|
||||
if (ast_strlen_zero(args->app[i])) {
|
||||
res = -1;
|
||||
break;
|
||||
}
|
||||
|
||||
res |= stasis_app_register(args->app[i], app_handler, NULL);
|
||||
}
|
||||
|
||||
if (res) {
|
||||
for (j = 0; j < i; ++j) {
|
||||
stasis_app_unregister(args->app[j]);
|
||||
}
|
||||
ast_http_error(ser, 400, "Bad Request", "Invalid application provided in param 'app'.");
|
||||
}
|
||||
|
||||
return res;
|
||||
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_attempted(
|
||||
struct ast_tcptls_session_instance *ser, struct ast_variable *headers,
|
||||
struct ast_ari_events_event_websocket_args *args, const char *session_id)
|
||||
{
|
||||
RAII_VAR(struct event_session *, session, NULL, session_cleanup);
|
||||
ast_debug(3, "/events WebSocket attempted\n");
|
||||
|
||||
/* Create the event session */
|
||||
return event_session_alloc(ser, args, session_id);
|
||||
}
|
||||
|
||||
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;
|
||||
int res;
|
||||
size_t i;
|
||||
const char *session_id;
|
||||
|
||||
ast_debug(3, "/events WebSocket connection\n");
|
||||
ast_debug(3, "/events WebSocket established\n");
|
||||
|
||||
session = session_create(ws_session);
|
||||
if (!session) {
|
||||
ast_ari_websocket_session_write(ws_session, ast_ari_oom_json());
|
||||
return;
|
||||
}
|
||||
ast_assert(ws_session != NULL);
|
||||
|
||||
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]);
|
||||
}
|
||||
session_id = ast_ari_websocket_session_id(ws_session);
|
||||
|
||||
if (ao2_container_count(session->websocket_apps) == 0) {
|
||||
RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
|
||||
/* Find the event_session and update its websocket */
|
||||
session = ao2_find(event_session_registry, session_id, OBJ_SEARCH_KEY);
|
||||
|
||||
msg = ast_json_pack("{s: s, s: [s]}",
|
||||
"type", "MissingParams",
|
||||
"params", "app");
|
||||
if (!msg) {
|
||||
msg = ast_json_ref(ast_ari_oom_json());
|
||||
}
|
||||
|
||||
ast_ari_websocket_session_write(session->ws_session, msg);
|
||||
return;
|
||||
}
|
||||
|
||||
if (res != 0) {
|
||||
ast_ari_websocket_session_write(ws_session, ast_ari_oom_json());
|
||||
return;
|
||||
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");
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -49,17 +49,27 @@ struct ast_ari_events_event_websocket_args {
|
|||
char *app_parse;
|
||||
};
|
||||
|
||||
/*!
|
||||
* \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 */
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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);
|
||||
|
|
|
@ -90,6 +90,18 @@ void ast_ari_{{c_name}}_{{c_nickname}}(struct ast_variable *headers, struct ast_
|
|||
{{/is_req}}
|
||||
{{#is_websocket}}
|
||||
|
||||
/*!
|
||||
* \brief {{summary}}
|
||||
{{#notes}}
|
||||
*
|
||||
* {{{notes}}}
|
||||
{{/notes}}
|
||||
*
|
||||
* \retval 0 success
|
||||
* \retval -1 error
|
||||
*/
|
||||
int ast_ari_websocket_{{c_name}}_{{c_nickname}}_init(void);
|
||||
|
||||
/*!
|
||||
* \brief {{summary}}
|
||||
{{#notes}}
|
||||
|
@ -100,11 +112,13 @@ void ast_ari_{{c_name}}_{{c_nickname}}(struct ast_variable *headers, struct ast_
|
|||
* \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}}
|
||||
|
|
|
@ -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;
|
||||
|
|
Loading…
Reference in New Issue