ari_websockets.c File Reference

WebSocket support for RESTful API's. More...

#include "asterisk.h"
#include "resource_events.h"
#include "ari_websockets.h"
#include "internal.h"
#include "asterisk/app.h"
#include "asterisk/ari.h"
#include "asterisk/astobj2.h"
#include "asterisk/http_websocket.h"
#include "asterisk/module.h"
#include "asterisk/stasis_app.h"

Include dependency graph for ari_websockets.c:

Go to the source code of this file.

Macros
#define	APPS_NUM_BUCKETS   7
#define	ARI_WS_SESSION_NUM_BUCKETS   23
#define	MAX_VALS   128
#define	MESSAGES_INIT_SIZE   23
#define	VALIDATION_FAILED

Functions
	AO2_STRING_FIELD_CMP_FN (ari_ws_session, session_id)
	AO2_STRING_FIELD_HASH_FN (ari_ws_session, session_id)
void	ari_handle_websocket (struct ast_tcptls_session_instance *ser, const char *uri, enum ast_http_method method, struct ast_variable *get_params, struct ast_variable *headers)
	Wrapper for invoking the websocket code for an incoming connection. More...
int	ari_websocket_load_module (void)
void	ari_websocket_send_event (struct ari_ws_session *ari_ws_session, const char *app_name, struct ast_json *message, int debug_app)
	Callback handler for Stasis application messages. More...
int	ari_websocket_unload_module (void)
static void	ari_ws_session_cleanup (struct ari_ws_session *ari_ws_session)
static int	ari_ws_session_create (int(*validator)(struct ast_json *), struct ast_tcptls_session_instance *ser, struct ast_ari_events_event_websocket_args *args, const char *session_id)
static void	ari_ws_session_dtor (void *obj)
static struct ast_json *	ari_ws_session_read (struct ari_ws_session *ari_ws_session)
static void	ari_ws_session_registry_dtor (void)
static void	ari_ws_session_reset (struct ari_ws_session *ari_ws_session)
static int	ari_ws_session_shutdown_cb (void *ari_ws_session, void *arg, int flags)
static int	ari_ws_session_update (struct ari_ws_session *ari_ws_session, struct ast_websocket *ast_ws_session)
static int	ari_ws_session_write (struct ari_ws_session *ari_ws_session, struct ast_json *message)
static int	null_validator (struct ast_json *json)
	Validator that always succeeds. More...
static int	parse_app_args (struct ast_variable *get_params, struct ast_ari_response *response, struct ast_ari_events_event_websocket_args *args)
static void	stasis_app_message_handler (void *data, const char *app_name, struct ast_json *message)
static int	websocket_attempted_cb (struct ast_tcptls_session_instance *ser, struct ast_variable *get_params, struct ast_variable *headers, const char *session_id)
static void	websocket_established_cb (struct ast_websocket *ast_ws_session, struct ast_variable *get_params, struct ast_variable *upgrade_headers)

Variables
static struct ao2_container *	ari_ws_session_registry
	Local registry for created event_session objects. More...
struct ast_websocket_server *	ast_ws_server

Detailed Description

WebSocket support for RESTful API's.

Author

David M. Lee, II dlee@.nosp@m.digi.nosp@m.um.co.nosp@m.m

Definition in file ari_websockets.c.

Macro Definition Documentation

APPS_NUM_BUCKETS

#define APPS_NUM_BUCKETS   7

Number of buckets for a websocket apps container. Remember to keep it a prime number!

Definition at line 45 of file ari_websockets.c.

ARI_WS_SESSION_NUM_BUCKETS

#define ARI_WS_SESSION_NUM_BUCKETS   23

Number of buckets for the event session registry. Remember to keep it a prime number!

Definition at line 42 of file ari_websockets.c.

MAX_VALS

#define MAX_VALS   128

Definition at line 56 of file ari_websockets.c.

MESSAGES_INIT_SIZE

#define MESSAGES_INIT_SIZE   23

Initial size of a message queue.

Definition at line 48 of file ari_websockets.c.

VALIDATION_FAILED

#define VALIDATION_FAILED

Value:

    "{"                     \
    "  \"error\": \"InvalidMessage\","      \
    "  \"message\": \"Message validation failed\""  \
    "}"

Definition at line 66 of file ari_websockets.c.

Function Documentation

AO2_STRING_FIELD_CMP_FN()

AO2_STRING_FIELD_CMP_FN	(	ari_ws_session	,
		session_id
	)

