chan_websocket.c File Reference

Websocket Media Channel. More...

#include "asterisk.h"
#include "asterisk/app.h"
#include "asterisk/causes.h"
#include "asterisk/channel.h"
#include "asterisk/codec.h"
#include "asterisk/http_websocket.h"
#include "asterisk/format_cache.h"
#include "asterisk/frame.h"
#include "asterisk/lock.h"
#include "asterisk/mod_format.h"
#include "asterisk/module.h"
#include "asterisk/pbx.h"
#include "asterisk/uuid.h"
#include "asterisk/timing.h"
#include "asterisk/translate.h"
#include "asterisk/websocket_client.h"

Include dependency graph for chan_websocket.c:

Go to the source code of this file.

Data Structures
struct	instance_proxy
struct	websocket_pvt

Macros
#define	ANSWER_CHANNEL   "ANSWER"
#define	CONTINUE_MEDIA   "CONTINUE_MEDIA"
#define	DRIVER_STATUS   "STATUS"
#define	DTMF_END   "DTMF_END"
#define	FLUSH_MEDIA   "FLUSH_MEDIA"
#define	GET_DRIVER_STATUS   "GET_STATUS"
#define	HANGUP_CHANNEL   "HANGUP"
#define	INCOMING_CONNECTION_ID   "INCOMING"
#define	MAX_TEXT_MESSAGE_LEN   MIN(128, (AST_WEBSOCKET_MAX_RX_PAYLOAD_SIZE - 1))
#define	MEDIA_BUFFERING_COMPLETED   "MEDIA_BUFFERING_COMPLETED"
#define	MEDIA_START   "MEDIA_START"
#define	MEDIA_WEBSOCKET_CONNECTION_ID   "MEDIA_WEBSOCKET_CONNECTION_ID"
#define	MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE   "MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE"
#define	MEDIA_XOFF   "MEDIA_XOFF"
#define	MEDIA_XON   "MEDIA_XON"
#define	PAUSE_MEDIA   "PAUSE_MEDIA"
#define	QUEUE_DRAINED   "QUEUE_DRAINED"
#define	QUEUE_LENGTH_MAX   1000
#define	QUEUE_LENGTH_XOFF_LEVEL   900
#define	QUEUE_LENGTH_XON_LEVEL   800
#define	REPORT_QUEUE_DRAINED   "REPORT_QUEUE_DRAINED"
#define	START_MEDIA_BUFFERING   "START_MEDIA_BUFFERING"
#define	STOP_MEDIA_BUFFERING   "STOP_MEDIA_BUFFERING"

Enumerations
enum	{ OPT_WS_CODEC = (1 << 0) , OPT_WS_NO_AUTO_ANSWER = (1 << 1) , OPT_WS_URI_PARAM = (1 << 2) , OPT_WS_PASSTHROUGH = (1 << 3) }
enum	{   OPT_ARG_WS_CODEC , OPT_ARG_WS_NO_AUTO_ANSWER , OPT_ARG_WS_URI_PARAM , OPT_ARG_WS_PASSTHROUGH ,   OPT_ARG_ARRAY_SIZE }

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
struct ast_module *	AST_MODULE_SELF_SYM (void)
static struct ast_frame *	dequeue_frame (struct websocket_pvt *instance)
static void	incoming_ws_established_cb (struct ast_websocket *ast_ws_session, struct ast_variable *get_params, struct ast_variable *upgrade_headers)
static int	incoming_ws_http_callback (struct ast_tcptls_session_instance *ser, const struct ast_http_uri *urih, const char *uri, enum ast_http_method method, struct ast_variable *get_params, struct ast_variable *headers)
static void	instance_proxy_cb (void *weakproxy, void *data)
static int	load_module (void)
	Function called when our module is loaded. More...
static int	process_binary_message (struct websocket_pvt *instance, char *payload, uint64_t payload_len)
static int	process_text_message (struct websocket_pvt *instance, char *payload, uint64_t payload_len)
static int	queue_frame_from_buffer (struct websocket_pvt *instance, char *buffer, size_t len)
static int	queue_option_frame (struct websocket_pvt *instance, char *buffer)
static int	read_from_ws_and_queue (struct websocket_pvt *instance)
static void *	read_thread_handler (void *obj)
static void	set_channel_format (struct websocket_pvt *instance, struct ast_format *fmt)
static int	set_channel_timer (struct websocket_pvt *instance)
static int	set_channel_variables (struct websocket_pvt *instance)
static int	set_instance_silence_frame (struct websocket_pvt *instance)
static int	set_instance_translator (struct websocket_pvt *instance)
static int	unload_module (void)
	Function called when our module is unloaded. More...
static int	validate_uri_parameters (const char *uri_params)
static int	webchan_call (struct ast_channel *ast, const char *dest, int timeout)
static int	webchan_hangup (struct ast_channel *ast)
static struct ast_frame *	webchan_read (struct ast_channel *ast)
static struct ast_channel *	webchan_request (const char *type, struct ast_format_cap *cap, const struct ast_assigned_ids *assignedids, const struct ast_channel *requestor, const char *data, int *cause)
static int	webchan_send_dtmf_text (struct ast_channel *ast, char digit, unsigned int duration)
static int	webchan_write (struct ast_channel *ast, struct ast_frame *f)
	Function called when we should write a frame to the channel. More...
static void	websocket_destructor (void *data)
static struct websocket_pvt *	websocket_new (const char *chan_name, const char *connection_id, struct ast_format *fmt)

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "Websocket Media Channel" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_CHANNEL_DRIVER, .requires = "res_http_websocket,res_websocket_client", }
static const struct ast_module_info *	ast_module_info = &__mod_info
static struct ast_websocket_server *	ast_ws_server
static struct ast_http_uri	http_uri
static struct ao2_container *	instances = NULL
static const struct ast_app_option	websocket_options [128] = { [ 'c' ] = { .flag = OPT_WS_CODEC , .arg_index = OPT_ARG_WS_CODEC + 1 }, [ 'n' ] = { .flag = OPT_WS_NO_AUTO_ANSWER }, [ 'v' ] = { .flag = OPT_WS_URI_PARAM , .arg_index = OPT_ARG_WS_URI_PARAM + 1 }, [ 'p' ] = { .flag = OPT_WS_PASSTHROUGH }, }
static struct ast_channel_tech	websocket_tech

Detailed Description

Websocket Media Channel.

Author

George Joseph gjose.nosp@m.ph@s.nosp@m.angom.nosp@m.a.co.nosp@m.m

Definition in file chan_websocket.c.

Macro Definition Documentation

ANSWER_CHANNEL

#define ANSWER_CHANNEL   "ANSWER"

Definition at line 159 of file chan_websocket.c.

CONTINUE_MEDIA

#define CONTINUE_MEDIA   "CONTINUE_MEDIA"

Definition at line 167 of file chan_websocket.c.

DRIVER_STATUS

#define DRIVER_STATUS   "STATUS"

Definition at line 173 of file chan_websocket.c.

DTMF_END

#define DTMF_END   "DTMF_END"

Definition at line 175 of file chan_websocket.c.

FLUSH_MEDIA

#define FLUSH_MEDIA   "FLUSH_MEDIA"

Definition at line 163 of file chan_websocket.c.

GET_DRIVER_STATUS

#define GET_DRIVER_STATUS   "GET_STATUS"

Definition at line 164 of file chan_websocket.c.

HANGUP_CHANNEL

#define HANGUP_CHANNEL   "HANGUP"

Definition at line 160 of file chan_websocket.c.

INCOMING_CONNECTION_ID

#define INCOMING_CONNECTION_ID   "INCOMING"

Definition at line 157 of file chan_websocket.c.

MAX_TEXT_MESSAGE_LEN

#define MAX_TEXT_MESSAGE_LEN   MIN(128, (AST_WEBSOCKET_MAX_RX_PAYLOAD_SIZE - 1))

Definition at line 180 of file chan_websocket.c.

MEDIA_BUFFERING_COMPLETED

#define MEDIA_BUFFERING_COMPLETED   "MEDIA_BUFFERING_COMPLETED"

Definition at line 174 of file chan_websocket.c.

MEDIA_START

#define MEDIA_START   "MEDIA_START"

Definition at line 169 of file chan_websocket.c.

MEDIA_WEBSOCKET_CONNECTION_ID

#define MEDIA_WEBSOCKET_CONNECTION_ID   "MEDIA_WEBSOCKET_CONNECTION_ID"

Definition at line 156 of file chan_websocket.c.

MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE

#define MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE   "MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE"

Definition at line 155 of file chan_websocket.c.

MEDIA_XOFF

#define MEDIA_XOFF   "MEDIA_XOFF"

Definition at line 171 of file chan_websocket.c.

MEDIA_XON

#define MEDIA_XON   "MEDIA_XON"

Definition at line 170 of file chan_websocket.c.

PAUSE_MEDIA

#define PAUSE_MEDIA   "PAUSE_MEDIA"

Definition at line 166 of file chan_websocket.c.

QUEUE_DRAINED

#define QUEUE_DRAINED   "QUEUE_DRAINED"

Definition at line 172 of file chan_websocket.c.

QUEUE_LENGTH_MAX

#define QUEUE_LENGTH_MAX   1000

Definition at line 177 of file chan_websocket.c.

QUEUE_LENGTH_XOFF_LEVEL

#define QUEUE_LENGTH_XOFF_LEVEL   900

Definition at line 178 of file chan_websocket.c.

QUEUE_LENGTH_XON_LEVEL

#define QUEUE_LENGTH_XON_LEVEL   800

Definition at line 179 of file chan_websocket.c.

REPORT_QUEUE_DRAINED

#define REPORT_QUEUE_DRAINED   "REPORT_QUEUE_DRAINED"

Definition at line 165 of file chan_websocket.c.

START_MEDIA_BUFFERING

#define START_MEDIA_BUFFERING   "START_MEDIA_BUFFERING"

Definition at line 161 of file chan_websocket.c.

STOP_MEDIA_BUFFERING

