ari_websockets.h File Reference

Internal API's for websockets. More...

#include "asterisk/http.h"
#include "asterisk/json.h"
#include "asterisk/vector.h"
#include "asterisk/websocket_client.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

Macros
#define	ARI_MAX_APP_NAME_LEN   (AST_MAX_CONTEXT - STASIS_CONTEXT_PREFIX_LEN)
#define	STASIS_CONTEXT_PREFIX   "stasis-"
#define	STASIS_CONTEXT_PREFIX_LEN   (sizeof(STASIS_CONTEXT_PREFIX) - 1)

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_outbound_websocket_start (struct ari_conf_outbound_websocket *owc)
struct ari_ws_session *	ari_websocket_get_session (const char *session_id)
struct ari_ws_session *	ari_websocket_get_session_by_app (const char *app_name)
struct ao2_container *	ari_websocket_get_sessions (void)
int	ari_websocket_load_module (int is_enabled)
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...
void	ari_websocket_shutdown (struct ari_ws_session *session)
void	ari_websocket_shutdown_all (void)
const char *	ari_websocket_type_to_str (enum ast_websocket_type type)
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.

Macro Definition Documentation

ARI_MAX_APP_NAME_LEN

#define ARI_MAX_APP_NAME_LEN   (AST_MAX_CONTEXT - STASIS_CONTEXT_PREFIX_LEN)

Definition at line 48 of file ari_websockets.h.

STASIS_CONTEXT_PREFIX

#define STASIS_CONTEXT_PREFIX   "stasis-"

Definition at line 46 of file ari_websockets.h.

STASIS_CONTEXT_PREFIX_LEN

#define STASIS_CONTEXT_PREFIX_LEN   (sizeof(STASIS_CONTEXT_PREFIX) - 1)

Definition at line 47 of 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 267 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_outbound_websocket_start()

int ari_outbound_websocket_start

(

struct ari_conf_outbound_websocket *

owc

)

Definition at line 1479 of file ari_websockets.c.

{
    if (owc) {
        return outbound_session_create(owc, NULL, 0);
    }
    return -1;
}

References NULL, and outbound_session_create().

Referenced by ari_start_owc().

ari_websocket_get_session()

struct ari_ws_session * ari_websocket_get_session

(

const char *

session_id

)

Definition at line 887 of file ari_websockets.c.

{
    return ao2_find(session_registry, session_id, OBJ_SEARCH_KEY);
}

References ao2_find, OBJ_SEARCH_KEY, ari_ws_session::session_id, and session_registry.

Referenced by ari_shut_session(), and outbound_session_create().

ari_websocket_get_session_by_app()

struct ari_ws_session * ari_websocket_get_session_by_app

(

const char *

app_name

)

ari_websocket_get_sessions()

struct ao2_container * ari_websocket_get_sessions

(

void

)

Definition at line 1360 of file ari_websockets.c.

{
    return ao2_bump(session_registry);
}

References ao2_bump, and session_registry.

Referenced by ari_show_sessions(), and ari_shut_session().

ari_websocket_load_module()

int ari_websocket_load_module

(

int

is_enabled

)

Definition at line 1528 of file ari_websockets.c.