AO2_STRING_FIELD_HASH_FN()

AO2_STRING_FIELD_HASH_FN	(	ari_ws_session	,
		session_id
	)

ari_handle_websocket()

void ari_handle_websocket	(	struct ast_tcptls_session_instance *	ser,
		const char *	uri,
		enum ast_http_method	method,
		struct ast_variable *	get_params,
		struct ast_variable *	headers
	)

Wrapper for invoking the websocket code for an incoming connection.

Parameters

ws_server	WebSocket server to invoke.
ser	HTTP session.
uri	Requested URI.
method	Requested HTTP method.
get_params	Parsed query parameters.
headers	Parsed HTTP headers.

Definition at line 221 of file ari_websockets.c.

{
    struct ast_http_uri fake_urih = {
        .data = ast_ws_server,
    };
 
    ast_websocket_uri_cb(ser, &fake_urih, uri, method, get_params,
        headers);
}

References ast_websocket_uri_cb(), ast_ws_server, ast_http_uri::data, method, and ast_http_uri::uri.

Referenced by ast_ari_invoke().

ari_websocket_load_module()

int ari_websocket_load_module

(

void

)

Definition at line 722 of file ari_websockets.c.

{
    int res = 0;
    struct ast_websocket_protocol *protocol;
 
    ari_ws_session_registry = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
        ARI_WS_SESSION_NUM_BUCKETS, ari_ws_session_hash_fn,
        NULL, ari_ws_session_cmp_fn);
    if (!ari_ws_session_registry) {
        ast_log(LOG_WARNING,
                "Failed to allocate the local registry for websocket applications\n");
        return AST_MODULE_LOAD_DECLINE;
    }
 
    ast_ws_server = ast_websocket_server_create();
    if (!ast_ws_server) {
        ari_ws_session_registry_dtor();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    protocol = ast_websocket_sub_protocol_alloc("ari");
    if (!protocol) {
        ao2_ref(ast_ws_server, -1);
        ast_ws_server = NULL;
        ari_ws_session_registry_dtor();
        return AST_MODULE_LOAD_DECLINE;
    }
    protocol->session_attempted = websocket_attempted_cb;
    protocol->session_established = websocket_established_cb;
    res = ast_websocket_server_add_protocol2(ast_ws_server, protocol);
 
    return res == 0 ? AST_MODULE_LOAD_SUCCESS : AST_MODULE_LOAD_DECLINE;
}

References AO2_ALLOC_OPT_LOCK_MUTEX, ao2_container_alloc_hash, ao2_ref, ARI_WS_SESSION_NUM_BUCKETS, ari_ws_session_registry, ari_ws_session_registry_dtor(), ast_log, AST_MODULE_LOAD_DECLINE, AST_MODULE_LOAD_SUCCESS, ast_websocket_server_add_protocol2(), ast_websocket_server_create(), ast_websocket_sub_protocol_alloc(), ast_ws_server, LOG_WARNING, NULL, ast_websocket_protocol::session_attempted, ast_websocket_protocol::session_established, websocket_attempted_cb(), and websocket_established_cb().

Referenced by load_module().

ari_websocket_send_event()

void ari_websocket_send_event	(	struct ari_ws_session *	ari_ws_session,
		const char *	app_name,
		struct ast_json *	message,
		int	debug_app
	)

Callback handler for Stasis application messages.

Definition at line 244 of file ari_websockets.c.