#define STOP_MEDIA_BUFFERING   "STOP_MEDIA_BUFFERING"

Definition at line 162 of file chan_websocket.c.

Enumeration Type Documentation

anonymous enum

anonymous enum

Enumerator
OPT_WS_CODEC
OPT_WS_NO_AUTO_ANSWER
OPT_WS_URI_PARAM
OPT_WS_PASSTHROUGH

Definition at line 1273 of file chan_websocket.c.

     {
    OPT_WS_CODEC =  (1 << 0),
    OPT_WS_NO_AUTO_ANSWER =  (1 << 1),
    OPT_WS_URI_PARAM =  (1 << 2),
    OPT_WS_PASSTHROUGH =  (1 << 3),
};

anonymous enum

anonymous enum

Enumerator
OPT_ARG_WS_CODEC
OPT_ARG_WS_NO_AUTO_ANSWER
OPT_ARG_WS_URI_PARAM
OPT_ARG_WS_PASSTHROUGH
OPT_ARG_ARRAY_SIZE

Definition at line 1280 of file chan_websocket.c.

     {
    OPT_ARG_WS_CODEC,
    OPT_ARG_WS_NO_AUTO_ANSWER,
    OPT_ARG_WS_URI_PARAM,
    OPT_ARG_WS_PASSTHROUGH,
    OPT_ARG_ARRAY_SIZE
};

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 1761 of file chan_websocket.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 1761 of file chan_websocket.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 1761 of file chan_websocket.c.

dequeue_frame()

static struct ast_frame * dequeue_frame ( struct websocket_pvt *  instance )

static

Definition at line 218 of file chan_websocket.c.

{
    struct ast_frame *queued_frame = NULL;
    SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
        AST_LIST_UNLOCK);
 
    /*
     * If the queue is paused, don't read a frame.  Processing
     * will continue down the function and a silence frame will
     * be sent in its place.
     */
    if (instance->queue_paused) {
        return NULL;
    }
 
    /*
     * We need to check if we need to send an XON before anything
     * else because there are multiple escape paths in this function
     * and we don't want to accidentally keep the queue in a "full"
     * state.
     */
    if (instance->queue_full && instance->frame_queue_length < QUEUE_LENGTH_XON_LEVEL) {
        instance->queue_full = 0;
        ast_debug(4, "%s: WebSocket sending MEDIA_XON\n",
            ast_channel_name(instance->channel));
        ast_websocket_write_string(instance->websocket, MEDIA_XON);
    }
 
    queued_frame = AST_LIST_REMOVE_HEAD(&instance->frame_queue, frame_list);
 
    /*
     * If there are no frames in the queue, we need to
     * return NULL so we can send a silence frame.  We also need
     * to send the QUEUE_DRAINED notification if we were requested
     * to do so.
     */
    if (!queued_frame) {
        if (instance->report_queue_drained) {
            instance->report_queue_drained = 0;
            ast_debug(4, "%s: WebSocket sending QUEUE_DRAINED\n",
                ast_channel_name(instance->channel));
            ast_websocket_write_string(instance->websocket, QUEUE_DRAINED);
        }
        return NULL;
    }
 
    /*
     * The only way a control frame could be present here is as
     * a result of us calling queue_option_frame() in response
     * to an incoming TEXT command from the websocket.
     * We'll be safe and make sure it's a AST_CONTROL_OPTION
     * frame anyway.
     *
     * It's quite possible that there are multiple control frames
     * in a row in the queue so we need to process consecutive ones
     * immediately.
     *
     * In any case, processing a control frame MUST not use up
     * a media timeslot so after all control frames have been
     * processed, we need to read an audio frame and process it.
     */
    while (queued_frame && queued_frame->frametype == AST_FRAME_CONTROL) {
        if (queued_frame->subclass.integer == AST_CONTROL_OPTION) {
            /*
             * We just need to send the data to the websocket.
             * The data should already be NULL terminated.
             */
            ast_websocket_write_string(instance->websocket,
                queued_frame->data.ptr);
            ast_debug(4, "%s: WebSocket sending %s\n",
                ast_channel_name(instance->channel), (char *)queued_frame->data.ptr);
        }
        /*
         * We do NOT send these to the core so we need to free
         * the frame and grab the next one.  If it's also a
         * control frame, we need to process it otherwise
         * continue down in the function.
         */
        ast_frame_free(queued_frame, 0);
        queued_frame = AST_LIST_REMOVE_HEAD(&instance->frame_queue, frame_list);
        /*
         * Jut FYI... We didn't bump the queue length when we added the control
         * frames so we don't need to decrement it here.
         */
    }
 
    /*
     * If, after reading all control frames,  there are no frames
     * left in the queue, we need to return NULL so we can send
     * a silence frame.
     */
    if (!queued_frame) {
        return NULL;
    }
 
    instance->frame_queue_length--;
 
    return queued_frame;
}

References ast_channel_name(), AST_CONTROL_OPTION, ast_debug, AST_FRAME_CONTROL, ast_frame_free(), AST_LIST_LOCK, AST_LIST_REMOVE_HEAD, AST_LIST_UNLOCK, ast_websocket_write_string(), websocket_pvt::channel, ast_frame::data, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, ast_frame::frametype, ast_frame_subclass::integer, MEDIA_XON, NULL, ast_frame::ptr, QUEUE_DRAINED, websocket_pvt::queue_full, QUEUE_LENGTH_XON_LEVEL, websocket_pvt::queue_paused, websocket_pvt::report_queue_drained, SCOPED_LOCK, ast_frame::subclass, and websocket_pvt::websocket.

Referenced by webchan_read().

incoming_ws_established_cb()

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

static

Definition at line 1536 of file chan_websocket.c.

{
    RAII_VAR(struct ast_websocket *, s, ast_ws_session, ast_websocket_unref);
    struct ast_variable *v;
    const char *connection_id = NULL;
    struct websocket_pvt *instance = NULL;
    int nodelay = 1;
 
    ast_debug(3, "WebSocket established\n");
 
    for (v = upgrade_headers; v; v = v->next) {
        ast_debug(4, "Header-> %s: %s\n", v->name, v->value);
    }
    for (v = get_params; v; v = v->next) {
        ast_debug(4, " Param-> %s: %s\n", v->name, v->value);
    }
 
    connection_id = ast_variable_find_in_list(get_params, "CONNECTION_ID");
    if (!connection_id) {
        /*
         * This can't really happen because websocket_http_callback won't
         * let it get this far if it can't add the connection_id to the
         * get_params.
         * Just in case though...
         */
        ast_log(LOG_WARNING, "WebSocket connection id not found\n");
        ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
        ast_websocket_close(ast_ws_session, 1000);
        return;
    }
 
    instance = ao2_weakproxy_find(instances, connection_id, OBJ_SEARCH_KEY | OBJ_NOLOCK, "");
    if (!instance) {
        /*
         * This also can't really happen because websocket_http_callback won't
         * let it get this far if it can't find the instance.
         * Just in case though...
         */
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n", connection_id);
        ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
        ast_websocket_close(ast_ws_session, 1000);
        return;
    }
    instance->websocket = ao2_bump(ast_ws_session);
 
    if (setsockopt(ast_websocket_fd(instance->websocket),
        IPPROTO_TCP, TCP_NODELAY, (char *) &nodelay, sizeof(nodelay)) < 0) {
        ast_log(LOG_WARNING, "Failed to set TCP_NODELAY on manager connection: %s\n", strerror(errno));
    }
 
    /* read_thread_handler cleans up the bump */
    read_thread_handler(ao2_bump(instance));
 
    ao2_cleanup(instance);
    ast_debug(3, "WebSocket closed\n");
}

References ao2_bump, ao2_cleanup, ao2_weakproxy_find, AST_CONTROL_HANGUP, ast_debug, ast_log, ast_queue_control(), ast_variable_find_in_list(), ast_websocket_close(), ast_websocket_fd(), ast_websocket_unref(), websocket_pvt::channel, websocket_pvt::connection_id, errno, instances, LOG_WARNING, ast_variable::name, ast_variable::next, NULL, OBJ_NOLOCK, OBJ_SEARCH_KEY, RAII_VAR, read_thread_handler(), ast_variable::value, and websocket_pvt::websocket.

Referenced by load_module().

incoming_ws_http_callback()

static int incoming_ws_http_callback ( struct ast_tcptls_session_instance *  ser, const struct ast_http_uri *  urih, const char *  uri, enum ast_http_method  method, struct ast_variable *  get_params, struct ast_variable *  headers  )

static

Definition at line 1604 of file chan_websocket.c.

{
    struct ast_http_uri fake_urih = {
        .data = ast_ws_server,
    };
    int res = 0;
    /*
     * Normally the http server will destroy the get_params
     * when the session ends but if there weren't any initially
     * and we create some and add them to the list, the http server
     * won't know about it so we have to destroy it ourselves.
     */
    int destroy_get_params = (get_params == NULL);
    struct ast_variable *v = NULL;
    RAII_VAR(struct websocket_pvt *, instance, NULL, ao2_cleanup);
 
    ast_debug(2, "URI: %s Starting\n", uri);
 
    /*
     * The client will have issued the GET request with a URI of
     * /media/<connection_id>
     *
     * Since this callback is registered for the /media URI prefix the
     * http server will strip that off the front of the URI passing in
     * only the path components after that in the 'uri' parameter.
     * This should leave only the connection id without a leading '/'.
     */
    instance = ao2_weakproxy_find(instances, uri, OBJ_SEARCH_KEY | OBJ_NOLOCK, "");
    if (!instance) {
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n", uri);
        ast_http_error(ser, 404, "Not found", "WebSocket instance not found");
        return -1;
    }
 
    /*
     * We don't allow additional connections using the same connection id.
     */
    if (instance->websocket) {
        ast_log(LOG_WARNING, "%s: Websocket already connected for channel '%s'\n",
            uri, instance->channel ? ast_channel_name(instance->channel) : "unknown");
        ast_http_error(ser, 409, "Conflict", "Another websocket connection exists for this connection id");
        return -1;
    }
 
    v = ast_variable_new("CONNECTION_ID", uri, "");
    if (!v) {
        ast_http_error(ser, 500, "Server error", "");
        return -1;
    }
    ast_variable_list_append(&get_params, v);
 
    for (v = get_params; v; v = v->next) {
        ast_debug(4, " Param-> %s: %s\n", v->name, v->value);
    }
 
    /*
     * This will ultimately call internal_ws_established_cb() so
     * this function will block until the websocket is closed and
     * internal_ws_established_cb() returns;
     */
    res = ast_websocket_uri_cb(ser, &fake_urih, uri, method,
        get_params, headers);
    if (destroy_get_params) {
        ast_variables_destroy(get_params);
    }
 
    ast_debug(2, "URI: %s DONE\n", uri);
 
    return res;
}

