ari_websockets.h File Reference

Internal API's for websockets. More...

#include "asterisk/http.h"
#include "asterisk/json.h"
#include "asterisk/vector.h"

Include dependency graph for ari_websockets.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ari_ws_session

Functions
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)
int	ari_websocket_process_request (struct ari_ws_session *ast_ws_session, const char *remote_addr, struct ast_variable *upgrade_headers, const char *app_name, struct ast_json *msg)
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)

Detailed Description

Internal API's for websockets.

Author

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

Definition in file ari_websockets.h.

Function Documentation

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_process_request()

int ari_websocket_process_request	(	struct ari_ws_session *	ast_ws_session,
		const char *	remote_addr,
		struct ast_variable *	upgrade_headers,
		const char *	app_name,
		struct ast_json *	msg
	)

Definition at line 261 of file ari_websocket_requests.c.

{
    int debug_app = stasis_app_get_debug_by_name(app_name);
    RAII_VAR(struct rest_request_msg *, request, NULL, request_destroy);
    struct ast_ari_response response = { .fd = -1, 0 };
 
    SCOPE_ENTER(3, "%s: New WebSocket Msg\n", remote_addr);
 
    if (TRACE_ATLEAST(3) || debug_app) {
        char *str = ast_json_dump_string_format(request_msg, AST_JSON_PRETTY);
        /* If we can't allocate a string, we can't respond to the client either. */
        if (!str) {
            SCOPE_EXIT_LOG_RTN_VALUE(-1, LOG_ERROR, "%s: Failed to dump JSON request\n",
                remote_addr);
        }
        ast_verbose("<--- Received ARI message from %s --->\n%s\n",
            remote_addr, str);
        ast_json_free(str);
    }
 
    request = SCOPE_CALL_WITH_RESULT(-1, struct rest_request_msg *,
        parse_rest_request_msg, remote_addr, request_msg, &response, debug_app);
 
    if (!request || response.response_code != 200) {
        SCOPE_CALL(-1, send_rest_response, ari_ws_session,
            remote_addr, app_name, request, &response, debug_app);
        SCOPE_EXIT_RTN_VALUE(0, "%s: Done with message\n", remote_addr);
    }
 
    /*
     * We don't actually use the headers in the response
     * but we have to allocate it because ast_ari_invoke
     * and the resource handlers expect it.
     */
    response.headers = ast_str_create(80);
    if (!response.headers) {
        /* If we can't allocate a string, we can't respond to the client either. */
        SCOPE_EXIT_LOG_RTN_VALUE(-1, LOG_ERROR, "%s: Failed allocate headers string\n",
            remote_addr);
    }
 
    SCOPE_CALL(-1, ast_ari_invoke, NULL, ARI_INVOKE_SOURCE_WEBSOCKET,
        NULL, request->uri, request->method, request->query_strings,
        upgrade_headers, request->body, &response);
 
    ast_free(response.headers);
 
    if (response.no_response) {
        SCOPE_EXIT_RTN_VALUE(0, "No response needed\n");
    }
 
    SCOPE_CALL(-1, send_rest_response, ari_ws_session,
        remote_addr, app_name, request, &response, debug_app);
 
    SCOPE_EXIT_RTN_VALUE(0, "%s: Done with message\n", remote_addr);
}

References app_name(), ARI_INVOKE_SOURCE_WEBSOCKET, ast_ari_invoke(), ast_free, ast_json_dump_string_format(), ast_json_free(), AST_JSON_PRETTY, ast_str_create, ast_verbose(), ast_ari_response::fd, ast_ari_response::headers, LOG_ERROR, ast_ari_response::no_response, NULL, parse_rest_request_msg(), RAII_VAR, request(), request_destroy(), ast_ari_response::response_code, SCOPE_CALL, SCOPE_CALL_WITH_RESULT, SCOPE_ENTER, SCOPE_EXIT_LOG_RTN_VALUE, SCOPE_EXIT_RTN_VALUE, send_rest_response(), stasis_app_get_debug_by_name(), str, and TRACE_ATLEAST.

Referenced by websocket_established_cb().

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().