{
    char *remote_addr = ast_sockaddr_stringify(
            ast_websocket_remote_address(ari_ws_session->ast_ws_session));
    const char *msg_type, *msg_application, *msg_timestamp, *msg_ast_id;
    SCOPE_ENTER(4, "%s: Dispatching message from Stasis app '%s'\n", remote_addr, app_name);
 
    ast_assert(ari_ws_session != NULL);
 
    ao2_lock(ari_ws_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(ari_ws_session->websocket_apps, msg_application,
            OBJ_UNLINK | OBJ_NODATA);
    }
 
    msg_timestamp = S_OR(
        ast_json_string_get(ast_json_object_get(message, "timestamp")), "");
    if (ast_strlen_zero(msg_timestamp)) {
        if (ast_json_object_set(message, "timestamp", ast_json_timeval(ast_tvnow(), NULL))) {
            ao2_unlock(ari_ws_session);
            SCOPE_EXIT_LOG_RTN(LOG_WARNING,
                "%s: Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
                remote_addr, msg_type, msg_application);
        }
    }
 
    msg_ast_id = S_OR(
        ast_json_string_get(ast_json_object_get(message, "asterisk_id")), "");
    if (ast_strlen_zero(msg_ast_id)) {
        char eid[20];
 
        if (ast_json_object_set(message, "asterisk_id",
            ast_json_string_create(ast_eid_to_str(eid, sizeof(eid), &ast_eid_default)))) {
            ao2_unlock(ari_ws_session);
            SCOPE_EXIT_LOG_RTN(LOG_WARNING,
                "%s: Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
                remote_addr, msg_type, msg_application);
        }
    }
 
    /* 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))) {
        ao2_unlock(ari_ws_session);
        SCOPE_EXIT_LOG_RTN(LOG_WARNING,
            "%s: Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
            remote_addr, msg_type, msg_application);
    }
 
    if (!ari_ws_session) {
        /* If the websocket is NULL, the message goes to the queue */
        if (!AST_VECTOR_APPEND(&ari_ws_session->message_queue, message)) {
            ast_json_ref(message);
        }
        ast_log(LOG_WARNING,
                "%s: Queued '%s' message for Stasis app '%s'; websocket is not ready\n",
                remote_addr,
                msg_type,
                msg_application);
    } else if (stasis_app_event_allowed(app_name, message)) {
 
        if (TRACE_ATLEAST(4) || debug_app) {
            char *str = ast_json_dump_string_format(message, AST_JSON_PRETTY);
 
            ast_verbose("<--- Sending ARI event to %s --->\n%s\n",
                remote_addr,
                str);
            ast_json_free(str);
        }
 
        ari_ws_session_write(ari_ws_session, message);
    }
 
    ao2_unlock(ari_ws_session);
    SCOPE_EXIT("%s: Dispatched '%s' message from Stasis app '%s'\n",
        remote_addr, msg_type, app_name);
}

References ao2_find, ao2_lock, ao2_unlock, app_name(), ari_ws_session_write(), ast_assert, ast_eid_default, ast_eid_to_str(), ast_json_dump_string_format(), ast_json_free(), ast_json_object_get(), ast_json_object_set(), AST_JSON_PRETTY, ast_json_ref(), ast_json_string_create(), ast_json_string_get(), ast_json_timeval(), ast_log, ast_sockaddr_stringify(), ast_strlen_zero(), ast_tvnow(), AST_VECTOR_APPEND, ast_verbose(), ast_websocket_remote_address(), ari_ws_session::ast_ws_session, LOG_WARNING, ari_ws_session::message_queue, msg_timestamp(), NULL, OBJ_NODATA, OBJ_UNLINK, S_OR, SCOPE_ENTER, SCOPE_EXIT, SCOPE_EXIT_LOG_RTN, stasis_app_event_allowed(), str, TRACE_ATLEAST, and ari_ws_session::websocket_apps.

Referenced by ari_ws_session_read(), send_rest_response(), and stasis_app_message_handler().

ari_websocket_unload_module()

int ari_websocket_unload_module

(

void

)

Definition at line 711 of file ari_websockets.c.

{
    ari_ws_session_registry_dtor();
    ao2_cleanup(ast_ws_server);
    ast_ws_server = NULL;
    return 0;
}

References ao2_cleanup, ari_ws_session_registry_dtor(), ast_ws_server, and NULL.

Referenced by unload_module().

ari_ws_session_cleanup()

static void ari_ws_session_cleanup ( struct ari_ws_session *  ari_ws_session )

static

Definition at line 447 of file ari_websockets.c.

{
    if (!ari_ws_session) {
        return;
    }
 
    ari_ws_session_reset(ari_ws_session);
    if (ari_ws_session_registry) {
        ao2_unlink(ari_ws_session_registry, ari_ws_session);
    }
    ao2_ref(ari_ws_session, -1);
}

References ao2_ref, ao2_unlink, ari_ws_session_registry, and ari_ws_session_reset().

Referenced by ari_ws_session_shutdown_cb(), and websocket_established_cb().

ari_ws_session_create()

static int ari_ws_session_create ( int(*)(struct ast_json *)  validator, struct ast_tcptls_session_instance *  ser, struct ast_ari_events_event_websocket_args *  args, const char *  session_id  )

static

Definition at line 477 of file ari_websockets.c.