References ao2_cleanup, ao2_weakproxy_find, ast_channel_name(), ast_debug, ast_http_error(), ast_log, ast_variable_list_append, ast_variable_new, ast_variables_destroy(), ast_websocket_uri_cb(), ast_ws_server, ast_http_uri::data, instances, LOG_WARNING, method, ast_variable::name, ast_variable::next, NULL, OBJ_NOLOCK, OBJ_SEARCH_KEY, RAII_VAR, and ast_variable::value.

instance_proxy_cb()

static void instance_proxy_cb ( void *  weakproxy, void *  data  )

static

Definition at line 1043 of file chan_websocket.c.

{
    struct instance_proxy *proxy = weakproxy;
    ast_debug(3, "%s: WebSocket instance removed from instances\n", proxy->connection_id);
    ao2_unlink(instances, weakproxy);
}

References ao2_unlink, ast_debug, instance_proxy::connection_id, and instances.

Referenced by websocket_new().

load_module()

static int load_module ( void  )

static

Function called when our module is loaded.

Definition at line 1710 of file chan_websocket.c.

{
    int res = 0;
    struct ast_websocket_protocol *protocol;
 
    if (!(websocket_tech.capabilities = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT))) {
        return AST_MODULE_LOAD_DECLINE;
    }
 
    ast_format_cap_append_by_type(websocket_tech.capabilities, AST_MEDIA_TYPE_UNKNOWN);
    if (ast_channel_register(&websocket_tech)) {
        ast_log(LOG_ERROR, "Unable to register channel class 'WebSocket'\n");
        unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    instances = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK,
        AO2_CONTAINER_ALLOC_OPT_DUPS_REPLACE, 17, instance_proxy_hash_fn,
        instance_proxy_sort_fn, instance_proxy_cmp_fn);
    if (!instances) {
        ast_log(LOG_WARNING,
            "Failed to allocate the chan_websocket instance registry\n");
        unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    ast_ws_server = ast_websocket_server_create();
    if (!ast_ws_server) {
        unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
 
    protocol = ast_websocket_sub_protocol_alloc("media");
    if (!protocol) {
        unload_module();
        return AST_MODULE_LOAD_DECLINE;
    }
    protocol->session_established = incoming_ws_established_cb;
    res = ast_websocket_server_add_protocol2(ast_ws_server, protocol);
 
    ast_http_uri_link(&http_uri);
 
    return res == 0 ? AST_MODULE_LOAD_SUCCESS : AST_MODULE_LOAD_DECLINE;
}

References AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_container_alloc_hash, AO2_CONTAINER_ALLOC_OPT_DUPS_REPLACE, ast_channel_register(), ast_format_cap_alloc, ast_format_cap_append_by_type(), AST_FORMAT_CAP_FLAG_DEFAULT, ast_http_uri_link(), ast_log, AST_MEDIA_TYPE_UNKNOWN, 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, ast_channel_tech::capabilities, http_uri, incoming_ws_established_cb(), instances, LOG_ERROR, LOG_WARNING, ast_websocket_protocol::session_established, unload_module(), and websocket_tech.

process_binary_message()

static int process_binary_message ( struct websocket_pvt *  instance, char *  payload, uint64_t  payload_len  )

static

Definition at line 640 of file chan_websocket.c.

{
    char *next_frame_ptr = NULL;
    size_t bytes_read = 0;
    int res = 0;
    size_t bytes_left = 0;
 
    {
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
        if (instance->frame_queue_length >= QUEUE_LENGTH_MAX) {
            ast_debug(4, "%s: WebSocket queue is full. Ignoring incoming binary message.\n",
                ast_channel_name(instance->channel));
            return 0;
        }
    }
 
    next_frame_ptr = payload;
    instance->bytes_read += payload_len;
 
    if (instance->passthrough) {
        res = queue_frame_from_buffer(instance, payload, payload_len);
        return res;
    }
 
    if (instance->bulk_media_in_progress && instance->leftover_len > 0) {
        /*
         * We have leftover data from a previous websocket message.
         * Try to make a complete frame by appending data from
         * the current message to the leftover data.
         */
        char *append_ptr = instance->leftover_data + instance->leftover_len;
        size_t bytes_needed_for_frame = instance->optimal_frame_size - instance->leftover_len;
        /*
         * It's possible that even the current message doesn't have enough
         * data to make a complete frame.
         */
        size_t bytes_avail_to_copy = MIN(bytes_needed_for_frame, payload_len);
 
        /*
         * Append whatever we can to the end of the leftover data
         * even if it's not enough to make a complete frame.
         */
        memcpy(append_ptr, payload, bytes_avail_to_copy);
 
        /*
         * If leftover data is still short, just return and wait for the
         * next websocket message.
         */
        if (bytes_avail_to_copy < bytes_needed_for_frame) {
            ast_debug(4, "%s: Leftover data %d bytes but only %d new bytes available of %d needed. Appending and waiting for next message.\n",
                ast_channel_name(instance->channel), (int)instance->leftover_len, (int)bytes_avail_to_copy, (int)bytes_needed_for_frame);
            instance->leftover_len += bytes_avail_to_copy;
            return 0;
        }
 
        res = queue_frame_from_buffer(instance, instance->leftover_data, instance->optimal_frame_size);
        if (res < 0) {
            return -1;
        }
 
        /*
         * We stole data from the current payload so decrement payload_len
         * and set the next frame pointer after the data in payload
         * we just copied.
         */
        payload_len -= bytes_avail_to_copy;
        next_frame_ptr = payload + bytes_avail_to_copy;
 
        ast_debug(5, "%s: --- BR: %4d  FQ: %4d  PL: %4d  LOL: %3d  P: %p  NFP: %p  OFF: %4d  NPL: %4d  BAC: %3d\n",
            ast_channel_name(instance->channel),
            instance->frame_queue_length,
            (int)instance->bytes_read,
            (int)(payload_len + bytes_avail_to_copy),
            (int)instance->leftover_len,
            payload,
            next_frame_ptr,
            (int)(next_frame_ptr - payload),
            (int)payload_len,
            (int)bytes_avail_to_copy
            );
 
 
        instance->leftover_len = 0;
    }
 
    if (!instance->bulk_media_in_progress && instance->leftover_len > 0) {
        instance->leftover_len = 0;
    }
 
    bytes_left = payload_len;
    while (bytes_read < payload_len && bytes_left >= instance->optimal_frame_size) {
        res = queue_frame_from_buffer(instance, next_frame_ptr,
            instance->optimal_frame_size);
        if (res < 0) {
            break;
        }
        bytes_read += instance->optimal_frame_size;
        next_frame_ptr += instance->optimal_frame_size;
        bytes_left -= instance->optimal_frame_size;
    }
 
    if (instance->bulk_media_in_progress && bytes_left > 0) {
        /*
         * We have a partial frame.  Save the leftover data.
         */
        ast_debug(5, "%s: +++ BR: %4d  FQ: %4d  PL: %4d  LOL: %3d  P: %p  NFP: %p  OFF: %4d  BL: %4d\n",
            ast_channel_name(instance->channel),
            (int)instance->bytes_read,
            instance->frame_queue_length,
            (int)payload_len,
            (int)instance->leftover_len,
            payload,
            next_frame_ptr,
            (int)(next_frame_ptr - payload),
            (int)bytes_left
            );
        memcpy(instance->leftover_data, next_frame_ptr, bytes_left);
        instance->leftover_len = bytes_left;
    }
 
    return 0;
}

References ast_channel_name(), ast_debug, AST_LIST_LOCK, AST_LIST_UNLOCK, websocket_pvt::bulk_media_in_progress, websocket_pvt::bytes_read, websocket_pvt::channel, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, websocket_pvt::leftover_data, websocket_pvt::leftover_len, MIN, NULL, websocket_pvt::optimal_frame_size, websocket_pvt::passthrough, queue_frame_from_buffer(), QUEUE_LENGTH_MAX, and SCOPED_LOCK.

Referenced by read_from_ws_and_queue().

process_text_message()

static int process_text_message ( struct websocket_pvt *  instance, char *  payload, uint64_t  payload_len  )

static

Definition at line 487 of file chan_websocket.c.

{
    int res = 0;
    char *command;
 
    if (payload_len > MAX_TEXT_MESSAGE_LEN) {
        ast_log(LOG_WARNING, "%s: WebSocket TEXT message of length %d exceeds maximum length of %d\n",
            ast_channel_name(instance->channel), (int)payload_len, MAX_TEXT_MESSAGE_LEN);
        return 0;
    }
 
    /*
     * Unfortunately, payload is not NULL terminated even when it's
     * a TEXT frame so we need to allocate a new buffer, copy
     * the data into it, and NULL terminate it.
     */
    command = ast_alloca(payload_len + 1);
    memcpy(command, payload, payload_len); /* Safe */
    command[payload_len] = '\0';
    command = ast_strip(command);
 
    ast_debug(4, "%s: WebSocket %s command received\n",
        ast_channel_name(instance->channel), command);
 
    if (ast_strings_equal(command, ANSWER_CHANNEL)) {
        ast_queue_control(instance->channel, AST_CONTROL_ANSWER);
 
    } else if (ast_strings_equal(command, HANGUP_CHANNEL)) {
        ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
 
    } else if (ast_strings_equal(command, START_MEDIA_BUFFERING)) {
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
        AST_LIST_LOCK(&instance->frame_queue);
        instance->bulk_media_in_progress = 1;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_begins_with(command, STOP_MEDIA_BUFFERING)) {
        char *id;
        char *option;
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
 
        id = ast_strip(command + strlen(STOP_MEDIA_BUFFERING));
 
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
 
        ast_debug(4, "%s: WebSocket %s '%s' with %d bytes in leftover_data.\n",
            ast_channel_name(instance->channel), STOP_MEDIA_BUFFERING, id,
            (int)instance->leftover_len);
 
        instance->bulk_media_in_progress = 0;
        if (instance->leftover_len > 0) {
            res = queue_frame_from_buffer(instance, instance->leftover_data, instance->leftover_len);
            if (res != 0) {
                return res;
            }
        }
        instance->leftover_len = 0;
        res = ast_asprintf(&option, "%s%s%s", MEDIA_BUFFERING_COMPLETED,
            S_COR(!ast_strlen_zero(id), " ", ""), S_OR(id, ""));
        if (res <= 0 || !option) {
            return res;
        }
        res = queue_option_frame(instance, option);
        ast_free(option);
 
    } else if (ast_strings_equal(command, FLUSH_MEDIA)) {
        struct ast_frame *frame = NULL;
 
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
 
        AST_LIST_LOCK(&instance->frame_queue);
        while ((frame = AST_LIST_REMOVE_HEAD(&instance->frame_queue, frame_list))) {
            ast_frfree(frame);
        }
        instance->frame_queue_length = 0;
        instance->bulk_media_in_progress = 0;
        instance->leftover_len = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(payload, REPORT_QUEUE_DRAINED)) {
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
 
        AST_LIST_LOCK(&instance->frame_queue);
        instance->report_queue_drained = 1;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, GET_DRIVER_STATUS)) {
        char *status = NULL;
 
        res = ast_asprintf(&status, "%s queue_length:%d xon_level:%d xoff_level:%d queue_full:%s bulk_media:%s media_paused:%s",
            DRIVER_STATUS,
            instance->frame_queue_length, QUEUE_LENGTH_XON_LEVEL,
            QUEUE_LENGTH_XOFF_LEVEL,
            S_COR(instance->queue_full, "true", "false"),
            S_COR(instance->bulk_media_in_progress, "true", "false"),
            S_COR(instance->queue_paused, "true", "false")
            );
        if (res <= 0 || !status) {
            ast_free(status);
            res = -1;
        } else {
            ast_debug(4, "%s: WebSocket status: %s\n",
                ast_channel_name(instance->channel), status);
            res = ast_websocket_write_string(instance->websocket, status);
            ast_free(status);
        }
 
    } else if (ast_strings_equal(payload, PAUSE_MEDIA)) {
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 1;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(payload, CONTINUE_MEDIA)) {
        if (instance->passthrough) {
            ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n",
                ast_channel_name(instance->channel), command);
            return 0;
        }
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else {
        ast_log(LOG_WARNING, "%s: WebSocket %s command unknown\n",
            ast_channel_name(instance->channel), command);
    }
 
    return res;
}