{
    int res = 0;
    struct ast_websocket_protocol *protocol;
 
    ast_debug(2, "Initializing ARI websockets.  Enabled: %s\n", is_enabled ? "yes" : "no");
 
    session_registry = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_MUTEX,
        AO2_CONTAINER_ALLOC_OPT_DUPS_REPLACE,
        ari_ws_session_sort_fn, ari_ws_session_cmp_fn);
    if (!session_registry) {
        ast_log(LOG_WARNING,
                "Failed to allocate the local registry for websocket applications\n");
        return AST_MODULE_LOAD_DECLINE;
    }
 
    res = ari_sorcery_observer_add("outbound_websocket", &observer_callbacks);
    if (res < 0) {
        ast_log(LOG_WARNING, "Failed to register ARI websocket observer\n");
        ari_websocket_unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    /*
     * The global "enabled" flag only controls whether the REST and
     * inbound websockets are enabled.  The outbound websocket
     * configs are always enabled.
    if (!is_enabled) {
        return AST_MODULE_LOAD_SUCCESS;
    }
     */
 
    ast_ws_server = ast_websocket_server_create();
    if (!ast_ws_server) {
        ari_websocket_unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    protocol = ast_websocket_sub_protocol_alloc("ari");
    if (!protocol) {
        ari_websocket_unload_module();
        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_OPT_DUPS_REPLACE, ao2_container_alloc_rbtree, ari_sorcery_observer_add(), ari_websocket_unload_module(), ast_debug, 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, is_enabled(), LOG_WARNING, observer_callbacks, ast_websocket_protocol::session_attempted, ast_websocket_protocol::session_established, session_registry, 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 outbound_session_handler_thread(), and 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 290 of file ari_websockets.c.

{
    char *remote_addr = session->ast_ws_session ? ast_sockaddr_stringify(
            ast_websocket_remote_address(session->ast_ws_session)) : "";
    const char *msg_type, *msg_application;
    SCOPE_ENTER(4, "%s: Dispatching message from Stasis app '%s'\n", remote_addr, app_name);
 
    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")), app_name);
 
    /* If we've been replaced, remove the application from our local
       websocket_apps container */
    if (session->type == AST_WS_TYPE_INBOUND
        && strcmp(msg_type, "ApplicationReplaced") == 0 &&
        strcmp(msg_application, app_name) == 0) {
        AST_VECTOR_REMOVE_CMP_ORDERED(&session->websocket_apps,
            app_name, ast_strings_equal, ast_free_ptr);
    }
 
    /* 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(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 (stasis_app_event_allowed(app_name, message)) {
        session_send_or_queue(session, message, msg_type,
            app_name, debug_app);
    }
 
    if (session->type == AST_WS_TYPE_CLIENT_PER_CALL
        && !ast_strlen_zero(session->channel_id)
        && ast_strings_equal(msg_type, "StasisEnd")) {
        struct ast_json *chan = ast_json_object_get(message, "channel");
        struct ast_json *id_obj = ast_json_object_get(chan, "id");
        const char *id = ast_json_string_get(id_obj);
        if (!ast_strlen_zero(id)
            && ast_strings_equal(id, session->channel_id)) {
            ast_debug(3, "%s: StasisEnd message sent for channel '%s'\n",
                remote_addr, id);
            session->stasis_end_sent = 1;
        }
    }
    ao2_unlock(session);
    SCOPE_EXIT("%s: Dispatched '%s' message from Stasis app '%s'\n",
        remote_addr, msg_type, app_name);
}

References ao2_lock, ao2_unlock, app_name(), ast_assert, ast_debug, ast_free_ptr(), ast_json_object_get(), ast_json_object_set(), ast_json_string_create(), ast_json_string_get(), ast_sockaddr_stringify(), ast_strings_equal(), ast_strlen_zero(), AST_VECTOR_REMOVE_CMP_ORDERED, ast_websocket_remote_address(), AST_WS_TYPE_CLIENT_PER_CALL, AST_WS_TYPE_INBOUND, LOG_WARNING, NULL, S_OR, SCOPE_ENTER, SCOPE_EXIT, SCOPE_EXIT_LOG_RTN, session, session_send_or_queue(), and stasis_app_event_allowed().

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

ari_websocket_shutdown()

void ari_websocket_shutdown

(

struct ari_ws_session *

session

)

Definition at line 1487 of file ari_websockets.c.

{
    if (session) {
        session_shutdown_cb(session, NULL, 0);
    }
}

References NULL, session, and session_shutdown_cb().

Referenced by ari_shut_session().

ari_websocket_shutdown_all()

void ari_websocket_shutdown_all

(

void

)

Definition at line 1494 of file ari_websockets.c.

{
    if (session_registry) {
        ao2_callback(session_registry, OBJ_MULTIPLE | OBJ_NODATA,
            session_shutdown_cb, NULL);
    }
}

References ao2_callback, NULL, OBJ_MULTIPLE, OBJ_NODATA, session_registry, and session_shutdown_cb().

Referenced by ari_shut_sessions().

ari_websocket_type_to_str()

const char * ari_websocket_type_to_str

(

enum ast_websocket_type

type

)

Definition at line 359 of file res/ari/config.c.

{
    switch (type) {
    case AST_WS_TYPE_CLIENT_PERSISTENT:
        return "persistent";
    case AST_WS_TYPE_CLIENT_PER_CALL:
        return "per_call";
    case AST_WS_TYPE_CLIENT_PER_CALL_CONFIG:
        return "per_call_config";
    case AST_WS_TYPE_INBOUND:
        return "inbound";
    case AST_WS_TYPE_ANY:
        return "any";
    default:
        return "unknown";
    }
}

References AST_WS_TYPE_ANY, AST_WS_TYPE_CLIENT_PER_CALL, AST_WS_TYPE_CLIENT_PER_CALL_CONFIG, AST_WS_TYPE_CLIENT_PERSISTENT, AST_WS_TYPE_INBOUND, and type.

Referenced by ari_show_owc(), session_shutdown_cb(), show_owc_cb(), and show_sessions_cb().

ari_websocket_unload_module()

int ari_websocket_unload_module

(

void

)

Definition at line 1516 of file ari_websockets.c.

{
    ari_sorcery_observer_remove("outbound_websocket", &observer_callbacks);
    session_registry_dtor();
    ao2_cleanup(ast_ws_server);
    ast_ws_server = NULL;
    return 0;
}

References ao2_cleanup, ari_sorcery_observer_remove(), ast_ws_server, NULL, observer_callbacks, and session_registry_dtor().

Referenced by ari_websocket_load_module(), and unload_module().