{
    RAII_VAR(struct ari_ws_session *, ari_ws_session, NULL, ao2_cleanup);
    int (* register_handler)(const char *, stasis_app_cb handler, void *data);
    size_t size, i;
 
    if (validator == NULL) {
        validator = null_validator;
    }
 
    size = sizeof(*ari_ws_session) + strlen(session_id) + 1;
 
    ari_ws_session = ao2_alloc(size, ari_ws_session_dtor);
    if (!ari_ws_session) {
        return -1;
    }
 
    ari_ws_session->app_name = ast_strdup(args->app_parse);
    if (!ari_ws_session->app_name) {
        ast_http_error(ser, 500, "Internal Server Error",
            "Allocation failed");
        return -1;
    }
 
    strcpy(ari_ws_session->session_id, session_id); /* Safe */
 
    /* Instantiate the hash table for Stasis apps */
    ari_ws_session->websocket_apps =
        ast_str_container_alloc(APPS_NUM_BUCKETS);
    if (!ari_ws_session->websocket_apps) {
        ast_http_error(ser, 500, "Internal Server Error",
            "Allocation failed");
        return -1;
    }
 
    /* Instantiate the message queue */
    if (AST_VECTOR_INIT(&ari_ws_session->message_queue, MESSAGES_INIT_SIZE)) {
        ast_http_error(ser, 500, "Internal Server Error",
            "Allocation failed");
        ao2_cleanup(ari_ws_session->websocket_apps);
        return -1;
    }
 
    /* Register the apps with Stasis */
    if (args->subscribe_all) {
        register_handler = &stasis_app_register_all;
    } else {
        register_handler = &stasis_app_register;
    }
 
    for (i = 0; i < args->app_count; ++i) {
        const char *app = args->app[i];
 
        if (ast_strlen_zero(app)) {
            ast_http_error(ser, 400, "Bad Request",
                "Invalid application provided in param [app].");
            ari_ws_session_reset(ari_ws_session);
            return -1;
        }
 
        if (ast_str_container_add(ari_ws_session->websocket_apps, app)) {
            ast_http_error(ser, 500, "Internal Server Error",
                "Allocation failed");
            ari_ws_session_reset(ari_ws_session);
            return -1;
        }
 
        if (register_handler(app, stasis_app_message_handler, ari_ws_session)) {
            ast_log(LOG_WARNING, "Stasis registration failed for application: '%s'\n", app);
            ast_http_error(ser, 500, "Internal Server Error",
                "Stasis registration failed");
            ari_ws_session_reset(ari_ws_session);
            return -1;
        }
    }
 
    ari_ws_session->validator = validator;
 
    /*
     * Add the event session to the session registry.
     * When this functions returns, the registry will have
     * the only reference to the session.
     */
    if (!ao2_link(ari_ws_session_registry, ari_ws_session)) {
        ast_http_error(ser, 500, "Internal Server Error",
            "Allocation failed");
        ari_ws_session_reset(ari_ws_session);
        return -1;
    }
 
    return 0;
}

References ao2_alloc, ao2_cleanup, ao2_link, app, ari_ws_session::app_name, APPS_NUM_BUCKETS, args, ari_ws_session_dtor(), ari_ws_session_registry, ari_ws_session_reset(), ast_http_error(), ast_log, ast_str_container_add(), ast_str_container_alloc, ast_strdup, ast_strlen_zero(), AST_VECTOR_INIT, handler(), LOG_WARNING, ari_ws_session::message_queue, MESSAGES_INIT_SIZE, NULL, null_validator(), RAII_VAR, ari_ws_session::session_id, stasis_app_message_handler(), stasis_app_register(), stasis_app_register_all(), ari_ws_session::validator, and ari_ws_session::websocket_apps.

Referenced by websocket_attempted_cb().

ari_ws_session_dtor()

static void ari_ws_session_dtor ( void *  obj )

static

Definition at line 465 of file ari_websockets.c.

{
    struct ari_ws_session *ari_ws_session = obj;
 
    ast_free(ari_ws_session->app_name);
    if (!ari_ws_session->ast_ws_session) {
        return;
    }
    ast_websocket_unref(ari_ws_session->ast_ws_session);
    ari_ws_session->ast_ws_session = NULL;
}

References ari_ws_session::app_name, ast_free, ast_websocket_unref(), ari_ws_session::ast_ws_session, and NULL.

Referenced by ari_ws_session_create().

ari_ws_session_read()

static struct ast_json * ari_ws_session_read ( struct ari_ws_session *  ari_ws_session )

static

Definition at line 153 of file ari_websockets.c.