References ANSWER_CHANNEL, ast_alloca, ast_asprintf, ast_begins_with(), ast_channel_name(), AST_CONTROL_ANSWER, AST_CONTROL_HANGUP, ast_debug, ast_free, ast_frfree, AST_LIST_LOCK, AST_LIST_REMOVE_HEAD, AST_LIST_UNLOCK, ast_log, ast_queue_control(), ast_strings_equal(), ast_strip(), ast_strlen_zero(), ast_websocket_write_string(), websocket_pvt::bulk_media_in_progress, websocket_pvt::channel, CONTINUE_MEDIA, DRIVER_STATUS, FLUSH_MEDIA, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, GET_DRIVER_STATUS, HANGUP_CHANNEL, id, websocket_pvt::leftover_data, websocket_pvt::leftover_len, LOG_WARNING, MAX_TEXT_MESSAGE_LEN, MEDIA_BUFFERING_COMPLETED, NULL, websocket_pvt::passthrough, PAUSE_MEDIA, queue_frame_from_buffer(), websocket_pvt::queue_full, QUEUE_LENGTH_XOFF_LEVEL, QUEUE_LENGTH_XON_LEVEL, queue_option_frame(), websocket_pvt::queue_paused, websocket_pvt::report_queue_drained, REPORT_QUEUE_DRAINED, S_COR, S_OR, SCOPED_LOCK, START_MEDIA_BUFFERING, status, STOP_MEDIA_BUFFERING, and websocket_pvt::websocket.

Referenced by read_from_ws_and_queue().

queue_frame_from_buffer()

static int queue_frame_from_buffer ( struct websocket_pvt *  instance, char *  buffer, size_t  len  )

static

Definition at line 423 of file chan_websocket.c.