{
    RAII_VAR(struct ast_json *, message, NULL, ast_json_unref);
 
    if (ast_websocket_fd(ari_ws_session->ast_ws_session) < 0) {
        return NULL;
    }
 
    while (!message) {
        int res;
        char *payload;
        uint64_t payload_len;
        enum ast_websocket_opcode opcode;
        int fragmented;
 
        res = ast_wait_for_input(
            ast_websocket_fd(ari_ws_session->ast_ws_session), -1);
 
        if (res <= 0) {
            ast_log(LOG_WARNING, "WebSocket poll error: %s\n",
                strerror(errno));
            return NULL;
        }
 
        res = ast_websocket_read(ari_ws_session->ast_ws_session, &payload,
            &payload_len, &opcode, &fragmented);
 
        if (res != 0) {
            ast_log(LOG_WARNING, "WebSocket read error: %s\n",
                strerror(errno));
            return NULL;
        }
 
        switch (opcode) {
        case AST_WEBSOCKET_OPCODE_CLOSE:
            ast_debug(1, "WebSocket closed\n");
            return NULL;
        case AST_WEBSOCKET_OPCODE_TEXT:
            message = ast_json_load_buf(payload, payload_len, NULL);
            if (message == NULL) {
                struct ast_json *error = ast_json_pack(
                    "{s:s, s:s, s:s, s:i, s:s, s:s }",
                    "type", "RESTResponse",
                    "transaction_id", "",
                    "request_id", "",
                    "status_code", 400,
                    "reason_phrase", "Failed to parse request message JSON",
                    "uri", ""
                    );
                ari_websocket_send_event(ari_ws_session, ari_ws_session->app_name,
                    error, 0);
                ast_json_unref(error);
                ast_log(LOG_WARNING,
                    "WebSocket input failed to parse\n");
 
            }
 
            break;
        default:
            /* Ignore all other message types */
            break;
        }
    }
 
    return ast_json_ref(message);
}

References ari_ws_session::app_name, ari_websocket_send_event(), ast_debug, ast_json_load_buf(), ast_json_pack(), ast_json_ref(), ast_json_unref(), ast_log, ast_wait_for_input(), ast_websocket_fd(), AST_WEBSOCKET_OPCODE_CLOSE, AST_WEBSOCKET_OPCODE_TEXT, ast_websocket_read(), ari_ws_session::ast_ws_session, errno, error(), LOG_WARNING, NULL, and RAII_VAR.

Referenced by websocket_established_cb().

ari_ws_session_registry_dtor()

static void ari_ws_session_registry_dtor ( void  )

static

Definition at line 698 of file ari_websockets.c.

{
    if (!ari_ws_session_registry) {
        return;
    }
 
    ao2_callback(ari_ws_session_registry, OBJ_MULTIPLE | OBJ_NODATA,
        ari_ws_session_shutdown_cb, NULL);
 
    ao2_cleanup(ari_ws_session_registry);
    ari_ws_session_registry = NULL;
}

References ao2_callback, ao2_cleanup, ari_ws_session_registry, ari_ws_session_shutdown_cb(), NULL, OBJ_MULTIPLE, and OBJ_NODATA.

Referenced by ari_websocket_load_module(), and ari_websocket_unload_module().

ari_ws_session_reset()

static void ari_ws_session_reset ( struct ari_ws_session *  ari_ws_session )

static

Definition at line 414 of file ari_websockets.c.

{
    struct ao2_iterator i;
    char *app;
    int j;
    SCOPED_AO2LOCK(lock, ari_ws_session);
 
    /* Clean up the websocket_apps container */
    if (ari_ws_session->websocket_apps) {
        i = ao2_iterator_init(ari_ws_session->websocket_apps, 0);
        while ((app = ao2_iterator_next(&i))) {
            stasis_app_unregister(app);
            ao2_cleanup(app);
        }
        ao2_iterator_destroy(&i);
        ao2_cleanup(ari_ws_session->websocket_apps);
        ari_ws_session->websocket_apps = NULL;
    }
 
    /* Clean up the message_queue container */
    for (j = 0; j < AST_VECTOR_SIZE(&ari_ws_session->message_queue); j++) {
        struct ast_json *msg = AST_VECTOR_GET(&ari_ws_session->message_queue, j);
        ast_json_unref(msg);
    }
    AST_VECTOR_FREE(&ari_ws_session->message_queue);
}

References ao2_cleanup, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, app, ast_json_unref(), AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_SIZE, lock, ari_ws_session::message_queue, NULL, SCOPED_AO2LOCK, stasis_app_unregister(), and ari_ws_session::websocket_apps.

Referenced by ari_ws_session_cleanup(), and ari_ws_session_create().

ari_ws_session_shutdown_cb()

static int ari_ws_session_shutdown_cb ( void *  ari_ws_session, void *  arg, int  flags  )

static

Definition at line 691 of file ari_websockets.c.

{
    ari_ws_session_cleanup(ari_ws_session);
 
    return 0;
}

References ari_ws_session_cleanup().

Referenced by ari_ws_session_registry_dtor().

ari_ws_session_update()

static int ari_ws_session_update ( struct ari_ws_session *  ari_ws_session, struct ast_websocket *  ast_ws_session  )

static

Definition at line 111 of file ari_websockets.c.

{
    RAII_VAR(struct ast_ari_conf *, config, ast_ari_config_get(), ao2_cleanup);
    int i;
 
    if (ast_ws_session == NULL) {
        return -1;
    }
 
    if (config == NULL || config->general == NULL) {
        return -1;
    }
 
    if (ast_websocket_set_nonblock(ast_ws_session) != 0) {
        ast_log(LOG_ERROR,
            "ARI web socket failed to set nonblock; closing: %s\n",
            strerror(errno));
        return -1;
    }
 
    if (ast_websocket_set_timeout(ast_ws_session, config->general->write_timeout)) {
        ast_log(LOG_WARNING, "Failed to set write timeout %d on ARI web socket\n",
            config->general->write_timeout);
    }
 
    ao2_ref(ast_ws_session, +1);
    ari_ws_session->ast_ws_session = ast_ws_session;
    ao2_lock(ari_ws_session);
    for (i = 0; i < AST_VECTOR_SIZE(&ari_ws_session->message_queue); i++) {
        struct ast_json *msg = AST_VECTOR_GET(&ari_ws_session->message_queue, i);
        ari_ws_session_write(ari_ws_session, msg);
        ast_json_unref(msg);
    }
 
    AST_VECTOR_RESET(&ari_ws_session->message_queue, AST_VECTOR_ELEM_CLEANUP_NOOP);
    ao2_unlock(ari_ws_session);
 
    return 0;
}

References ao2_cleanup, ao2_lock, ao2_ref, ao2_unlock, ari_ws_session_write(), ast_ari_config_get(), ast_json_unref(), ast_log, AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_GET, AST_VECTOR_RESET, AST_VECTOR_SIZE, ast_websocket_set_nonblock(), ast_websocket_set_timeout(), ari_ws_session::ast_ws_session, config, errno, LOG_ERROR, LOG_WARNING, ari_ws_session::message_queue, NULL, and RAII_VAR.

Referenced by websocket_established_cb().

ari_ws_session_write()

static int ari_ws_session_write ( struct ari_ws_session *  ari_ws_session, struct ast_json *  message  )

static

Definition at line 72 of file ari_websockets.c.

{
    RAII_VAR(char *, str, NULL, ast_json_free);
 
#ifdef AST_DEVMODE
    if (!ari_ws_session->validator(message)) {
        ast_log(LOG_ERROR, "Outgoing message failed validation\n");
        return ast_websocket_write_string(ari_ws_session->ast_ws_session, VALIDATION_FAILED);
    }
#endif
 
    str = ast_json_dump_string_format(message, ast_ari_json_format());
 
    if (str == NULL) {
        ast_log(LOG_ERROR, "Failed to encode JSON object\n");
        return -1;
    }
 
    if (ast_websocket_write_string(ari_ws_session->ast_ws_session, str)) {
        ast_log(LOG_NOTICE, "Problem occurred during websocket write to %s, websocket closed\n",
            ast_sockaddr_stringify(ast_websocket_remote_address(ari_ws_session->ast_ws_session)));
        return -1;
    }
    return 0;
}

References ast_ari_json_format(), ast_json_dump_string_format(), ast_json_free(), ast_log, ast_sockaddr_stringify(), ast_websocket_remote_address(), ast_websocket_write_string(), ari_ws_session::ast_ws_session, LOG_ERROR, LOG_NOTICE, NULL, RAII_VAR, str, VALIDATION_FAILED, and ari_ws_session::validator.

Referenced by ari_websocket_send_event(), and ari_ws_session_update().

null_validator()

static int null_validator ( struct ast_json *  json )

static