{
    struct ast_frame fr = { 0, };
    struct ast_frame *duped_frame = NULL;
 
    AST_FRAME_SET_BUFFER(&fr, buffer, 0, len);
    fr.frametype = AST_FRAME_VOICE;
    fr.subclass.format = instance->native_format;
    fr.samples = instance->native_codec->samples_count(&fr);
 
    duped_frame = ast_frisolate(&fr);
    if (!duped_frame) {
        ast_log(LOG_WARNING, "%s: Failed to isolate frame\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    {
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
        AST_LIST_INSERT_TAIL(&instance->frame_queue, duped_frame, frame_list);
        instance->frame_queue_length++;
        if (!instance->queue_full && instance->frame_queue_length >= QUEUE_LENGTH_XOFF_LEVEL) {
            instance->queue_full = 1;
            ast_debug(4, "%s: WebSocket sending %s\n",
                ast_channel_name(instance->channel), MEDIA_XOFF);
            ast_websocket_write_string(instance->websocket, MEDIA_XOFF);
        }
    }
 
    ast_debug(5, "%s: Queued %d byte frame\n", ast_channel_name(instance->channel),
        duped_frame->datalen);
 
    return 0;
}

References ast_channel_name(), ast_debug, AST_FRAME_SET_BUFFER, AST_FRAME_VOICE, ast_frisolate, AST_LIST_INSERT_TAIL, AST_LIST_LOCK, AST_LIST_UNLOCK, ast_log, ast_websocket_write_string(), websocket_pvt::channel, ast_frame::datalen, ast_frame_subclass::format, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, ast_frame::frametype, len(), LOG_WARNING, MEDIA_XOFF, websocket_pvt::native_codec, websocket_pvt::native_format, NULL, websocket_pvt::queue_full, QUEUE_LENGTH_XOFF_LEVEL, ast_frame::samples, ast_codec::samples_count, SCOPED_LOCK, ast_frame::subclass, and websocket_pvt::websocket.

Referenced by process_binary_message(), and process_text_message().

queue_option_frame()

static int queue_option_frame ( struct websocket_pvt *  instance, char *  buffer  )

static

Definition at line 460 of file chan_websocket.c.

{
    struct ast_frame fr = { 0, };
    struct ast_frame *duped_frame = NULL;
 
    AST_FRAME_SET_BUFFER(&fr, buffer, 0, strlen(buffer) + 1);
    fr.frametype = AST_FRAME_CONTROL;
    fr.subclass.integer = AST_CONTROL_OPTION;
 
    duped_frame = ast_frisolate(&fr);
    if (!duped_frame) {
        ast_log(LOG_WARNING, "%s: Failed to isolate frame\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    AST_LIST_LOCK(&instance->frame_queue);
    AST_LIST_INSERT_TAIL(&instance->frame_queue, duped_frame, frame_list);
    AST_LIST_UNLOCK(&instance->frame_queue);
 
    ast_debug(4, "%s: Queued '%s' option frame\n",
        ast_channel_name(instance->channel), buffer);
 
    return 0;
}

References ast_channel_name(), AST_CONTROL_OPTION, ast_debug, AST_FRAME_CONTROL, AST_FRAME_SET_BUFFER, ast_frisolate, AST_LIST_INSERT_TAIL, AST_LIST_LOCK, AST_LIST_UNLOCK, ast_log, websocket_pvt::channel, websocket_pvt::frame_queue, ast_frame::frametype, ast_frame_subclass::integer, LOG_WARNING, NULL, and ast_frame::subclass.

Referenced by process_text_message().

read_from_ws_and_queue()

static int read_from_ws_and_queue ( struct websocket_pvt *  instance )

static

Definition at line 765 of file chan_websocket.c.

{
    uint64_t payload_len = 0;
    char *payload = NULL;
    enum ast_websocket_opcode opcode;
    int fragmented = 0;
    int res = 0;
 
    if (!instance || !instance->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    ast_debug(9, "%s: Waiting for websocket to have data\n", ast_channel_name(instance->channel));
    res = ast_wait_for_input(
        ast_websocket_fd(instance->websocket), -1);
    if (res <= 0) {
        ast_log(LOG_WARNING, "%s: WebSocket read failed: %s\n",
            ast_channel_name(instance->channel), strerror(errno));
        return -1;
    }
 
    /*
     * We need to lock here to prevent the websocket handle from
     * being pulled out from under us if the core sends us a
     * hangup request.
     */
    ao2_lock(instance);
    if (!instance->websocket) {
        ao2_unlock(instance);
        return -1;
    }
 
    res = ast_websocket_read(instance->websocket, &payload, &payload_len,
        &opcode, &fragmented);
    ao2_unlock(instance);
    if (res) {
        return -1;
    }
    ast_debug(5, "%s: WebSocket read %d bytes\n", ast_channel_name(instance->channel),
        (int)payload_len);
 
    if (opcode == AST_WEBSOCKET_OPCODE_TEXT) {
        return process_text_message(instance, payload, payload_len);
    }
 
    if (opcode == AST_WEBSOCKET_OPCODE_CLOSE) {
        ast_debug(5, "%s: WebSocket closed by remote\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    if (opcode != AST_WEBSOCKET_OPCODE_BINARY) {
        ast_debug(5, "%s: WebSocket frame type %d not supported. Ignoring.\n",
            ast_channel_name(instance->channel), (int)opcode);
        return 0;
    }
 
    return process_binary_message(instance, payload, payload_len);
}

References ao2_lock, ao2_unlock, ast_channel_name(), ast_debug, ast_log, ast_wait_for_input(), ast_websocket_fd(), AST_WEBSOCKET_OPCODE_BINARY, AST_WEBSOCKET_OPCODE_CLOSE, AST_WEBSOCKET_OPCODE_TEXT, ast_websocket_read(), websocket_pvt::channel, errno, LOG_WARNING, NULL, process_binary_message(), process_text_message(), and websocket_pvt::websocket.

Referenced by read_thread_handler().

read_thread_handler()

static void * read_thread_handler ( void *  obj )

static

Definition at line 837 of file chan_websocket.c.

{
    RAII_VAR(struct websocket_pvt *, instance, obj, ao2_cleanup);
    RAII_VAR(char *, command, NULL, ast_free);
    int res = 0;
 
    ast_debug(3, "%s: Read thread started\n", ast_channel_name(instance->channel));
 
    /*
     * We need to tell the remote app what channel this media is for.
     * This is especially important for outbound connections otherwise
     * the app won't know who the media is for.
     */
    res = ast_asprintf(&command, "%s connection_id:%s channel:%s format:%s optimal_frame_size:%d ptime:%d", MEDIA_START,
        instance->connection_id, ast_channel_name(instance->channel),
        ast_format_get_name(instance->native_format),
        instance->optimal_frame_size, instance->native_codec->default_ms);
    if (res <= 0 || !command) {
        ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
        ast_log(LOG_ERROR, "%s: Failed to create MEDIA_START\n", ast_channel_name(instance->channel));
        return NULL;
    }
    res = ast_websocket_write_string(instance->websocket, command);
    if (res != 0) {
        ast_log(LOG_ERROR, "%s: Failed to send MEDIA_START\n", ast_channel_name(instance->channel));
        ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
        return NULL;
    }
    ast_debug(3, "%s: Sent %s\n", ast_channel_name(instance->channel),
        command);
 
    if (!instance->no_auto_answer) {
        ast_debug(3, "%s: ANSWER by auto_answer\n", ast_channel_name(instance->channel));
        ast_queue_control(instance->channel, AST_CONTROL_ANSWER);
    }
 
    while (read_from_ws_and_queue(instance) == 0)
    {
    }
 
    /*
     * websocket_hangup will take care of closing the websocket if needed.
     */
    ast_debug(3, "%s: HANGUP by websocket close/error\n", ast_channel_name(instance->channel));
    ast_queue_control(instance->channel, AST_CONTROL_HANGUP);
 
    return NULL;
}

References ao2_cleanup, ast_asprintf, ast_channel_name(), AST_CONTROL_ANSWER, AST_CONTROL_HANGUP, ast_debug, ast_format_get_name(), ast_free, ast_log, ast_queue_control(), ast_websocket_write_string(), LOG_ERROR, MEDIA_START, NULL, RAII_VAR, and read_from_ws_and_queue().

Referenced by incoming_ws_established_cb(), and webchan_call().

set_channel_format()

static void set_channel_format ( struct websocket_pvt *  instance, struct ast_format *  fmt  )

static

Definition at line 201 of file chan_websocket.c.

{
    if (ast_format_cmp(ast_channel_rawreadformat(instance->channel), fmt)
        == AST_FORMAT_CMP_NOT_EQUAL) {
        ast_channel_set_rawreadformat(instance->channel, fmt);
        ast_set_read_format(instance->channel, ast_channel_readformat(instance->channel));
        ast_debug(4, "Switching readformat to %s\n", ast_format_get_name(fmt));
    }
}

References ast_channel_rawreadformat(), ast_channel_readformat(), ast_channel_set_rawreadformat(), ast_debug, ast_format_cmp(), AST_FORMAT_CMP_NOT_EQUAL, ast_format_get_name(), ast_set_read_format(), and websocket_pvt::channel.

Referenced by webchan_read().

set_channel_timer()

static int set_channel_timer ( struct websocket_pvt *  instance )

static

Definition at line 1208 of file chan_websocket.c.

{
    int rate = 0;
    instance->timer = ast_timer_open();
    if (!instance->timer) {
        return -1;
    }
    /* Rate is the number of ticks per second, not the interval. */
    rate = 1000 / ast_format_get_default_ms(instance->native_format);
    ast_debug(3, "%s: WebSocket timer rate %d\n",
        ast_channel_name(instance->channel), rate);
    ast_timer_set_rate(instance->timer, rate);
    /*
     * Calling ast_channel_set_fd will cause the channel thread to call
     * webchan_read at 'rate' times per second.
     */
    ast_channel_set_fd(instance->channel, 0, ast_timer_fd(instance->timer));
 
    return 0;
}

References ast_channel_name(), ast_channel_set_fd(), ast_debug, ast_format_get_default_ms(), ast_timer_fd(), ast_timer_open(), ast_timer_set_rate(), websocket_pvt::channel, websocket_pvt::native_format, and websocket_pvt::timer.

Referenced by webchan_request().

set_channel_variables()

static int set_channel_variables ( struct websocket_pvt *  instance )

static

Definition at line 1229 of file chan_websocket.c.

{
    char *pkt_size = NULL;
    int res = ast_asprintf(&pkt_size, "%d", instance->optimal_frame_size);
    if (res <= 0) {
        return -1;
    }
 
    pbx_builtin_setvar_helper(instance->channel, MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE,
        pkt_size);
    ast_free(pkt_size);
    pbx_builtin_setvar_helper(instance->channel, MEDIA_WEBSOCKET_CONNECTION_ID,
        instance->connection_id);
 
    return 0;
}

References ast_asprintf, ast_free, websocket_pvt::channel, websocket_pvt::connection_id, MEDIA_WEBSOCKET_CONNECTION_ID, MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE, NULL, websocket_pvt::optimal_frame_size, and pbx_builtin_setvar_helper().

Referenced by webchan_request().

set_instance_silence_frame()

static int set_instance_silence_frame ( struct websocket_pvt *  instance )

static

Definition at line 1185 of file chan_websocket.c.

{
    instance->silence.frametype = AST_FRAME_VOICE;
    instance->silence.datalen =
        (instance->slin_codec->default_ms * instance->slin_codec->minimum_bytes) / instance->slin_codec->minimum_ms;
    instance->silence.samples = instance->silence.datalen / sizeof(uint16_t);
    /*
     * Even though we'll calloc the data pointer, we don't mark it as
     * mallocd because this frame will be around for a while and we don't
     * want it accidentally freed before we're done with it.
     */
    instance->silence.mallocd = 0;
    instance->silence.offset = 0;
    instance->silence.src = __PRETTY_FUNCTION__;
    instance->silence.subclass.format = instance->slin_format;
    instance->silence.data.ptr = ast_calloc(1, instance->silence.datalen);
    if (!instance->silence.data.ptr) {
        return -1;
    }
 
    return 0;
}

References ast_calloc, AST_FRAME_VOICE, ast_frame::data, ast_frame::datalen, ast_codec::default_ms, ast_frame_subclass::format, ast_frame::frametype, ast_frame::mallocd, ast_codec::minimum_bytes, ast_codec::minimum_ms, ast_frame::offset, ast_frame::ptr, ast_frame::samples, websocket_pvt::silence, websocket_pvt::slin_codec, websocket_pvt::slin_format, ast_frame::src, and ast_frame::subclass.

Referenced by webchan_request().

set_instance_translator()

static int set_instance_translator ( struct websocket_pvt *  instance )

static

Definition at line 1154 of file chan_websocket.c.

{
    if (ast_format_cache_is_slinear(instance->native_format)) {
        instance->slin_format = ao2_bump(instance->native_format);
        instance->slin_codec = ast_format_get_codec(instance->slin_format);
        return 0;
    }
 
    instance->slin_format = ao2_bump(ast_format_cache_get_slin_by_rate(instance->native_codec->sample_rate));
    if (!instance->slin_format) {
        ast_log(LOG_ERROR, "%s: Unable to get slin format for rate %d\n",
            ast_channel_name(instance->channel), instance->native_codec->sample_rate);
        return -1;
    }
    ast_debug(3, "%s: WebSocket channel slin format '%s' Sample rate: %d ptime: %dms\n",
        ast_channel_name(instance->channel), ast_format_get_name(instance->slin_format),
        ast_format_get_sample_rate(instance->slin_format),
        ast_format_get_default_ms(instance->slin_format));
 
    instance->translator = ast_translator_build_path(instance->slin_format, instance->native_format);
    if (!instance->translator) {
        ast_log(LOG_ERROR, "%s: Unable to build translator path from '%s' to '%s'\n",
            ast_channel_name(instance->channel), ast_format_get_name(instance->native_format),
            ast_format_get_name(instance->slin_format));
        return -1;
    }
 
    instance->slin_codec = ast_format_get_codec(instance->slin_format);
    return 0;
}

References ao2_bump, ast_channel_name(), ast_debug, ast_format_cache_get_slin_by_rate(), ast_format_cache_is_slinear(), ast_format_get_codec(), ast_format_get_default_ms(), ast_format_get_name(), ast_format_get_sample_rate(), ast_log, ast_translator_build_path(), websocket_pvt::channel, LOG_ERROR, websocket_pvt::native_codec, websocket_pvt::native_format, ast_codec::sample_rate, websocket_pvt::slin_codec, websocket_pvt::slin_format, and websocket_pvt::translator.

Referenced by webchan_request().

unload_module()

static int unload_module ( void  )

static

Function called when our module is unloaded.

Definition at line 1689 of file chan_websocket.c.

{
    ast_http_uri_unlink(&http_uri);
    ao2_cleanup(ast_ws_server);
    ast_ws_server = NULL;
 
    ast_channel_unregister(&websocket_tech);
    ao2_cleanup(websocket_tech.capabilities);
    websocket_tech.capabilities = NULL;
 
    ao2_cleanup(instances);
    instances = NULL;
 
    return 0;
}

References ao2_cleanup, ast_channel_unregister(), ast_http_uri_unlink(), ast_ws_server, ast_channel_tech::capabilities, http_uri, instances, NULL, and websocket_tech.

Referenced by load_module().

validate_uri_parameters()

static int validate_uri_parameters ( const char *  uri_params )

static

Definition at line 1246 of file chan_websocket.c.

{
    char *params = ast_strdupa(uri_params);
    char *nvp = NULL;
    char *nv = NULL;
 
    /*
     * uri_params should be a comma-separated list of key=value pairs.
     * For example:
     * name1=value1,name2=value2
     * We're verifying that each name and value either doesn't need
     * to be encoded or that it already is.
     */
 
    while((nvp = ast_strsep(&params, ',', 0))) {
        /* nvp will be name1=value1 */
        while((nv = ast_strsep(&nvp, '=', 0))) {
            /* nv will be either name1 or value1 */
            if (!ast_uri_verify_encoded(nv)) {
                return 0;
            }
        }
    }
 
    return 1;
}

References ast_strdupa, ast_strsep(), ast_uri_verify_encoded(), NULL, and websocket_pvt::uri_params.

Referenced by webchan_request().

webchan_call()

static int webchan_call ( struct ast_channel *  ast, const char *  dest, int  timeout  )

static

Definition at line 919 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
    int nodelay = 1;
    enum ast_websocket_result result;
 
    if (!instance) {
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n",
            ast_channel_name(ast));
        return -1;
    }
 
    if (instance->type == AST_WS_TYPE_SERVER) {
        ast_debug(3, "%s: Websocket call incoming\n", ast_channel_name(instance->channel));
        return 0;
    }
    ast_debug(3, "%s: Websocket call outgoing\n", ast_channel_name(instance->channel));
 
    if (!instance->client) {
        ast_log(LOG_WARNING, "%s: WebSocket client not found\n",
            ast_channel_name(ast));
        return -1;
    }
 
    ast_debug(3, "%s: WebSocket call requested to %s. cid: %s\n",
        ast_channel_name(ast), dest, instance->connection_id);
 
    if (!ast_strlen_zero(instance->uri_params)) {
        ast_websocket_client_add_uri_params(instance->client, instance->uri_params);
    }
 
    instance->websocket = ast_websocket_client_connect(instance->client,
        instance, ast_channel_name(ast), &result);
    if (!instance->websocket || result != WS_OK) {
        ast_log(LOG_WARNING, "%s: WebSocket connection failed to %s: %s\n",
            ast_channel_name(ast), dest, ast_websocket_result_to_str(result));
        return -1;
    }
 
    if (setsockopt(ast_websocket_fd(instance->websocket),
        IPPROTO_TCP, TCP_NODELAY, (char *) &nodelay, sizeof(nodelay)) < 0) {
        ast_log(LOG_WARNING, "Failed to set TCP_NODELAY on websocket connection: %s\n", strerror(errno));
    }
 
    ast_debug(3, "%s: WebSocket connection to %s established\n",
        ast_channel_name(ast), dest);
 
    /* read_thread_handler() will clean up the bump */
    if (ast_pthread_create_detached_background(&instance->outbound_read_thread, NULL,
            read_thread_handler, ao2_bump(instance))) {
        ast_log(LOG_WARNING, "%s: Failed to create thread.\n", ast_channel_name(ast));
        ao2_cleanup(instance);
        return -1;
    }
 
    return 0;
}

References ao2_bump, ao2_cleanup, ast_channel_name(), ast_channel_tech_pvt(), ast_debug, ast_log, ast_pthread_create_detached_background, ast_strlen_zero(), ast_websocket_client_add_uri_params(), ast_websocket_client_connect(), ast_websocket_fd(), ast_websocket_result_to_str(), AST_WS_TYPE_SERVER, websocket_pvt::channel, websocket_pvt::client, websocket_pvt::connection_id, errno, LOG_WARNING, NULL, websocket_pvt::outbound_read_thread, read_thread_handler(), result, websocket_pvt::type, websocket_pvt::uri_params, websocket_pvt::websocket, and WS_OK.

webchan_hangup()

static int webchan_hangup ( struct ast_channel *  ast )

static

Definition at line 1470 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
 
    if (!instance) {
        return -1;
    }
    ast_debug(3, "%s: WebSocket call hangup. cid: %s\n",
        ast_channel_name(ast), instance->connection_id);
 
    /*
     * We need to lock because read_from_ws_and_queue() is probably waiting
     * on the websocket file descriptor and will unblock and immediately try to
     * check the websocket and read from it. We don't want to pull the
     * websocket out from under it between the check and read.
     */
    ao2_lock(instance);
    if (instance->websocket) {
        ast_websocket_close(instance->websocket, 1000);
        ast_websocket_unref(instance->websocket);
        instance->websocket = NULL;
    }
    ast_channel_tech_pvt_set(ast, NULL);
    ao2_unlock(instance);
 
    /* Clean up the reference from adding the instance to the channel */
    ao2_cleanup(instance);
 
    return 0;
}

References ao2_cleanup, ao2_lock, ao2_unlock, ast_channel_name(), ast_channel_tech_pvt(), ast_channel_tech_pvt_set(), ast_debug, ast_websocket_close(), ast_websocket_unref(), websocket_pvt::connection_id, NULL, and websocket_pvt::websocket.

webchan_read()

static struct ast_frame * webchan_read ( struct ast_channel *  ast )

static

Definition at line 323 of file chan_websocket.c.

{
    struct websocket_pvt *instance = NULL;
    struct ast_frame *native_frame = NULL;
    struct ast_frame *slin_frame = NULL;
 
    instance = ast_channel_tech_pvt(ast);
    if (!instance) {
        return NULL;
    }
 
    if (ast_timer_get_event(instance->timer) == AST_TIMING_EVENT_EXPIRED) {
        ast_timer_ack(instance->timer, 1);
    }
 
    native_frame = dequeue_frame(instance);
 
    /*
     * No frame when the timer fires means we have to create and
     * return a silence frame in its place.
     */
    if (!native_frame) {
        ast_debug(5, "%s: WebSocket read timer fired with no frame available.  Returning silence.\n", ast_channel_name(ast));
        set_channel_format(instance, instance->slin_format);
        slin_frame = ast_frdup(&instance->silence);
        return slin_frame;
    }
 
    /*
     * If we're in passthrough mode or the frame length is already optimal_frame_size,
     * we can just return it.
     */
    if (instance->passthrough || native_frame->datalen == instance->optimal_frame_size) {
        set_channel_format(instance, instance->native_format);
        return native_frame;
    }
 
    /*
     * If we're here, we have a short frame that we need to pad
     * with silence.
     */
 
    if (instance->translator) {
        slin_frame = ast_translate(instance->translator, native_frame, 0);
        if (!slin_frame) {
            ast_log(LOG_WARNING, "%s: Failed to translate %d byte frame\n",
                ast_channel_name(ast), native_frame->datalen);
            return NULL;
        }
        ast_frame_free(native_frame, 0);
    } else {
        /*
         * If there was no translator then the native format
         * was already slin.
         */
        slin_frame = native_frame;
    }
 
    set_channel_format(instance, instance->slin_format);
 
    /*
     * So now we have an slin frame but it's probably still short
     * so we create a new data buffer with the correct length
     * which is filled with zeros courtesy of ast_calloc.
     * We then copy the short frame data into the new buffer
     * and set the offset to AST_FRIENDLY_OFFSET so that
     * the core can read the data without any issues.
     * If the original frame data was mallocd, we need to free the old
     * data buffer so we don't leak memory and we need to set
     * mallocd to AST_MALLOCD_DATA so that the core knows
     * it needs to free the new data buffer when it's done.
     */
 
    if (slin_frame->datalen != instance->silence.datalen) {
        char *old_data = slin_frame->data.ptr;
        int old_len = slin_frame->datalen;
        int old_offset = slin_frame->offset;
        ast_debug(4, "%s: WebSocket read short frame. Expected %d got %d.  Filling with silence\n",
            ast_channel_name(ast), instance->silence.datalen,
            slin_frame->datalen);
 
        slin_frame->data.ptr = ast_calloc(1, instance->silence.datalen + AST_FRIENDLY_OFFSET);
        if (!slin_frame->data.ptr) {
            ast_frame_free(slin_frame, 0);
            return NULL;
        }
        slin_frame->data.ptr += AST_FRIENDLY_OFFSET;
        slin_frame->offset = AST_FRIENDLY_OFFSET;
        memcpy(slin_frame->data.ptr, old_data, old_len);
        if (slin_frame->mallocd & AST_MALLOCD_DATA) {
            ast_free(old_data - old_offset);
        }
        slin_frame->mallocd |= AST_MALLOCD_DATA;
        slin_frame->datalen = instance->silence.datalen;
        slin_frame->samples = instance->silence.samples;
    }
 
    return slin_frame;
}

References ast_calloc, ast_channel_name(), ast_channel_tech_pvt(), ast_debug, ast_frame_free(), ast_frdup, ast_free, AST_FRIENDLY_OFFSET, ast_log, AST_MALLOCD_DATA, ast_timer_ack(), ast_timer_get_event(), AST_TIMING_EVENT_EXPIRED, ast_translate(), ast_frame::data, ast_frame::datalen, dequeue_frame(), LOG_WARNING, ast_frame::mallocd, websocket_pvt::native_format, NULL, ast_frame::offset, websocket_pvt::optimal_frame_size, websocket_pvt::passthrough, ast_frame::ptr, ast_frame::samples, set_channel_format(), websocket_pvt::silence, websocket_pvt::slin_format, websocket_pvt::timer, and websocket_pvt::translator.

webchan_request()

static struct ast_channel * webchan_request ( const char *  type, struct ast_format_cap *  cap, const struct ast_assigned_ids *  assignedids, const struct ast_channel *  requestor, const char *  data, int *  cause  )

static

Definition at line 1295 of file chan_websocket.c.

{
    char *parse;
    RAII_VAR(struct websocket_pvt *, instance, NULL, ao2_cleanup);
    struct ast_channel *chan = NULL;
    struct ast_format *fmt = NULL;
    struct ast_format_cap *caps = NULL;
    AST_DECLARE_APP_ARGS(args,
        AST_APP_ARG(connection_id);
        AST_APP_ARG(options);
    );
    struct ast_flags opts = { 0, };
    char *opt_args[OPT_ARG_ARRAY_SIZE];
    const char *requestor_name = requestor ? ast_channel_name(requestor) : "no channel";
 
    ast_debug(3, "%s: WebSocket channel requested\n",
        requestor_name);
 
    if (ast_strlen_zero(data)) {
        ast_log(LOG_ERROR, "%s: A connection id is required for the 'WebSocket' channel\n",
            requestor_name);
        goto failure;
    }
    parse = ast_strdupa(data);
    AST_NONSTANDARD_APP_ARGS(args, parse, '/');
 
    if (ast_strlen_zero(args.connection_id)) {
        ast_log(LOG_ERROR, "%s: connection_id is required for the 'WebSocket' channel\n",
            requestor_name);
        goto failure;
    }
 
    if (!ast_strlen_zero(args.options)
        && ast_app_parse_options(websocket_options, &opts, opt_args,
            ast_strdupa(args.options))) {
        ast_log(LOG_ERROR, "%s: 'WebSocket' channel options '%s' parse error\n",
            requestor_name, args.options);
        goto failure;
    }
 
    if (ast_test_flag(&opts, OPT_WS_CODEC)
        && !ast_strlen_zero(opt_args[OPT_ARG_WS_CODEC])) {
        ast_debug(3, "%s: Using specified format %s\n",
            requestor_name, opt_args[OPT_ARG_WS_CODEC]);
        fmt = ast_format_cache_get(opt_args[OPT_ARG_WS_CODEC]);
    } else {
        /*
         * If codec wasn't specified in the dial string,
         * use the first format in the capabilities.
         */
        ast_debug(3, "%s: Using format %s from requesting channel\n",
            requestor_name, opt_args[OPT_ARG_WS_CODEC]);
        fmt = ast_format_cap_get_format(cap, 0);
    }
 
    if (!fmt) {
        ast_log(LOG_WARNING, "%s: No codec found for sending media to connection '%s'\n",
            requestor_name, args.connection_id);
        goto failure;
    }
 
    instance = websocket_new(requestor_name, args.connection_id, fmt);
    if (!instance) {
        ast_log(LOG_ERROR, "%s: Failed to allocate WebSocket channel pvt\n",
            requestor_name);
        goto failure;
    }
 
    instance->no_auto_answer = ast_test_flag(&opts, OPT_WS_NO_AUTO_ANSWER);
    if (!instance->passthrough) {
        instance->passthrough = ast_test_flag(&opts, OPT_WS_PASSTHROUGH);
    }
 
    if (ast_test_flag(&opts, OPT_WS_URI_PARAM)
        && !ast_strlen_zero(opt_args[OPT_ARG_WS_URI_PARAM])) {
        char *comma;
 
        if (ast_strings_equal(args.connection_id, INCOMING_CONNECTION_ID)) {
            ast_log(LOG_ERROR,
                "%s: URI parameters are not allowed for 'WebSocket/INCOMING' channels\n",
                requestor_name);
            goto failure;
        }
 
        ast_debug(3, "%s: Using URI parameters '%s'\n",
            requestor_name, opt_args[OPT_ARG_WS_URI_PARAM]);
 
        if (!validate_uri_parameters(opt_args[OPT_ARG_WS_URI_PARAM])) {
            ast_log(LOG_ERROR, "%s: Invalid URI parameters '%s' in WebSocket/%s dial string\n",
                requestor_name, opt_args[OPT_ARG_WS_URI_PARAM],
                args.connection_id);
            goto failure;
        }
 
        instance->uri_params = ast_strdup(opt_args[OPT_ARG_WS_URI_PARAM]);
        comma = instance->uri_params;
        /*
         * The normal separator for query string components is an
         * ampersand ('&') but the Dial app interprets them as additional
         * channels to dial in parallel so we instruct users to separate
         * the parameters with commas (',') instead.  We now have to
         * convert those commas back to ampersands.
         */
        while ((comma = strchr(comma,','))) {
            *comma = '&';
        }
        ast_debug(3, "%s: Using final URI '%s'\n", requestor_name, instance->uri_params);
    }
 
    chan = ast_channel_alloc(1, AST_STATE_DOWN, "", "", "", "", "", assignedids,
        requestor, 0, "WebSocket/%s/%p", args.connection_id, instance);
    if (!chan) {
        ast_log(LOG_ERROR, "%s: Unable to alloc channel\n", ast_channel_name(requestor));
        goto failure;
    }
 
    ast_debug(3, "%s: WebSocket channel %s allocated for connection %s\n",
        ast_channel_name(chan), requestor_name,
        instance->connection_id);
 
    instance->channel = ao2_bump(chan);
    ast_channel_tech_set(instance->channel, &websocket_tech);
 
    if (set_instance_translator(instance) != 0) {
        goto failure;
    }
 
    if (set_instance_silence_frame(instance) != 0) {
        goto failure;
    }
 
    if (set_channel_timer(instance) != 0) {
        goto failure;
    }
 
    if (set_channel_variables(instance) != 0) {
        goto failure;
    }
 
    caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
    if (!caps) {
        ast_log(LOG_ERROR, "%s: Unable to alloc caps\n", requestor_name);
        goto failure;
    }
 
    ast_format_cap_append(caps, instance->native_format, 0);
    ast_channel_nativeformats_set(instance->channel, caps);
    ast_channel_set_writeformat(instance->channel, instance->native_format);
    ast_channel_set_rawwriteformat(instance->channel, instance->native_format);
    ast_channel_set_readformat(instance->channel, instance->native_format);
    ast_channel_set_rawreadformat(instance->channel, instance->native_format);
    ast_channel_tech_pvt_set(chan, ao2_bump(instance));
    ast_channel_unlock(chan);
    ao2_cleanup(caps);
 
    ast_debug(3, "%s: WebSocket channel created to %s\n",
        ast_channel_name(chan), args.connection_id);
 
    return chan;
 
failure:
    if (chan) {
        ast_channel_unlock(chan);
    }
    *cause = AST_CAUSE_FAILURE;
    return NULL;
}

References ao2_bump, ao2_cleanup, args, AST_APP_ARG, ast_app_parse_options(), AST_CAUSE_FAILURE, ast_channel_alloc, ast_channel_name(), ast_channel_nativeformats_set(), ast_channel_set_rawreadformat(), ast_channel_set_rawwriteformat(), ast_channel_set_readformat(), ast_channel_set_writeformat(), ast_channel_tech_pvt_set(), ast_channel_tech_set(), ast_channel_unlock, ast_debug, AST_DECLARE_APP_ARGS, ast_format_cache_get, ast_format_cap_alloc, ast_format_cap_append, AST_FORMAT_CAP_FLAG_DEFAULT, ast_format_cap_get_format(), ast_log, AST_NONSTANDARD_APP_ARGS, AST_STATE_DOWN, ast_strdup, ast_strdupa, ast_strings_equal(), ast_strlen_zero(), ast_test_flag, INCOMING_CONNECTION_ID, LOG_ERROR, LOG_WARNING, NULL, OPT_ARG_ARRAY_SIZE, OPT_ARG_WS_CODEC, OPT_ARG_WS_URI_PARAM, OPT_WS_CODEC, OPT_WS_NO_AUTO_ANSWER, OPT_WS_PASSTHROUGH, OPT_WS_URI_PARAM, options, RAII_VAR, set_channel_timer(), set_channel_variables(), set_instance_silence_frame(), set_instance_translator(), validate_uri_parameters(), websocket_new(), websocket_options, and websocket_tech.

webchan_send_dtmf_text()

static int webchan_send_dtmf_text ( struct ast_channel *  ast, char  digit, unsigned int  duration  )

static

Definition at line 1501 of file chan_websocket.c.

{
        struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
        char *command;
        int res = 0;
 
        if (!instance) {
            return -1;
        }
 
        res = ast_asprintf(&command, "%s digit:%c", DTMF_END, digit);
        if (res <= 0 || !command) {
            ast_log(LOG_ERROR, "%s: Failed to create DTMF_END\n", ast_channel_name(instance->channel));
            return 0;
        }
        res = ast_websocket_write_string(instance->websocket, command);
        if (res != 0) {
            ast_log(LOG_ERROR, "%s: Failed to send DTMF_END\n", ast_channel_name(instance->channel));
            ast_free(command);
            return 0;
        }
        ast_debug(3, "%s: Sent %s\n", ast_channel_name(instance->channel), command);
        ast_free(command);
        return 0;
}

References ast_asprintf, ast_channel_name(), ast_channel_tech_pvt(), ast_debug, ast_free, ast_log, ast_websocket_write_string(), websocket_pvt::channel, digit, DTMF_END, LOG_ERROR, and websocket_pvt::websocket.

webchan_write()

static int webchan_write ( struct ast_channel *  ast, struct ast_frame *  f  )

static

Function called when we should write a frame to the channel.

Definition at line 887 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
 
    if (!instance || !instance->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket instance or client not found\n",
            ast_channel_name(ast));
        return -1;
    }
 
    if (f->frametype != AST_FRAME_VOICE) {
        ast_log(LOG_WARNING, "%s: This WebSocket channel only supports AST_FRAME_VOICE frames\n",
            ast_channel_name(ast));
        return -1;
    }
 
    if (ast_format_cmp(f->subclass.format, instance->native_format) == AST_FORMAT_CMP_NOT_EQUAL) {
        ast_log(LOG_WARNING, "%s: This WebSocket channel only supports the '%s' format, not '%s'\n",
            ast_channel_name(ast), ast_format_get_name(instance->native_format),
            ast_format_get_name(f->subclass.format));
        return -1;
    }
 
    return ast_websocket_write(instance->websocket, AST_WEBSOCKET_OPCODE_BINARY,
        (char *)f->data.ptr, (uint64_t)f->datalen);
}

References ast_channel_name(), ast_channel_tech_pvt(), ast_format_cmp(), AST_FORMAT_CMP_NOT_EQUAL, ast_format_get_name(), AST_FRAME_VOICE, ast_log, AST_WEBSOCKET_OPCODE_BINARY, ast_websocket_write(), ast_frame::data, ast_frame::datalen, ast_frame_subclass::format, ast_frame::frametype, LOG_WARNING, websocket_pvt::native_format, ast_frame::ptr, ast_frame::subclass, and websocket_pvt::websocket.

websocket_destructor()

static void websocket_destructor ( void *  data )

static

Definition at line 978 of file chan_websocket.c.

{
    struct websocket_pvt *instance = data;
    struct ast_frame *frame = NULL;
    ast_debug(3, "%s: WebSocket instance freed\n", instance->connection_id);
 
    AST_LIST_LOCK(&instance->frame_queue);
    while ((frame = AST_LIST_REMOVE_HEAD(&instance->frame_queue, frame_list))) {
        ast_frfree(frame);
    }
    AST_LIST_UNLOCK(&instance->frame_queue);
 
    if (instance->timer) {
        ast_timer_close(instance->timer);
        instance->timer = NULL;
    }
 
    if (instance->channel) {
        ast_channel_unref(instance->channel);
        instance->channel = NULL;
    }
    if (instance->websocket) {
        ast_websocket_unref(instance->websocket);
        instance->websocket = NULL;
    }
 
    ao2_cleanup(instance->client);
    instance->client = NULL;
 
    ao2_cleanup(instance->native_codec);
    instance->native_codec = NULL;
 
    ao2_cleanup(instance->native_format);
    instance->native_format = NULL;
 
    ao2_cleanup(instance->slin_codec);
    instance->slin_codec = NULL;
 
    ao2_cleanup(instance->slin_format);
    instance->slin_format = NULL;
 
    if (instance->silence.data.ptr) {
        ast_free(instance->silence.data.ptr);
        instance->silence.data.ptr = NULL;
    }
 
    if (instance->translator) {
        ast_translator_free_path(instance->translator);
        instance->translator = NULL;
    }
 
    if (instance->leftover_data) {
        ast_free(instance->leftover_data);
        instance->leftover_data = NULL;
    }
 
    ast_free(instance->uri_params);
}