Validator that always succeeds.

Definition at line 61 of file ari_websockets.c.

{
    return 1;
}

Referenced by ari_ws_session_create().

parse_app_args()

static int parse_app_args ( struct ast_variable *  get_params, struct ast_ari_response *  response, struct ast_ari_events_event_websocket_args *  args  )

static

Definition at line 339 of file ari_websockets.c.

{
    struct ast_variable *i;
    RAII_VAR(char *, app_parse, NULL, ast_free);
 
    for (i = get_params; i; i = i->next) {
        if (strcmp(i->name, "app") == 0) {
            /* Parse comma separated list */
            char *vals[MAX_VALS];
            size_t j;
 
            app_parse = ast_strdup(i->value);
            if (!app_parse) {
                ast_ari_response_alloc_failed(response);
                return -1;
            }
 
            if (strlen(app_parse) == 0) {
                /* ast_app_separate_args can't handle "" */
                args->app_count = 1;
                vals[0] = app_parse;
            } else {
                args->app_count = ast_app_separate_args(
                    app_parse, ',', vals,
                    ARRAY_LEN(vals));
            }
 
            if (args->app_count == 0) {
                ast_ari_response_alloc_failed(response);
                return -1;
            }
 
            if (args->app_count >= MAX_VALS) {
                ast_ari_response_error(response, 400,
                    "Bad Request",
                    "Too many values for app");
                return -1;
            }
 
            args->app = ast_malloc(sizeof(*args->app) * args->app_count);
            if (!args->app) {
                ast_ari_response_alloc_failed(response);
                return -1;
            }
 
            for (j = 0; j < args->app_count; ++j) {
                args->app[j] = (vals[j]);
            }
        } else if (strcmp(i->name, "subscribeAll") == 0) {
            args->subscribe_all = ast_true(i->value);
        }
    }
 
    args->app_parse = app_parse;
    app_parse = NULL;
 
    return 0;
}

References args, ARRAY_LEN, ast_app_separate_args, ast_ari_response_alloc_failed(), ast_ari_response_error(), ast_free, ast_malloc, ast_strdup, ast_true(), MAX_VALS, ast_variable::name, ast_variable::next, NULL, RAII_VAR, and ast_variable::value.

Referenced by websocket_attempted_cb().

stasis_app_message_handler()

static void stasis_app_message_handler ( void *  data, const char *  app_name, struct ast_json *  message  )

static

Definition at line 330 of file ari_websockets.c.

{
    int debug_app = stasis_app_get_debug_by_name(app_name);
    struct ari_ws_session *ari_ws_session = data;
    ast_assert(ari_ws_session != NULL);
    ari_websocket_send_event(ari_ws_session, app_name, message, debug_app);
}

References app_name(), ari_websocket_send_event(), ast_assert, NULL, and stasis_app_get_debug_by_name().

Referenced by ari_ws_session_create().

websocket_attempted_cb()

static int websocket_attempted_cb ( struct ast_tcptls_session_instance *  ser, struct ast_variable *  get_params, struct ast_variable *  headers, const char *  session_id  )

static

Definition at line 579 of file ari_websockets.c.

{
    struct ast_ari_events_event_websocket_args args = {};
    int res = 0;
    RAII_VAR(struct ast_ari_response *, response, NULL, ast_free);
    char *remote_addr = ast_sockaddr_stringify(&ser->remote_address);
 
    response = ast_calloc(1, sizeof(*response));
    if (!response) {
        ast_log(LOG_ERROR, "Failed to create response.\n");
        ast_http_error(ser, 500, "Server Error", "Memory allocation error");
        return -1;
    }
 
    res = parse_app_args(get_params, response, &args);
    if (res != 0) {
        /* Param parsing failure */
        RAII_VAR(char *, msg, NULL, ast_json_free);
        if (response->message) {
            msg = ast_json_dump_string(response->message);
        } else {
            ast_log(LOG_ERROR, "Missing response message\n");
        }
 
        if (msg) {
            ast_http_error(ser, response->response_code, response->response_text, msg);
            return -1;
        }
    }
 
    if (args.app_count == 0) {
        ast_http_error(ser, 400, "Bad Request",
            "HTTP request is missing param: [app]");
        return -1;
    }
 
#if defined(AST_DEVMODE)
    res = ari_ws_session_create(ast_ari_validate_message_fn(),
        ser, &args, session_id);
#else
    res = ari_ws_session_create(NULL, ser, &args, session_id);
#endif
    if (res != 0) {
        ast_log(LOG_ERROR,
            "%s: Failed to create ARI ari_session\n", remote_addr);
    }
 
    ast_free(args.app_parse);
    ast_free(args.app);
    return res;
}