References ao2_cleanup, ast_channel_unref, ast_debug, ast_free, ast_frfree, AST_LIST_LOCK, AST_LIST_REMOVE_HEAD, AST_LIST_UNLOCK, ast_timer_close(), ast_translator_free_path(), ast_websocket_unref(), websocket_pvt::channel, websocket_pvt::client, websocket_pvt::connection_id, ast_frame::data, websocket_pvt::frame_queue, websocket_pvt::leftover_data, websocket_pvt::native_codec, websocket_pvt::native_format, NULL, ast_frame::ptr, websocket_pvt::silence, websocket_pvt::slin_codec, websocket_pvt::slin_format, websocket_pvt::timer, websocket_pvt::translator, websocket_pvt::uri_params, and websocket_pvt::websocket.

Referenced by websocket_new().

websocket_new()

static struct websocket_pvt * websocket_new ( const char *  chan_name, const char *  connection_id, struct ast_format *  fmt  )

static

Definition at line 1050 of file chan_websocket.c.

{
    RAII_VAR(struct instance_proxy *, proxy, NULL, ao2_cleanup);
    RAII_VAR(struct websocket_pvt *, instance, NULL, ao2_cleanup);
    char uuid[AST_UUID_STR_LEN];
    enum ast_websocket_type ws_type;
 
    SCOPED_AO2WRLOCK(locker, instances);
 
    if (ast_strings_equal(connection_id, INCOMING_CONNECTION_ID)) {
        connection_id = ast_uuid_generate_str(uuid, sizeof(uuid));
        ws_type = AST_WS_TYPE_SERVER;
    } else {
        ws_type = AST_WS_TYPE_CLIENT;
    }
 
    proxy = ao2_weakproxy_alloc(sizeof(*proxy) + strlen(connection_id) + 1, NULL);
    if (!proxy) {
        return NULL;
    }
    strcpy(proxy->connection_id, connection_id); /* Safe */
 
    instance = ao2_alloc(sizeof(*instance) + strlen(connection_id) + 1,
        websocket_destructor);
    if (!instance) {
        return NULL;
    }
    strcpy(instance->connection_id, connection_id); /* Safe */
 
    instance->type = ws_type;
    if (ws_type == AST_WS_TYPE_CLIENT) {
        instance->client = ast_websocket_client_retrieve_by_id(instance->connection_id);
        if (!instance->client) {
            ast_log(LOG_ERROR, "%s: WebSocket client connection '%s' not found\n",
                chan_name, instance->connection_id);
            return NULL;
        }
    }
 
    AST_LIST_HEAD_INIT(&instance->frame_queue);
 
    /*
     * We need the codec to calculate the number of samples in a frame
     * so we'll get it once and store it in the instance.
     *
     * References for native_format and native_codec are now held by the
     * instance and will be released when the instance is destroyed.
     */
    instance->native_format = fmt;
    instance->native_codec = ast_format_get_codec(instance->native_format);
    /*
     * References for native_format and native_codec are now held by the
     * instance and will be released when the instance is destroyed.
     */
 
    /*
     * It's not possible for us to re-time or re-frame media if the data
     * stream can't be broken up on arbitrary byte boundaries.  This is usually
     * indicated by the codec's minimum_bytes being small (10 bytes or less).
     * We need to force passthrough mode in this case.
     */
    if (instance->native_codec->minimum_bytes <= 10) {
        instance->passthrough = 1;
        instance->optimal_frame_size = 0;
    } else {
        instance->optimal_frame_size =
            (instance->native_codec->default_ms * instance->native_codec->minimum_bytes)
                / instance->native_codec->minimum_ms;
        instance->leftover_data = ast_calloc(1, instance->optimal_frame_size);
        if (!instance->leftover_data) {
            return NULL;
        }
    }
 
    ast_debug(3,
        "%s: WebSocket channel native format '%s' Sample rate: %d ptime: %dms minms: %u  minbytes: %u passthrough: %d optimal_frame_size: %d\n",
        chan_name, ast_format_get_name(instance->native_format),
        ast_format_get_sample_rate(instance->native_format),
        ast_format_get_default_ms(instance->native_format),
        ast_format_get_minimum_ms(instance->native_format),
        ast_format_get_minimum_bytes(instance->native_format),
        instance->passthrough,
        instance->optimal_frame_size);
 
    /* We have exclusive access to proxy and sorcery, no need for locking here. */
    if (ao2_weakproxy_set_object(proxy, instance, OBJ_NOLOCK)) {
        return NULL;
    }
 
    if (ao2_weakproxy_subscribe(proxy, instance_proxy_cb, NULL, OBJ_NOLOCK)) {
        return NULL;
    }
 
    if (!ao2_link_flags(instances, proxy, OBJ_NOLOCK)) {
        ast_log(LOG_ERROR, "%s: Unable to link WebSocket instance to instances\n",
            proxy->connection_id);
        return NULL;
    }
    ast_debug(3, "%s: WebSocket instance created and linked\n", proxy->connection_id);
 
    return ao2_bump(instance);
}

References ao2_alloc, ao2_bump, ao2_cleanup, ao2_link_flags, ao2_weakproxy_alloc, ao2_weakproxy_set_object, ao2_weakproxy_subscribe(), ast_calloc, ast_debug, ast_format_get_codec(), ast_format_get_default_ms(), ast_format_get_minimum_bytes(), ast_format_get_minimum_ms(), ast_format_get_name(), ast_format_get_sample_rate(), AST_LIST_HEAD_INIT, ast_log, ast_strings_equal(), ast_uuid_generate_str(), AST_UUID_STR_LEN, ast_websocket_client_retrieve_by_id(), AST_WS_TYPE_CLIENT, AST_WS_TYPE_SERVER, websocket_pvt::connection_id, INCOMING_CONNECTION_ID, instance_proxy_cb(), instances, LOG_ERROR, NULL, OBJ_NOLOCK, RAII_VAR, SCOPED_AO2WRLOCK, uuid(), and websocket_destructor().

Referenced by webchan_request().

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "Websocket Media Channel" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_CHANNEL_DRIVER, .requires = "res_http_websocket,res_websocket_client", }

static

Definition at line 1761 of file chan_websocket.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 1761 of file chan_websocket.c.

ast_ws_server

struct ast_websocket_server* ast_ws_server

static

Definition at line 122 of file chan_websocket.c.

Referenced by incoming_ws_http_callback(), load_module(), and unload_module().

http_uri

struct ast_http_uri http_uri

static

Definition at line 1678 of file chan_websocket.c.

Referenced by load_module(), and unload_module().

instances

struct ao2_container* instances = NULL

static

Definition at line 124 of file chan_websocket.c.

Referenced by incoming_ws_established_cb(), incoming_ws_http_callback(), instance_proxy_cb(), load_module(), unload_module(), and websocket_new().

websocket_options

const struct ast_app_option websocket_options[128] = { [ 'c' ] = { .flag = OPT_WS_CODEC , .arg_index = OPT_ARG_WS_CODEC + 1 }, [ 'n' ] = { .flag = OPT_WS_NO_AUTO_ANSWER }, [ 'v' ] = { .flag = OPT_WS_URI_PARAM , .arg_index = OPT_ARG_WS_URI_PARAM + 1 }, [ 'p' ] = { .flag = OPT_WS_PASSTHROUGH }, }

static

Definition at line 1293 of file chan_websocket.c.

Referenced by webchan_request().

websocket_tech

struct ast_channel_tech websocket_tech

static

Definition at line 190 of file chan_websocket.c.

Referenced by load_module(), unload_module(), and webchan_request().