References args, ari_ws_session_create(), ast_ari_validate_message_fn(), ast_calloc, ast_free, ast_http_error(), ast_json_dump_string, ast_json_free(), ast_log, ast_sockaddr_stringify(), LOG_ERROR, NULL, parse_app_args(), RAII_VAR, and ast_tcptls_session_instance::remote_address.

Referenced by ari_websocket_load_module().

websocket_established_cb()

static void websocket_established_cb ( struct ast_websocket *  ast_ws_session, struct ast_variable *  get_params, struct ast_variable *  upgrade_headers  )

static

Definition at line 638 of file ari_websockets.c.

{
    RAII_VAR(struct ast_ari_response *, response, NULL, ast_free);
    /*
     * ast_ws_session is passed in with it's refcount bumped so
     * we need to unref it when we're done.  The refcount will
     * be bumped again when we add it to the ari_ws_session.
     */
    RAII_VAR(struct ast_websocket *, s, ast_ws_session, ast_websocket_unref);
    RAII_VAR(struct ari_ws_session *, ari_ws_session, NULL, ari_ws_session_cleanup);
    struct ast_json *msg;
    struct ast_variable *v;
    char *remote_addr = ast_sockaddr_stringify(
        ast_websocket_remote_address(ast_ws_session));
    const char *session_id = ast_websocket_session_id(ast_ws_session);
 
    SCOPE_ENTER(2, "%s: WebSocket established\n", remote_addr);
 
    if (TRACE_ATLEAST(2)) {
        ast_trace(2, "%s: Websocket Upgrade Headers:\n", remote_addr);
        for (v = upgrade_headers; v; v = v->next) {
            ast_trace(3, "--> %s: %s\n", v->name, v->value);
        }
    }
 
    response = ast_calloc(1, sizeof(*response));
    if (!response) {
        SCOPE_EXIT_LOG_RTN(LOG_ERROR,
            "%s: Failed to create response\n", remote_addr);
    }
 
    /* Find the event_session and update its websocket  */
    ari_ws_session = ao2_find(ari_ws_session_registry, session_id, OBJ_SEARCH_KEY);
    if (ari_ws_session) {
        ao2_unlink(ari_ws_session_registry, ari_ws_session);
        ari_ws_session_update(ari_ws_session, ast_ws_session);
    } else {
         SCOPE_EXIT_LOG_RTN(LOG_ERROR,
            "%s: Failed to locate an event session for the websocket session\n",
            remote_addr);
    }
 
    ast_trace(-1, "%s: Waiting for messages\n", remote_addr);
    while ((msg = ari_ws_session_read(ari_ws_session))) {
        ari_websocket_process_request(ari_ws_session, remote_addr,
            upgrade_headers, ari_ws_session->app_name, msg);
        ast_json_unref(msg);
    }
 
    SCOPE_EXIT("%s: Websocket closed\n", remote_addr);
}

References ao2_find, ao2_unlink, ari_ws_session::app_name, ari_websocket_process_request(), ari_ws_session_cleanup(), ari_ws_session_read(), ari_ws_session_registry, ari_ws_session_update(), ast_calloc, ast_free, ast_json_unref(), ast_sockaddr_stringify(), ast_trace, ast_websocket_remote_address(), ast_websocket_session_id(), ast_websocket_unref(), LOG_ERROR, ast_variable::name, ast_variable::next, NULL, OBJ_SEARCH_KEY, RAII_VAR, SCOPE_ENTER, SCOPE_EXIT, SCOPE_EXIT_LOG_RTN, TRACE_ATLEAST, and ast_variable::value.

Referenced by ari_websocket_load_module().

Variable Documentation

ari_ws_session_registry

struct ao2_container* ari_ws_session_registry

static

Local registry for created event_session objects.

Definition at line 52 of file ari_websockets.c.

Referenced by ari_websocket_load_module(), ari_ws_session_cleanup(), ari_ws_session_create(), ari_ws_session_registry_dtor(), and websocket_established_cb().

ast_ws_server

struct ast_websocket_server* ast_ws_server

Definition at line 54 of file ari_websockets.c.

Referenced by ari_handle_websocket(), ari_websocket_load_module(), and ari_websocket_unload_module().