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/json.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 "asterisk/sorcery.h"

Include dependency graph for chan_websocket.c:

Go to the source code of this file.

Data Structures
struct	instance_proxy
struct	webchan_conf_global
struct	websocket_pvt

Macros
#define	_create_event_MEDIA_XOFF(_instance)   _create_event_nodata(_instance, "MEDIA_XOFF");
#define	_create_event_MEDIA_XON(_instance)   _create_event_nodata(_instance, "MEDIA_XON");
#define	_create_event_QUEUE_DRAINED(_instance)   _create_event_nodata(_instance, "QUEUE_DRAINED");
#define	ANSWER_CHANNEL   "ANSWER"
#define	CONTINUE_MEDIA   "CONTINUE_MEDIA"
#define	create_event(_instance, _event, ...)    _create_event_ ## _event(_instance, ##__VA_ARGS__)
	Use this macro to create events passing in any event-specific parameters.
#define	ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, direction)
#define	ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command)
#define	ERROR_ON_UNBUFFERED_MODE_RTN(instance, command)
#define	FLUSH_MEDIA   "FLUSH_MEDIA"
#define	GET_DRIVER_STATUS   "GET_STATUS"
#define	HANGUP_CHANNEL   "HANGUP"
#define	INCOMING_CONNECTION_ID   "INCOMING"
#define	MARK_MEDIA   "MARK_MEDIA"
#define	MAX_TEXT_MESSAGE_LEN   MIN(128, (AST_WEBSOCKET_MAX_RX_PAYLOAD_SIZE - 1))
#define	MEDIA_WEBSOCKET_CONNECTION_ID   "MEDIA_WEBSOCKET_CONNECTION_ID"
#define	MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE   "MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE"
#define	PAUSE_MEDIA   "PAUSE_MEDIA"
#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	send_event(_instance, _event, ...)
	Use this macro to create and send events passing in any event-specific parameters.
#define	SET_MEDIA_DIRECTION   "SET_MEDIA_DIRECTION"
#define	START_MEDIA_BUFFERING   "START_MEDIA_BUFFERING"
#define	STOP_MEDIA_BUFFERING   "STOP_MEDIA_BUFFERING"
#define	websocket_request_hangup(_instance, _cause, _tech)    _websocket_request_hangup(_instance, _cause, _tech, __LINE__, __FUNCTION__)
#define	WS_TIMER_FDNO   (AST_EXTENDED_FDS + 1)
#define	WS_WEBSOCKET_FDNO   (AST_EXTENDED_FDS + 2)

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) ,   OPT_WS_MSG_FORMAT = (1 << 4) , OPT_WS_MEDIA_DIRECTION = (1 << 5) , OPT_WS_UNBUFFERED = (1 << 6) }
enum	{   OPT_ARG_WS_CODEC , OPT_ARG_WS_NO_AUTO_ANSWER , OPT_ARG_WS_URI_PARAM , OPT_ARG_WS_PASSTHROUGH ,   OPT_ARG_WS_MSG_FORMAT , OPT_ARG_WS_MEDIA_DIRECTION , OPT_ARG_WS_UNBUFFERED , OPT_ARG_ARRAY_SIZE }
enum	webchan_control_msg_format { WEBCHAN_CONTROL_MSG_FORMAT_PLAIN = 0 , WEBCHAN_CONTROL_MSG_FORMAT_JSON , WEBCHAN_CONTROL_MSG_FORMAT_INVALID }
enum	webchan_media_direction { WEBCHAN_MEDIA_DIRECTION_BOTH , WEBCHAN_MEDIA_DIRECTION_OUT , WEBCHAN_MEDIA_DIRECTION_IN }

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
static char *	_create_event_DTMF_END (struct websocket_pvt *instance, const char digit)
static char *	_create_event_ERROR (struct websocket_pvt *instance, const char *format,...)
static char *	_create_event_MEDIA_BUFFERING_COMPLETED (struct websocket_pvt *instance, const char *id)
static char *	_create_event_MEDIA_MARK_PROCESSED (struct websocket_pvt *instance, const char *id)
static char *	_create_event_MEDIA_START (struct websocket_pvt *instance)
static char *	_create_event_nodata (struct websocket_pvt *instance, char *event)
static char *	_create_event_STATUS (struct websocket_pvt *instance)
static void	_websocket_request_hangup (struct websocket_pvt *instance, int ast_cause, enum ast_websocket_status_code tech_cause, int line, const char *function)
struct ast_module *	AST_MODULE_SELF_SYM (void)
static enum webchan_control_msg_format	control_msg_format_from_str (const char *value)
static const char *	control_msg_format_to_str (enum webchan_control_msg_format value)
static struct ast_frame *	create_frame_from_buffer (struct websocket_pvt *instance, char *buffer, size_t len)
static struct ast_frame *	dequeue_frame (struct websocket_pvt *instance)
static void *	global_alloc (const char *name)
static int	global_apply (const struct ast_sorcery *sorcery, void *obj)
static int	global_control_message_format_from_str (const struct aco_option *opt, struct ast_variable *var, void *obj)
static int	global_control_message_format_to_str (const void *obj, const intptr_t *args, char **buf)
static int	handle_command (struct websocket_pvt *instance, char *buffer)
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_config (void)
static int	load_module (void)
	Function called when our module is loaded.
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 int	reload_module (void)
static int	set_channel_timer (struct websocket_pvt *instance)
static int	set_channel_variables (struct websocket_pvt *instance)
static int	unload_module (void)
	Function called when our module is unloaded.
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.
static void	websocket_destructor (void *data)
static int	websocket_handoff_to_channel (struct websocket_pvt *instance)
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 = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module, .reload = reload_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 char *	msg_format_map []
static struct ast_sorcery *	sorcery = NULL
static const char *	websocket_media_direction_map []
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 }, [ 'f' ] = { .flag = OPT_WS_MSG_FORMAT , .arg_index = OPT_ARG_WS_MSG_FORMAT + 1 }, [ 'd' ] = { .flag = OPT_WS_MEDIA_DIRECTION , .arg_index = OPT_ARG_WS_MEDIA_DIRECTION + 1 }, [ 'u' ] = { .flag = OPT_WS_UNBUFFERED }, }
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

_create_event_MEDIA_XOFF

#define _create_event_MEDIA_XOFF

(

_instance

)

_create_event_nodata(_instance, "MEDIA_XOFF");

Definition at line 221 of file chan_websocket.c.

_create_event_MEDIA_XON

#define _create_event_MEDIA_XON

(

_instance

)

_create_event_nodata(_instance, "MEDIA_XON");

Definition at line 220 of file chan_websocket.c.

_create_event_QUEUE_DRAINED

#define _create_event_QUEUE_DRAINED

(

_instance

)

_create_event_nodata(_instance, "QUEUE_DRAINED");

Definition at line 222 of file chan_websocket.c.

ANSWER_CHANNEL

#define ANSWER_CHANNEL   "ANSWER"

Definition at line 132 of file chan_websocket.c.

CONTINUE_MEDIA

#define CONTINUE_MEDIA   "CONTINUE_MEDIA"

Definition at line 141 of file chan_websocket.c.

create_event

#define create_event	(		_instance,
			_event,
			...
	)		_create_event_ ## _event(_instance, ##__VA_ARGS__)

Use this macro to create events passing in any event-specific parameters.

Definition at line 441 of file chan_websocket.c.

 { \
    int _res = -1; \
    char *_payload = _create_event_ ## _event(_instance, ##__VA_ARGS__); \
    if (_payload && _instance->websocket) { \
        _res = ast_websocket_write_string(_instance->websocket, _payload); \
        if (_res != 0) { \
            ast_log(LOG_ERROR, "%s: Unable to send event %s\n", \
                ast_channel_name(instance->channel), _payload); \
        } else { \
            ast_debug(3, "%s: Sent %s\n", \
                ast_channel_name(instance->channel), _payload); \
        }\
        ast_free(_payload); \
    } \
    (_res); \
})
 
/*!
 * \internal
 *
 * This function gets called by webchan_read which is triggered by the channel
 * timer firing or by a websocket frame being received while in unbuffered mode.
 * It always gets called at least every 20ms (or whatever the timer is set to)
 * even if there are no frames in the queue.
 */
static struct ast_frame *dequeue_frame(struct websocket_pvt *instance)
{
    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.
     */
    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));
        send_event(instance, 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.
     * 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));
            send_event(instance, 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.
             */
            int res = ast_websocket_write_string(instance->websocket,
                queued_frame->data.ptr);
            if (res != 0) {
                ast_log(LOG_ERROR, "%s: Unable to send event %s\n",
                    ast_channel_name(instance->channel), (char *)queued_frame->data.ptr);
            } else {
                ast_debug(4, "%s: Sent %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.
     */
    if (!queued_frame) {
        return NULL;
    }
 
    instance->frame_queue_length--;
 
    return queued_frame;
}
 
static struct ast_frame *create_frame_from_buffer(struct websocket_pvt *instance,
    char *buffer, size_t len)
{
    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;
    if (instance->native_codec->samples_count) {
        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 NULL;
    }
 
    return duped_frame;
}
 
/*!
 * \internal
 *
 * There are two file descriptors on this channel that can trigger
 * this function...
 *
 *   The timer fd (WS_TIMER_FDNO) which gets triggered at a constant
 *   rate determined by the format.  In this case, we need to pull a
 *   frame OFF the queue and return it to the core.
 *
 *   The websocket fd (WS_WEBSOCKET_FDNO) which gets triggered when
 *   there's incoming data to read from the websocket.  In this case,
 *   we read the data and put it ON the queue.  If in unbuffered mode,
 *   we'll read it off the queue immediately and return it.  If not in
 *   unbuffered mode, we'll return a null frame now and the frame read
 *   will be processed by the timer tick when it comes to the head
 *   of the queue.
 *
 *   The reason for queueing and dequeueing immediately in unbuffered
 *   mode is that read_from_ws_and_queue does a lot of work before adding
 *   a frame to the frame queue, if it even needs to.  It's much simpler
 *   to just queue and dequeue, both from a code organization standpoint
 *   as well as an instruction-path-length standpoint, than it would be
 *   refactor that code so it can return a frame directly.
 */
static struct ast_frame *webchan_read(struct ast_channel *ast)
{
    struct websocket_pvt *instance = NULL;
    struct ast_frame *native_frame = NULL;
    int fdno = ast_channel_fdno(ast);
 
    instance = ast_channel_tech_pvt(ast);
    if (!instance) {
        return NULL;
    }
 
    if (fdno != WS_TIMER_FDNO && fdno != WS_WEBSOCKET_FDNO) {
        return &ast_null_frame;
    }
 
    if (fdno == WS_WEBSOCKET_FDNO) {
        read_from_ws_and_queue(instance);
        if (!instance->unbuffered) {
            return &ast_null_frame;
        }
    } else {
        if (ast_timer_get_event(instance->timer) == AST_TIMING_EVENT_EXPIRED) {
            ast_timer_ack(instance->timer, 1);
        }
    }
 
    native_frame = dequeue_frame(instance);
    if (!native_frame) {
        if (instance->leftover_len > 0) {
            native_frame = create_frame_from_buffer(instance, instance->leftover_data, instance->leftover_len);
            if (native_frame) {
                ast_debug(4, "%s: Triggered with no frame available but with %d bytes in leftover_data.  Returning partial frame.\n",
                    ast_channel_name(ast), (int)instance->leftover_len);
                instance->leftover_len = 0;
                return native_frame;
            }
        }
        ast_debug(4, "%s: Triggered with no frame available and no data in leftover_data.  Returning NULL frame.\n",
            ast_channel_name(ast));
        return &ast_null_frame;
    }
 
    ast_debug(5, "%s: Dequeued %d byte frame.  Left in buffer: %d\n",
        ast_channel_name(ast), native_frame->datalen, (int)instance->leftover_len);
 
    return native_frame;
}
 
static int queue_frame_from_buffer(struct websocket_pvt *instance,
    char *buffer, size_t len)
{
    struct ast_frame *duped_frame = NULL;
 
    duped_frame = create_frame_from_buffer(instance, buffer, len);
    if (!duped_frame) {
        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;
            send_event(instance, MEDIA_XOFF);
        }
    }
 
    ast_debug(5, "%s: Queued %d byte frame\n", ast_channel_name(instance->channel),
        duped_frame->datalen);
 
    return 0;
}
 
static int queue_option_frame(struct websocket_pvt *instance,
    char *buffer)
{
    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;
}
 
#define ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command) \
({ \
    if (instance->passthrough) { \
        send_event(instance, ERROR, "%s not supported in passthrough mode", command); \
        ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), command); \
        return 0; \
    } \
})
 
#define ERROR_ON_UNBUFFERED_MODE_RTN(instance, command) \
({ \
    if (instance->unbuffered) { \
        send_event(instance, ERROR, "%s not supported in unbuffered mode", command); \
        ast_debug(4, "%s: WebSocket in unbuffered mode. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), command); \
        return 0; \
    } \
})
 
#define ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, direction) \
({ \
    if (instance->media_direction == direction) { \
        send_event(instance, ERROR, "%s not supported while media direction " \
            "is '%s'", command, websocket_media_direction_map[direction]); \
        ast_debug(4, "%s: WebSocket media direction is '%s'. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), websocket_media_direction_map[direction], command); \
        return 0; \
    } \
})
 
/*!
 * \internal
 * \brief Handle commands from the websocket
 *
 * \param instance
 * \param buffer Allocated by caller so don't free.
 * \retval 0 Success
 * \retval -1 Failure
 */
static int handle_command(struct websocket_pvt *instance, char *buffer)
{
    int res = 0;
    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
    const char *command = NULL;
    char *data = NULL;
 
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json_error json_error;
 
        json = ast_json_load_buf(buffer, strlen(buffer), &json_error);
        if (!json) {
            send_event(instance, ERROR, "Unable to parse JSON command");
            return -1;
        }
        command = ast_json_object_string_get(json, "command");
    } else {
        command = buffer;
        data = strchr(buffer, ' ');
        if (data) {
            *data = '\0';
            data++;
        }
    }
 
    if (ast_strings_equal(command, ANSWER_CHANNEL)) {
        ast_queue_control(instance->channel, AST_CONTROL_ANSWER);
 
    } else if (ast_strings_equal(command, HANGUP_CHANNEL)) {
        websocket_request_hangup(instance, AST_CAUSE_NORMAL, AST_WEBSOCKET_STATUS_NORMAL);
 
    } else if (ast_strings_equal(command, START_MEDIA_BUFFERING)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        if (instance->bulk_media_in_progress) {
            send_event(instance, ERROR, "START_MEDIA_BUFFERING can't be called when media buffering is already active.\n");
            return 0;
        }
 
        AST_LIST_LOCK(&instance->frame_queue);
        instance->bulk_media_in_progress = 1;
        instance->last_unbuffered = instance->unbuffered;
        instance->unbuffered = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, STOP_MEDIA_BUFFERING)) {
        const char *id;
        char *option;
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            id = ast_json_object_string_get(json, "correlation_id");
        } else {
            id = data;
        }
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
 
        if (!instance->bulk_media_in_progress) {
            send_event(instance, ERROR, "STOP_MEDIA_BUFFERING can't be called when media buffering isn't active.\n");
            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;
        instance->unbuffered = instance->last_unbuffered;
        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;
        option = create_event(instance, MEDIA_BUFFERING_COMPLETED, id);
        if (!option) {
            return -1;
        }
        res = queue_option_frame(instance, option);
        ast_free(option);
 
    } else if (ast_strings_equal(command, MARK_MEDIA)) {
        const char *id;
        char *option;
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
 
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            id = ast_json_object_string_get(json, "correlation_id");
        } else {
            id = data;
        }
 
        ast_debug(4, "%s: %s %s\n",
            ast_channel_name(instance->channel), MARK_MEDIA, id);
 
        option = create_event(instance, MEDIA_MARK_PROCESSED, id);
        if (!option) {
            return -1;
        }
        res = queue_option_frame(instance, option);
        ast_free(option);
 
    } else if (ast_strings_equal(command, FLUSH_MEDIA)) {
        struct ast_frame *frame = NULL;
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
 
        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;
        instance->unbuffered = instance->last_unbuffered;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, REPORT_QUEUE_DRAINED)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
 
        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)) {
        return send_event(instance, STATUS);
 
    } else if (ast_strings_equal(command, PAUSE_MEDIA)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 1;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, CONTINUE_MEDIA)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, SET_MEDIA_DIRECTION)) {
        const char *direction;
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
 
        if (instance->control_msg_format != WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            send_event(instance, ERROR, "%s only supports JSON format.\n", command);
            return 0;
        }
 
        direction = ast_json_object_string_get(json, "direction");
        if (!direction) {
            send_event(instance, ERROR, "%s requires a 'direction' parameter.\n", command);
            return 0;
        }
 
        if (!strcmp("both", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_BOTH) {
                return 0;
            }
 
            if (!instance->timer) {
                set_channel_timer(instance);
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_BOTH;
 
        } else if (!strcmp("out", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_OUT) {
                return 0;
            }
 
            if (!instance->timer) {
                set_channel_timer(instance);
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_OUT;
 
        } else if (!strcmp("in", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_IN) {
                return 0;
            }
 
            if (instance->timer) {
                ast_channel_internal_fd_clear(instance->channel, WS_TIMER_FDNO);
                ast_timer_close(instance->timer);
                instance->timer = NULL;
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_IN;
 
        } else {
            send_event(instance, ERROR, "'%s' is not a valid direction for %s.\n",
                direction, command);
            return 0;
        }
 
    } else {
        ast_log(LOG_WARNING, "%s: WebSocket %s command unknown\n",
            ast_channel_name(instance->channel), command);
    }
 
    return res;
}
 
static int process_text_message(struct websocket_pvt *instance,
    char *payload, uint64_t payload_len)
{
    char *command;
 
    if (payload_len == 0) {
        ast_log(LOG_WARNING, "%s: WebSocket TEXT message has 0 length\n",
            ast_channel_name(instance->channel));
        return 0;
    }
 
    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: Received: %s\n",
        ast_channel_name(instance->channel), command);
 
    return handle_command(instance, command);
}
 
static int process_binary_message(struct websocket_pvt *instance,
    char *payload, uint64_t payload_len)
{
    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->unbuffered) {
        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;
}
 
static int read_from_ws_and_queue(struct websocket_pvt *instance)
{
    uint64_t payload_len = 0;
    char *payload = NULL;
    enum ast_websocket_opcode opcode;
    int fragmented = 0;
    int res = 0;
 
    if (!instance->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket session not found\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    res = ast_websocket_read(instance->websocket, &payload, &payload_len,
        &opcode, &fragmented);
 
    if (res) {
        ast_debug(3, "%s: WebSocket read error\n",
            ast_channel_name(instance->channel));
        websocket_request_hangup(instance, AST_CAUSE_NETWORK_OUT_OF_ORDER, AST_WEBSOCKET_STATUS_GOING_AWAY);
        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);
    }
 
    /*
     * PINGs and PONGs will have been handled by res_http_websocket.
     * We also need to ignore CONTINUATION frames as they will be accumulated
     * by res_http_websocket until the threshold set in websocket_handoff_to_channel()
     * is reached, then it will send us a TEXT or BINARY frame.
     */
    if (opcode == AST_WEBSOCKET_OPCODE_PING || opcode == AST_WEBSOCKET_OPCODE_PONG
        || opcode == AST_WEBSOCKET_OPCODE_CONTINUATION) {
        return 0;
    }
 
    if (opcode == AST_WEBSOCKET_OPCODE_CLOSE) {
        ast_debug(3, "%s: WebSocket closed by remote\n",
            ast_channel_name(instance->channel));
        websocket_request_hangup(instance, AST_CAUSE_NORMAL, AST_WEBSOCKET_STATUS_GOING_AWAY);
        return -1;
    }
 
    if (opcode == AST_WEBSOCKET_OPCODE_BINARY) {
        /* If the application's media direction is 'in', drop any media we receive from it */
        if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_IN) {
            ast_debug(5, "%s: WebSocket dropped frame (application media direction is 'in')\n",
                ast_channel_name(instance->channel));
            return 0;
        }
    } else {
        ast_log(LOG_WARNING, "%s: WebSocket frame type %d not supported\n",
            ast_channel_name(instance->channel), (int)opcode);
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_UNSUPPORTED_DATA);
        return 0;
    }
 
    return process_binary_message(instance, payload, payload_len);
}
 
static int websocket_handoff_to_channel(struct websocket_pvt *instance)
{
    int res = 0;
    int nodelay = 1;
    struct ast_sockaddr *remote_addr = ast_websocket_remote_address(instance->websocket);
 
    instance->remote_addr = ast_strdup(ast_sockaddr_stringify(remote_addr));
    ast_debug(3, "%s: WebSocket connection with %s established\n",
        ast_channel_name(instance->channel), instance->remote_addr);
 
    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));
    }
 
    /*
     * The way write timeouts are handled in iostream requires the socket to be
     * in non-blocking mode.  This is fine for reads as well because we already
     * set the websocket file descriptor on the channel and let it call
     * webchan_read() when data is available.
     */
    ast_websocket_set_nonblock(instance->websocket);
 
    /*
     * Tell res_http_websocket to accumulate incoming WebSocket CONTINUATION frames
     * into chunks of 1024 bytes and send us a TEXT or BINARY frame when the threshold
     * is reached.
     */
    ast_websocket_reconstruct_enable(instance->websocket, 1024);
 
    ast_channel_set_fd(instance->channel, WS_WEBSOCKET_FDNO, ast_websocket_fd(instance->websocket));
 
    res = send_event(instance, MEDIA_START);
    if (res != 0 ) {
        if (instance->type == AST_WS_TYPE_SERVER) {
            websocket_request_hangup(instance, AST_CAUSE_NETWORK_OUT_OF_ORDER, AST_WEBSOCKET_STATUS_GOING_AWAY);
        } else {
            /*
             * We were called by webchan_call so just need to set causes.
             * The core will hangup the channel.
             */
            ast_channel_tech_hangupcause_set(instance->channel, AST_WEBSOCKET_STATUS_GOING_AWAY);
            ast_channel_hangupcause_set(instance->channel, AST_CAUSE_NETWORK_OUT_OF_ORDER);
        }
        return -1;
    }
 
    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);
    }
 
    return 0;
}
 
static void _websocket_request_hangup(struct websocket_pvt *instance, int ast_cause,
    enum ast_websocket_status_code tech_cause, int line, const char *function)
{
    if (!instance || !instance->channel) {
        return;
    }
    ast_debug(3, "%s:%s: Hangup requested from %s line %d.  cause: %s(%d)  tech_cause: %s(%d)",
        ast_channel_name(instance->channel), instance->remote_addr,
        function, line,
        ast_cause2str(ast_cause), ast_cause, ast_websocket_status_to_str(tech_cause), tech_cause);
 
    if (tech_cause) {
        ast_channel_tech_hangupcause_set(instance->channel, tech_cause);
    }
    ast_queue_hangup_with_cause(instance->channel, ast_cause);
}
 
/*! \brief Function called when we should write a frame to the channel */
static int webchan_write(struct ast_channel *ast, struct ast_frame *f)
{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
    int res = 0;
 
    if (!instance || !instance->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket instance or client not found\n",
            ast_channel_name(ast));
        return -1;
    }
 
    /* The app doesn't want media right now */
    if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_OUT) {
        return 0;
    }
 
    if (f->frametype == AST_FRAME_CNG) {
        return 0;
    }
 
    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 0;
    }
 
    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;
    }
 
    res = ast_websocket_write(instance->websocket, AST_WEBSOCKET_OPCODE_BINARY,
        (char *)f->data.ptr, (uint64_t)f->datalen);
    if (res != 0) {
        ast_log(LOG_WARNING, "%s: WebSocket write failure\n", ast_channel_name(ast));
    }
 
    return res;
}
 
/*!
 * \internal
 *
 * Called by the core to actually call the remote.
 * The core will hang up the channel if a non-zero  is returned.
 * We just need to set hangup causes if appropriate.
 */
static int webchan_call(struct ast_channel *ast, const char *dest,
    int timeout)
{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
    enum ast_websocket_result result;
    struct webchan_conf_global *global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
    int global_write_timeout = global_cfg ? global_cfg->write_timeout : AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
 
    ao2_cleanup(global_cfg);
 
    if (!instance) {
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n",
            ast_channel_name(ast));
        ast_channel_hangupcause_set(ast, AST_CAUSE_FAILURE);
        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));
        ast_channel_hangupcause_set(ast, AST_CAUSE_FAILURE);
        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));
        ast_channel_hangupcause_set(ast, AST_CAUSE_NO_ROUTE_DESTINATION);
        return -1;
    }
 
    /*
     * If websocket_client->write_timeout was set in websocket_client.conf, it will
     * have been applied to the websocket by ast_websocket_client_connect() above.
     * If it wasn't set in websocket_client.conf, the value will be INT_MAX and
     * and ast_websocket_client_connect() will have set AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT
     * on the websocket.  However, the user may have set write_timeout in the global section
     * of chan_websocket.conf so if it wasn't set in websocket_client.conf, we'll now set
     * the websocket timeout to that.  If they  haven't set it in chan_websocket.conf either,
     * it'll default to AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT as well so the call below will
     * basically become a no-op.
     */
    if (instance->client->write_timeout == INT_MAX) {
        ast_websocket_set_timeout(instance->websocket, global_write_timeout);
    }
 
    return websocket_handoff_to_channel(instance);
}
 
static void websocket_destructor(void *data)
{
    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;
 
    if (instance->leftover_data) {
        ast_free(instance->leftover_data);
        instance->leftover_data = NULL;
    }
 
    ast_free(instance->uri_params);
    ast_free(instance->remote_addr);
}
 
struct instance_proxy {
    AO2_WEAKPROXY();
    /*! \brief The name of the module owning this sorcery instance */
    char connection_id[0];
};
 
static void instance_proxy_cb(void *weakproxy, void *data)
{
    struct instance_proxy *proxy = weakproxy;
    ast_debug(3, "%s: WebSocket instance removed from instances\n", proxy->connection_id);
    ao2_unlink(instances, weakproxy);
}
 
static struct websocket_pvt* websocket_new(const char *chan_name,
    const char *connection_id, struct ast_format *fmt)
{
    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 the passthrough and unbuffered modes in this case.
     */
    if (instance->native_codec->minimum_bytes <= 10) {
        instance->passthrough = 1;
        instance->unbuffered = 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);
}
 
static int set_channel_timer(struct websocket_pvt *instance)
{
    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, WS_TIMER_FDNO, ast_timer_fd(instance->timer));
 
    return 0;
}
 
static int set_channel_variables(struct websocket_pvt *instance)
{
    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;
}
 
static int validate_uri_parameters(const char *uri_params)
{
    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;
}
 
enum {
    OPT_WS_CODEC =  (1 << 0),
    OPT_WS_NO_AUTO_ANSWER =  (1 << 1),
    OPT_WS_URI_PARAM =  (1 << 2),
    OPT_WS_PASSTHROUGH =  (1 << 3),
    OPT_WS_MSG_FORMAT =  (1 << 4),
    OPT_WS_MEDIA_DIRECTION = (1 << 5),
    OPT_WS_UNBUFFERED = (1 << 6),
};
 
enum {
    OPT_ARG_WS_CODEC,
    OPT_ARG_WS_NO_AUTO_ANSWER,
    OPT_ARG_WS_URI_PARAM,
    OPT_ARG_WS_PASSTHROUGH,
    OPT_ARG_WS_MSG_FORMAT,
    OPT_ARG_WS_MEDIA_DIRECTION,
    OPT_ARG_WS_UNBUFFERED,
    OPT_ARG_ARRAY_SIZE
};
 
AST_APP_OPTIONS(websocket_options, BEGIN_OPTIONS
    AST_APP_OPTION_ARG('c', OPT_WS_CODEC, OPT_ARG_WS_CODEC),
    AST_APP_OPTION('n', OPT_WS_NO_AUTO_ANSWER),
    AST_APP_OPTION_ARG('v', OPT_WS_URI_PARAM, OPT_ARG_WS_URI_PARAM),
    AST_APP_OPTION('p', OPT_WS_PASSTHROUGH),
    AST_APP_OPTION_ARG('f', OPT_WS_MSG_FORMAT, OPT_ARG_WS_MSG_FORMAT),
    AST_APP_OPTION_ARG('d', OPT_WS_MEDIA_DIRECTION, OPT_ARG_WS_MEDIA_DIRECTION),
    AST_APP_OPTION('u', OPT_WS_UNBUFFERED),
    END_OPTIONS );
 
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)
{
    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) :
        (assignedids && !ast_strlen_zero(assignedids->uniqueid) ? assignedids->uniqueid : "<unknown>");
    RAII_VAR(struct webchan_conf_global *, global_cfg, NULL, ao2_cleanup);
 
    global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
 
    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])) {
        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.
         */
        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;
    }
 
    ast_debug(3, "%s: Using format %s from %s\n",
        requestor_name, ast_format_get_name(fmt),
        ast_test_flag(&opts, OPT_WS_CODEC) ? "dialstring" : "requester");
 
    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->media_direction = WEBCHAN_MEDIA_DIRECTION_BOTH;
    if (ast_test_flag(&opts, OPT_WS_MEDIA_DIRECTION)) {
        if (!strcmp("both", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            /* The default. Don't need to do anything here other than
             * ensure it is an allowed value. */
        } else if (!strcmp("out", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_OUT;
        } else if (!strcmp("in", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_IN;
        } else {
            ast_log(LOG_ERROR, "Unrecognized option for media direction: '%s'.\n",
                opt_args[OPT_ARG_WS_MEDIA_DIRECTION]);
            goto failure;
        }
    }
 
    instance->no_auto_answer = ast_test_flag(&opts, OPT_WS_NO_AUTO_ANSWER);
 
    /*
     * Passthrough requires unbuffered.  If passthrough was forced by choice of
     * codec, unbuffered will already have been set.  If passthrough was set by
     * the dialstring option, we need to force unbuffered.
     */
    if (!instance->passthrough) {
        instance->passthrough = ast_test_flag(&opts, OPT_WS_PASSTHROUGH);
        instance->unbuffered = instance->passthrough;
    }
    /*
     * If unbuffered hasn't been forced by passthrough, set it according to the
     * dialstring "u" option.
     */
    if (!instance->unbuffered) {
        instance->unbuffered = ast_test_flag(&opts, OPT_WS_UNBUFFERED);
    }
 
    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);
    }
 
    if (ast_test_flag(&opts, OPT_WS_MSG_FORMAT)) {
        instance->control_msg_format = control_msg_format_from_str(opt_args[OPT_ARG_WS_MSG_FORMAT]);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_INVALID) {
            ast_log(LOG_WARNING, "%s: 'f/control message format' dialstring parameter value missing or invalid. "
                "Defaulting to 'plain-text'\n",
                requestor_name);
            instance->control_msg_format = WEBCHAN_CONTROL_MSG_FORMAT_PLAIN;
        }
    } else if (global_cfg) {
        instance->control_msg_format = global_cfg->control_msg_format;
    }
 
    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", requestor_name);
        goto failure;
    }
 
    /* Prevent device state caching as this channel involves ephemeral destinations or sources */
    ast_set_flag(ast_channel_flags(chan), AST_FLAG_DISABLE_DEVSTATE_CACHE);
    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 the application's media direction is 'both' or 'out', we need the channel timer. */
    if (instance->media_direction != WEBCHAN_MEDIA_DIRECTION_IN
        && 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;
}
 
/*!
 * \internal
 *
 * Called by the core to hang up the channel.
 */
static int webchan_hangup(struct ast_channel *ast)
{
    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);
 
    if (instance->websocket) {
        ast_websocket_close(instance->websocket, ast_channel_tech_hangupcause(ast) ?: 1000);
        ast_websocket_unref(instance->websocket);
        instance->websocket = NULL;
    }
    ast_channel_tech_pvt_set(ast, NULL);
 
    /* Clean up the reference from adding the instance to the channel */
    ao2_cleanup(instance);
 
    return 0;
}
 
static int webchan_send_dtmf_text(struct ast_channel *ast, char digit, unsigned int duration)
{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
 
    if (!instance) {
        return -1;
    }
 
    return send_event(instance, DTMF_END, digit);
}
 
/*!
 * \internal
 *
 * Called by res_http_websocket after a client has connected and
 * successfully upgraded from HTTP to WebSocket.
 *
 * Depends on incoming_ws_http_callback parsing the connection_id from
 * the HTTP request and storing it in get_params.
 */
static void incoming_ws_established_cb(struct ast_websocket *ast_ws_session,
    struct ast_variable *get_params, struct ast_variable *upgrade_headers)
{
    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;
    struct webchan_conf_global *global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
    int global_write_timeout = global_cfg ? global_cfg->write_timeout : AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
 
    ao2_cleanup(global_cfg);
 
    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");
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        ast_websocket_close(ast_ws_session, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        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);
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        ast_websocket_close(ast_ws_session, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        return;
    }
    instance->websocket = ao2_bump(ast_ws_session);
 
    ast_websocket_set_timeout(instance->websocket, global_write_timeout);
 
    websocket_handoff_to_channel(instance);
    ao2_cleanup(instance);
    /*
     * The instance is the channel's responsibility now.
     * We just return here.
     */
}
 
/*!
 * \internal
 *
 * Called by the core http server after a client connects but before
 * the upgrade from HTTP to Websocket.  We need to save the URI in
 * the CONNECTION_ID in a get_param because it contains the connection UUID
 * we gave to the client when they used externalMedia to create the channel.
 * incoming_ws_established_cb() will use this to retrieve the chan_websocket
 * instance.
 */
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)
{
    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;
}
 
static struct ast_http_uri http_uri = {
    .callback = incoming_ws_http_callback,
    .description = "Media over Websocket",
    .uri = "media",
    .has_subtree = 1,
    .data = NULL,
    .key = __FILE__,
    .no_decode_uri = 1,
};
 
AO2_STRING_FIELD_HASH_FN(instance_proxy, connection_id)
AO2_STRING_FIELD_CMP_FN(instance_proxy, connection_id)
AO2_STRING_FIELD_SORT_FN(instance_proxy, connection_id)
 
static int global_control_message_format_from_str(const struct aco_option *opt,
    struct ast_variable *var, void *obj)
{
    struct webchan_conf_global *cfg = obj;
 
    cfg->control_msg_format = control_msg_format_from_str(var->value);
 
    if (cfg->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_INVALID) {
        ast_log(LOG_ERROR, "chan_websocket.conf: Invalid value '%s' for "
            "control_mesage_format. Must be 'plain-text' or 'json'\n",
            var->value);
        return -1;
    }
 
    return 0;
}
 
static int global_control_message_format_to_str(const void *obj, const intptr_t *args, char **buf)
{
    const struct webchan_conf_global *cfg = obj;
 
    *buf = ast_strdup(control_msg_format_to_str(cfg->control_msg_format));
 
    return 0;
}
 
static void *global_alloc(const char *name)
{
    struct webchan_conf_global *cfg = ast_sorcery_generic_alloc(
        sizeof(*cfg), NULL);
 
    if (!cfg) {
        return NULL;
    }
 
    return cfg;
}
 
static int global_apply(const struct ast_sorcery *sorcery, void *obj)
{
    struct webchan_conf_global *cfg = obj;
 
    ast_debug(1, "control_msg_format: %s\n",
        control_msg_format_to_str(cfg->control_msg_format));
 
    if (cfg->write_timeout <= 0) {
        ast_log(LOG_WARNING, "The write_timeout parameter must be > 0\n");
        return -1;
    }
 
    return 0;
}
 
static int load_config(void)
{
    ast_debug(2, "Initializing Websocket Client Configuration\n");
    sorcery = ast_sorcery_open();
    if (!sorcery) {
        ast_log(LOG_ERROR, "Failed to open sorcery\n");
        return -1;
    }
 
    ast_sorcery_apply_default(sorcery, "global", "config",
        "chan_websocket.conf,criteria=type=global,single_object=yes,explicit_name=global");
 
    if (ast_sorcery_object_register(sorcery, "global", global_alloc, NULL, global_apply)) {
        ast_log(LOG_ERROR, "Failed to register chan_websocket global object with sorcery\n");
        ast_sorcery_unref(sorcery);
        sorcery = NULL;
        return -1;
    }
 
    ast_sorcery_object_field_register_nodoc(sorcery, "global", "type", "", OPT_NOOP_T, 0, 0);
    ast_sorcery_register_cust(global, control_message_format, "plain-text");
    ast_sorcery_register_int(global, webchan_conf_global, write_timeout, write_timeout,
        AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT);
 
    ast_sorcery_load(sorcery);
 
    return 0;
}
 
/*! \brief Function called when our module is unloaded */
static int unload_module(void)
{
    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;
 
    ast_sorcery_unref(sorcery);
    sorcery = NULL;
 
    return 0;
}
 
static int reload_module(void)
{
    ast_debug(2, "Reloading chan_websocket configuration\n");
    ast_sorcery_reload(sorcery);
 
    return 0;
}
 
/*! \brief Function called when our module is loaded */
static int load_module(void)
{
    int res = 0;
    struct ast_websocket_protocol *protocol;
 
    res = load_config();
    if (res != 0) {
        return AST_MODULE_LOAD_DECLINE;
    }
 
    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;
}
 
AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_LOAD_ORDER, "Websocket Media Channel",
    .support_level = AST_MODULE_SUPPORT_CORE,
    .load = load_module,
    .unload = unload_module,
    .reload = reload_module,
    .load_pri = AST_MODPRI_CHANNEL_DRIVER,
    .requires = "res_http_websocket,res_websocket_client",
);

ERROR_ON_INVALID_MEDIA_DIRECTION_RTN

#define ERROR_ON_INVALID_MEDIA_DIRECTION_RTN	(		instance,
			command,
			direction
	)

Definition at line 745 of file chan_websocket.c.

 { \
    if (instance->media_direction == direction) { \
        send_event(instance, ERROR, "%s not supported while media direction " \
            "is '%s'", command, websocket_media_direction_map[direction]); \
        ast_debug(4, "%s: WebSocket media direction is '%s'. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), websocket_media_direction_map[direction], command); \
        return 0; \
    } \
})

ERROR_ON_PASSTHROUGH_MODE_RTN

#define ERROR_ON_PASSTHROUGH_MODE_RTN	(		instance,
			command
	)

Definition at line 725 of file chan_websocket.c.

 { \
    if (instance->passthrough) { \
        send_event(instance, ERROR, "%s not supported in passthrough mode", command); \
        ast_debug(4, "%s: WebSocket in passthrough mode. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), command); \
        return 0; \
    } \
})

ERROR_ON_UNBUFFERED_MODE_RTN

#define ERROR_ON_UNBUFFERED_MODE_RTN	(		instance,
			command
	)

Definition at line 735 of file chan_websocket.c.

 { \
    if (instance->unbuffered) { \
        send_event(instance, ERROR, "%s not supported in unbuffered mode", command); \
        ast_debug(4, "%s: WebSocket in unbuffered mode. Ignoring %s command.\n", \
            ast_channel_name(instance->channel), command); \
        return 0; \
    } \
})

FLUSH_MEDIA

#define FLUSH_MEDIA   "FLUSH_MEDIA"

Definition at line 137 of file chan_websocket.c.

GET_DRIVER_STATUS

#define GET_DRIVER_STATUS   "GET_STATUS"

Definition at line 138 of file chan_websocket.c.

HANGUP_CHANNEL

#define HANGUP_CHANNEL   "HANGUP"

Definition at line 133 of file chan_websocket.c.

INCOMING_CONNECTION_ID

#define INCOMING_CONNECTION_ID   "INCOMING"

Definition at line 130 of file chan_websocket.c.

MARK_MEDIA

#define MARK_MEDIA   "MARK_MEDIA"

Definition at line 136 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 147 of file chan_websocket.c.

MEDIA_WEBSOCKET_CONNECTION_ID

#define MEDIA_WEBSOCKET_CONNECTION_ID   "MEDIA_WEBSOCKET_CONNECTION_ID"

Definition at line 129 of file chan_websocket.c.

MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE

#define MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE   "MEDIA_WEBSOCKET_OPTIMAL_FRAME_SIZE"

Definition at line 128 of file chan_websocket.c.

PAUSE_MEDIA

#define PAUSE_MEDIA   "PAUSE_MEDIA"

Definition at line 140 of file chan_websocket.c.

QUEUE_LENGTH_MAX

#define QUEUE_LENGTH_MAX   1000

Definition at line 144 of file chan_websocket.c.

QUEUE_LENGTH_XOFF_LEVEL

#define QUEUE_LENGTH_XOFF_LEVEL   900

Definition at line 145 of file chan_websocket.c.

QUEUE_LENGTH_XON_LEVEL

#define QUEUE_LENGTH_XON_LEVEL   800

Definition at line 146 of file chan_websocket.c.

REPORT_QUEUE_DRAINED

#define REPORT_QUEUE_DRAINED   "REPORT_QUEUE_DRAINED"

Definition at line 139 of file chan_websocket.c.

send_event

#define send_event	(		_instance,
			_event,
			...
	)

Use this macro to create and send events passing in any event-specific parameters.

Definition at line 448 of file chan_websocket.c.

 { \
    int _res = -1; \
    char *_payload = _create_event_ ## _event(_instance, ##__VA_ARGS__); \
    if (_payload && _instance->websocket) { \
        _res = ast_websocket_write_string(_instance->websocket, _payload); \
        if (_res != 0) { \
            ast_log(LOG_ERROR, "%s: Unable to send event %s\n", \
                ast_channel_name(instance->channel), _payload); \
        } else { \
            ast_debug(3, "%s: Sent %s\n", \
                ast_channel_name(instance->channel), _payload); \
        }\
        ast_free(_payload); \
    } \
    (_res); \
})

SET_MEDIA_DIRECTION

#define SET_MEDIA_DIRECTION   "SET_MEDIA_DIRECTION"

Definition at line 142 of file chan_websocket.c.

START_MEDIA_BUFFERING

#define START_MEDIA_BUFFERING   "START_MEDIA_BUFFERING"

Definition at line 134 of file chan_websocket.c.

STOP_MEDIA_BUFFERING

#define STOP_MEDIA_BUFFERING   "STOP_MEDIA_BUFFERING"

Definition at line 135 of file chan_websocket.c.

websocket_request_hangup

#define websocket_request_hangup	(		_instance,
			_cause,
			_tech
	)		_websocket_request_hangup(_instance, _cause, _tech, __LINE__, __FUNCTION__)

Definition at line 161 of file chan_websocket.c.

WS_TIMER_FDNO

#define WS_TIMER_FDNO   (AST_EXTENDED_FDS + 1)

Definition at line 125 of file chan_websocket.c.

WS_WEBSOCKET_FDNO

#define WS_WEBSOCKET_FDNO   (AST_EXTENDED_FDS + 2)

Definition at line 126 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
OPT_WS_MSG_FORMAT
OPT_WS_MEDIA_DIRECTION
OPT_WS_UNBUFFERED

Definition at line 1620 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),
    OPT_WS_MSG_FORMAT =  (1 << 4),
    OPT_WS_MEDIA_DIRECTION = (1 << 5),
    OPT_WS_UNBUFFERED = (1 << 6),
};

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_WS_MSG_FORMAT
OPT_ARG_WS_MEDIA_DIRECTION
OPT_ARG_WS_UNBUFFERED
OPT_ARG_ARRAY_SIZE

Definition at line 1630 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_WS_MSG_FORMAT,
    OPT_ARG_WS_MEDIA_DIRECTION,
    OPT_ARG_WS_UNBUFFERED,
    OPT_ARG_ARRAY_SIZE
};

webchan_control_msg_format

enum webchan_control_msg_format

Enumerator
WEBCHAN_CONTROL_MSG_FORMAT_PLAIN
WEBCHAN_CONTROL_MSG_FORMAT_JSON
WEBCHAN_CONTROL_MSG_FORMAT_INVALID

Definition at line 56 of file chan_websocket.c.

                                {
    WEBCHAN_CONTROL_MSG_FORMAT_PLAIN = 0,
    WEBCHAN_CONTROL_MSG_FORMAT_JSON,
    WEBCHAN_CONTROL_MSG_FORMAT_INVALID,
};

webchan_media_direction

enum webchan_media_direction

Enumerator
WEBCHAN_MEDIA_DIRECTION_BOTH
WEBCHAN_MEDIA_DIRECTION_OUT
WEBCHAN_MEDIA_DIRECTION_IN

Definition at line 75 of file chan_websocket.c.

                             {
    WEBCHAN_MEDIA_DIRECTION_BOTH,
    WEBCHAN_MEDIA_DIRECTION_OUT,
    WEBCHAN_MEDIA_DIRECTION_IN,
};

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 2237 of file chan_websocket.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 2237 of file chan_websocket.c.

_create_event_DTMF_END()

static char * _create_event_DTMF_END ( struct websocket_pvt *  instance, const char  digit  )

static

Definition at line 334 of file chan_websocket.c.

{
    char *payload = NULL;
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:s#}",
            "event", "DTMF_END",
            "channel_id", ast_channel_uniqueid(instance->channel),
            "digit", &digit, 1
            );
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s digit:%c channel_id:%s",
            "DTMF_END", digit, ast_channel_uniqueid(instance->channel));
    }
 
    return payload;
}

References ast_asprintf, ast_channel_uniqueid(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), websocket_pvt::channel, websocket_pvt::control_msg_format, digit, NULL, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_ERROR()

static char * _create_event_ERROR ( struct websocket_pvt *  instance, const char *  format,   ...  )

static

Definition at line 402 of file chan_websocket.c.

{
    char *payload = NULL;
    char *error_text = NULL;
    va_list ap;
    int res = 0;
 
    va_start(ap, format);
    res = ast_vasprintf(&error_text, format, ap);
    va_end(ap);
    if (res < 0 || !error_text) {
        return NULL;
    }
 
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:s}",
            "event", "ERROR",
            "channel_id", ast_channel_uniqueid(instance->channel),
            "error_text", error_text);
        ast_free(error_text);
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s channel_id:%s error_text:%s",
            "ERROR", ast_channel_uniqueid(instance->channel), error_text);
        ast_free(error_text);
    }
 
    return payload;
}

References ast_asprintf, ast_channel_uniqueid(), ast_free, AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), ast_vasprintf, NULL, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_MEDIA_BUFFERING_COMPLETED()

static char * _create_event_MEDIA_BUFFERING_COMPLETED ( struct websocket_pvt *  instance, const char *  id  )

static

Definition at line 274 of file chan_websocket.c.

{
    char *payload = NULL;
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:s}",
            "event", "MEDIA_BUFFERING_COMPLETED",
            "channel_id", ast_channel_uniqueid(instance->channel),
            "correlation_id", S_OR(id, "")
            );
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s%s%s",
            "MEDIA_BUFFERING_COMPLETED",
            S_COR(id, " ",""), S_OR(id, ""));
 
    }
 
    return payload;
}

References ast_asprintf, ast_channel_uniqueid(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), websocket_pvt::channel, websocket_pvt::control_msg_format, NULL, S_COR, S_OR, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_MEDIA_MARK_PROCESSED()

static char * _create_event_MEDIA_MARK_PROCESSED ( struct websocket_pvt *  instance, const char *  id  )

static

Definition at line 304 of file chan_websocket.c.

{
    char *payload = NULL;
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:s}",
            "event", "MEDIA_MARK_PROCESSED",
            "channel_id", ast_channel_uniqueid(instance->channel),
            "correlation_id", S_OR(id, "")
            );
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s%s%s",
            "MEDIA_MARK_PROCESSED",
            S_COR(id, " ",""), S_OR(id, ""));
 
    }
 
    return payload;
}

References ast_asprintf, ast_channel_uniqueid(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), websocket_pvt::channel, websocket_pvt::control_msg_format, NULL, S_COR, S_OR, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_MEDIA_START()

static char * _create_event_MEDIA_START ( struct websocket_pvt *  instance )

static

Definition at line 229 of file chan_websocket.c.

{
    char *payload = NULL;
 
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:s, s:s, s:s, s:i, s:i, s:b, s:b, s:o }",
            "event", "MEDIA_START",
            "connection_id", instance->connection_id,
            "channel", ast_channel_name(instance->channel),
            "channel_id", ast_channel_uniqueid(instance->channel),
            "format", ast_format_get_name(instance->native_format),
            "optimal_frame_size", instance->optimal_frame_size,
            "ptime", instance->native_codec->default_ms,
            "passthrough", instance->passthrough,
            "unbuffered", instance->unbuffered,
            "channel_variables", ast_json_channel_vars(ast_channel_varshead(
                        instance->channel))
            );
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s %s:%s %s:%s %s:%s %s:%s %s:%d %s:%d %s:%s %s:%s",
            "MEDIA_START",
            "connection_id", instance->connection_id,
            "channel", ast_channel_name(instance->channel),
            "channel_id", ast_channel_uniqueid(instance->channel),
            "format", ast_format_get_name(instance->native_format),
            "optimal_frame_size", instance->optimal_frame_size,
            "ptime", instance->native_codec->default_ms,
            "passthrough", instance->passthrough ? "true" : "false",
            "unbuffered", instance->unbuffered ? "true" : "false"
            );
    }
 
    return payload;
}

References ast_asprintf, ast_channel_name(), ast_channel_uniqueid(), ast_channel_varshead(), ast_format_get_name(), ast_json_channel_vars(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), websocket_pvt::channel, websocket_pvt::connection_id, websocket_pvt::control_msg_format, ast_codec::default_ms, websocket_pvt::native_codec, websocket_pvt::native_format, NULL, websocket_pvt::optimal_frame_size, websocket_pvt::passthrough, websocket_pvt::unbuffered, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_nodata()

static char * _create_event_nodata ( struct websocket_pvt *  instance, char *  event  )

static

Definition at line 201 of file chan_websocket.c.

{
    char *payload = NULL;
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json * msg = ast_json_pack("{ s:s s:s }",
            "event", event,
            "channel_id", ast_channel_uniqueid(instance->channel));
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        payload = ast_strdup(event);
    }
 
    return payload;
}

References ast_channel_uniqueid(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), ast_strdup, websocket_pvt::channel, websocket_pvt::control_msg_format, NULL, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_create_event_STATUS()

static char * _create_event_STATUS ( struct websocket_pvt *  instance )

static

Definition at line 362 of file chan_websocket.c.

{
    char *payload = NULL;
 
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json *msg = ast_json_pack("{s:s, s:s, s:i, s:i, s:i, s:b, s:b, s:b }",
            "event", "STATUS",
            "channel_id", ast_channel_uniqueid(instance->channel),
            "queue_length", instance->frame_queue_length,
            "xon_level", QUEUE_LENGTH_XON_LEVEL,
            "xoff_level", QUEUE_LENGTH_XOFF_LEVEL,
            "queue_full", instance->queue_full,
            "bulk_media", instance->bulk_media_in_progress,
            "media_paused", instance->queue_paused
            );
        if (!msg) {
            return NULL;
        }
        payload = ast_json_dump_string_format(msg, AST_JSON_COMPACT);
        ast_json_unref(msg);
    } else {
        ast_asprintf(&payload, "%s channel_id:%s queue_length:%d xon_level:%d xoff_level:%d queue_full:%s bulk_media:%s media_paused:%s",
            "STATUS",
            ast_channel_uniqueid(instance->channel),
            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")
            );
    }
 
    return payload;
}

References ast_asprintf, ast_channel_uniqueid(), AST_JSON_COMPACT, ast_json_dump_string_format(), ast_json_pack(), ast_json_unref(), websocket_pvt::bulk_media_in_progress, websocket_pvt::channel, websocket_pvt::control_msg_format, websocket_pvt::frame_queue_length, NULL, websocket_pvt::queue_full, QUEUE_LENGTH_XOFF_LEVEL, QUEUE_LENGTH_XON_LEVEL, websocket_pvt::queue_paused, S_COR, and WEBCHAN_CONTROL_MSG_FORMAT_JSON.

_websocket_request_hangup()

static void _websocket_request_hangup ( struct websocket_pvt *  instance, int  ast_cause, enum ast_websocket_status_code  tech_cause, int  line, const char *  function  )

static

Definition at line 1262 of file chan_websocket.c.

{
    if (!instance || !instance->channel) {
        return;
    }
    ast_debug(3, "%s:%s: Hangup requested from %s line %d.  cause: %s(%d)  tech_cause: %s(%d)",
        ast_channel_name(instance->channel), instance->remote_addr,
        function, line,
        ast_cause2str(ast_cause), ast_cause, ast_websocket_status_to_str(tech_cause), tech_cause);
 
    if (tech_cause) {
        ast_channel_tech_hangupcause_set(instance->channel, tech_cause);
    }
    ast_queue_hangup_with_cause(instance->channel, ast_cause);
}

References ast_cause2str(), ast_channel_name(), ast_channel_tech_hangupcause_set(), ast_debug, ast_queue_hangup_with_cause(), ast_websocket_status_to_str(), websocket_pvt::channel, and websocket_pvt::remote_addr.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 2237 of file chan_websocket.c.

control_msg_format_from_str()

static enum webchan_control_msg_format control_msg_format_from_str ( const char *  value )

static

Definition at line 175 of file chan_websocket.c.

{
    if (ast_strlen_zero(value)) {
        return WEBCHAN_CONTROL_MSG_FORMAT_INVALID;
    } else if (strcasecmp(value, msg_format_map[WEBCHAN_CONTROL_MSG_FORMAT_PLAIN]) == 0) {
        return WEBCHAN_CONTROL_MSG_FORMAT_PLAIN;
    } else if (strcasecmp(value, msg_format_map[WEBCHAN_CONTROL_MSG_FORMAT_JSON]) == 0) {
        return WEBCHAN_CONTROL_MSG_FORMAT_JSON;
    } else {
        return WEBCHAN_CONTROL_MSG_FORMAT_INVALID;
    }
}

References ast_strlen_zero(), msg_format_map, value, WEBCHAN_CONTROL_MSG_FORMAT_INVALID, WEBCHAN_CONTROL_MSG_FORMAT_JSON, and WEBCHAN_CONTROL_MSG_FORMAT_PLAIN.

Referenced by global_control_message_format_from_str(), and webchan_request().

control_msg_format_to_str()

static const char * control_msg_format_to_str ( enum webchan_control_msg_format  value )

static

Definition at line 188 of file chan_websocket.c.

{
    if (!ARRAY_IN_BOUNDS(value, msg_format_map)) {
        return NULL;
    }
    return msg_format_map[value];
}

References ARRAY_IN_BOUNDS, msg_format_map, NULL, and value.

Referenced by global_apply(), and global_control_message_format_to_str().

create_frame_from_buffer()

static struct ast_frame * create_frame_from_buffer ( struct websocket_pvt *  instance, char *  buffer, size_t  len  )

static

Definition at line 575 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;
    if (instance->native_codec->samples_count) {
        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 NULL;
    }
 
    return duped_frame;
}

References ast_channel_name(), AST_FRAME_SET_BUFFER, AST_FRAME_VOICE, ast_frisolate, ast_log, websocket_pvt::channel, ast_frame_subclass::format, ast_frame::frametype, len(), LOG_WARNING, websocket_pvt::native_codec, websocket_pvt::native_format, NULL, ast_frame::samples, ast_codec::samples_count, and ast_frame::subclass.

Referenced by queue_frame_from_buffer(), and webchan_read().

dequeue_frame()

static struct ast_frame * dequeue_frame ( struct websocket_pvt *  instance )

static

Definition at line 474 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.
     */
    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));
        send_event(instance, 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.
     * 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));
            send_event(instance, 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.
             */
            int res = ast_websocket_write_string(instance->websocket,
                queued_frame->data.ptr);
            if (res != 0) {
                ast_log(LOG_ERROR, "%s: Unable to send event %s\n",
                    ast_channel_name(instance->channel), (char *)queued_frame->data.ptr);
            } else {
                ast_debug(4, "%s: Sent %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.
     */
    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_log, 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, LOG_ERROR, NULL, ast_frame::ptr, websocket_pvt::queue_full, QUEUE_LENGTH_XON_LEVEL, websocket_pvt::queue_paused, websocket_pvt::report_queue_drained, SCOPED_LOCK, send_event, ast_frame::subclass, and websocket_pvt::websocket.

Referenced by webchan_read().

global_alloc()

static void * global_alloc ( const char *  name )

static

Definition at line 2095 of file chan_websocket.c.

{
    struct webchan_conf_global *cfg = ast_sorcery_generic_alloc(
        sizeof(*cfg), NULL);
 
    if (!cfg) {
        return NULL;
    }
 
    return cfg;
}

References ast_sorcery_generic_alloc(), and NULL.

Referenced by load_config().

global_apply()

static int global_apply ( const struct ast_sorcery *  sorcery, void *  obj  )

static

Definition at line 2107 of file chan_websocket.c.

{
    struct webchan_conf_global *cfg = obj;
 
    ast_debug(1, "control_msg_format: %s\n",
        control_msg_format_to_str(cfg->control_msg_format));
 
    if (cfg->write_timeout <= 0) {
        ast_log(LOG_WARNING, "The write_timeout parameter must be > 0\n");
        return -1;
    }
 
    return 0;
}

References ast_debug, ast_log, webchan_conf_global::control_msg_format, control_msg_format_to_str(), LOG_WARNING, and webchan_conf_global::write_timeout.

Referenced by load_config().

global_control_message_format_from_str()

static int global_control_message_format_from_str ( const struct aco_option *  opt, struct ast_variable *  var, void *  obj  )

static

Definition at line 2069 of file chan_websocket.c.

{
    struct webchan_conf_global *cfg = obj;
 
    cfg->control_msg_format = control_msg_format_from_str(var->value);
 
    if (cfg->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_INVALID) {
        ast_log(LOG_ERROR, "chan_websocket.conf: Invalid value '%s' for "
            "control_mesage_format. Must be 'plain-text' or 'json'\n",
            var->value);
        return -1;
    }
 
    return 0;
}

References ast_log, webchan_conf_global::control_msg_format, control_msg_format_from_str(), LOG_ERROR, var, and WEBCHAN_CONTROL_MSG_FORMAT_INVALID.

global_control_message_format_to_str()

static int global_control_message_format_to_str ( const void *  obj, const intptr_t *  args, char **  buf  )

static

Definition at line 2086 of file chan_websocket.c.

{
    const struct webchan_conf_global *cfg = obj;
 
    *buf = ast_strdup(control_msg_format_to_str(cfg->control_msg_format));
 
    return 0;
}

References ast_strdup, buf, webchan_conf_global::control_msg_format, and control_msg_format_to_str().

handle_command()

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

static

Definition at line 765 of file chan_websocket.c.

{
    int res = 0;
    RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
    const char *command = NULL;
    char *data = NULL;
 
    if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
        struct ast_json_error json_error;
 
        json = ast_json_load_buf(buffer, strlen(buffer), &json_error);
        if (!json) {
            send_event(instance, ERROR, "Unable to parse JSON command");
            return -1;
        }
        command = ast_json_object_string_get(json, "command");
    } else {
        command = buffer;
        data = strchr(buffer, ' ');
        if (data) {
            *data = '\0';
            data++;
        }
    }
 
    if (ast_strings_equal(command, ANSWER_CHANNEL)) {
        ast_queue_control(instance->channel, AST_CONTROL_ANSWER);
 
    } else if (ast_strings_equal(command, HANGUP_CHANNEL)) {
        websocket_request_hangup(instance, AST_CAUSE_NORMAL, AST_WEBSOCKET_STATUS_NORMAL);
 
    } else if (ast_strings_equal(command, START_MEDIA_BUFFERING)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        if (instance->bulk_media_in_progress) {
            send_event(instance, ERROR, "START_MEDIA_BUFFERING can't be called when media buffering is already active.\n");
            return 0;
        }
 
        AST_LIST_LOCK(&instance->frame_queue);
        instance->bulk_media_in_progress = 1;
        instance->last_unbuffered = instance->unbuffered;
        instance->unbuffered = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, STOP_MEDIA_BUFFERING)) {
        const char *id;
        char *option;
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            id = ast_json_object_string_get(json, "correlation_id");
        } else {
            id = data;
        }
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
 
        if (!instance->bulk_media_in_progress) {
            send_event(instance, ERROR, "STOP_MEDIA_BUFFERING can't be called when media buffering isn't active.\n");
            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;
        instance->unbuffered = instance->last_unbuffered;
        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;
        option = create_event(instance, MEDIA_BUFFERING_COMPLETED, id);
        if (!option) {
            return -1;
        }
        res = queue_option_frame(instance, option);
        ast_free(option);
 
    } else if (ast_strings_equal(command, MARK_MEDIA)) {
        const char *id;
        char *option;
        SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
            AST_LIST_UNLOCK);
 
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            id = ast_json_object_string_get(json, "correlation_id");
        } else {
            id = data;
        }
 
        ast_debug(4, "%s: %s %s\n",
            ast_channel_name(instance->channel), MARK_MEDIA, id);
 
        option = create_event(instance, MEDIA_MARK_PROCESSED, id);
        if (!option) {
            return -1;
        }
        res = queue_option_frame(instance, option);
        ast_free(option);
 
    } else if (ast_strings_equal(command, FLUSH_MEDIA)) {
        struct ast_frame *frame = NULL;
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
 
        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;
        instance->unbuffered = instance->last_unbuffered;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, REPORT_QUEUE_DRAINED)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
 
        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)) {
        return send_event(instance, STATUS);
 
    } else if (ast_strings_equal(command, PAUSE_MEDIA)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 1;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, CONTINUE_MEDIA)) {
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
        ERROR_ON_UNBUFFERED_MODE_RTN(instance, command);
        ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
        AST_LIST_LOCK(&instance->frame_queue);
        instance->queue_paused = 0;
        AST_LIST_UNLOCK(&instance->frame_queue);
 
    } else if (ast_strings_equal(command, SET_MEDIA_DIRECTION)) {
        const char *direction;
 
        ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
 
        if (instance->control_msg_format != WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
            send_event(instance, ERROR, "%s only supports JSON format.\n", command);
            return 0;
        }
 
        direction = ast_json_object_string_get(json, "direction");
        if (!direction) {
            send_event(instance, ERROR, "%s requires a 'direction' parameter.\n", command);
            return 0;
        }
 
        if (!strcmp("both", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_BOTH) {
                return 0;
            }
 
            if (!instance->timer) {
                set_channel_timer(instance);
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_BOTH;
 
        } else if (!strcmp("out", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_OUT) {
                return 0;
            }
 
            if (!instance->timer) {
                set_channel_timer(instance);
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_OUT;
 
        } else if (!strcmp("in", direction)) {
            if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_IN) {
                return 0;
            }
 
            if (instance->timer) {
                ast_channel_internal_fd_clear(instance->channel, WS_TIMER_FDNO);
                ast_timer_close(instance->timer);
                instance->timer = NULL;
                ast_queue_frame(instance->channel, &ast_null_frame);
            }
 
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_IN;
 
        } else {
            send_event(instance, ERROR, "'%s' is not a valid direction for %s.\n",
                direction, command);
            return 0;
        }
 
    } else {
        ast_log(LOG_WARNING, "%s: WebSocket %s command unknown\n",
            ast_channel_name(instance->channel), command);
    }
 
    return res;
}

References ANSWER_CHANNEL, AST_CAUSE_NORMAL, ast_channel_internal_fd_clear(), ast_channel_name(), AST_CONTROL_ANSWER, ast_debug, ast_free, ast_frfree, ast_json_load_buf(), ast_json_object_string_get, ast_json_unref(), AST_LIST_LOCK, AST_LIST_REMOVE_HEAD, AST_LIST_UNLOCK, ast_log, ast_null_frame, ast_queue_control(), ast_queue_frame(), ast_strings_equal(), ast_timer_close(), AST_WEBSOCKET_STATUS_NORMAL, websocket_pvt::bulk_media_in_progress, websocket_pvt::channel, CONTINUE_MEDIA, websocket_pvt::control_msg_format, create_event, ast_frame::data, ERROR_ON_INVALID_MEDIA_DIRECTION_RTN, ERROR_ON_PASSTHROUGH_MODE_RTN, ERROR_ON_UNBUFFERED_MODE_RTN, FLUSH_MEDIA, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, GET_DRIVER_STATUS, HANGUP_CHANNEL, id, websocket_pvt::last_unbuffered, websocket_pvt::leftover_data, websocket_pvt::leftover_len, LOG_WARNING, MARK_MEDIA, websocket_pvt::media_direction, NULL, PAUSE_MEDIA, queue_frame_from_buffer(), queue_option_frame(), websocket_pvt::queue_paused, RAII_VAR, websocket_pvt::report_queue_drained, REPORT_QUEUE_DRAINED, SCOPED_LOCK, send_event, set_channel_timer(), SET_MEDIA_DIRECTION, START_MEDIA_BUFFERING, STOP_MEDIA_BUFFERING, websocket_pvt::timer, websocket_pvt::unbuffered, WEBCHAN_CONTROL_MSG_FORMAT_JSON, WEBCHAN_MEDIA_DIRECTION_BOTH, WEBCHAN_MEDIA_DIRECTION_IN, WEBCHAN_MEDIA_DIRECTION_OUT, websocket_request_hangup, and WS_TIMER_FDNO.

Referenced by process_text_message().

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 1912 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;
    struct webchan_conf_global *global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
    int global_write_timeout = global_cfg ? global_cfg->write_timeout : AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
 
    ao2_cleanup(global_cfg);
 
    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");
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        ast_websocket_close(ast_ws_session, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        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);
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        ast_websocket_close(ast_ws_session, AST_WEBSOCKET_STATUS_INTERNAL_ERROR);
        return;
    }
    instance->websocket = ao2_bump(ast_ws_session);
 
    ast_websocket_set_timeout(instance->websocket, global_write_timeout);
 
    websocket_handoff_to_channel(instance);
    ao2_cleanup(instance);
    /*
     * The instance is the channel's responsibility now.
     * We just return here.
     */
}

References ao2_bump, ao2_cleanup, ao2_weakproxy_find, AST_CAUSE_FAILURE, ast_debug, AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT, ast_log, ast_sorcery_retrieve_by_id(), ast_variable_find_in_list(), ast_websocket_close(), ast_websocket_set_timeout(), AST_WEBSOCKET_STATUS_INTERNAL_ERROR, ast_websocket_unref(), instances, LOG_WARNING, ast_variable::name, ast_variable::next, NULL, OBJ_NOLOCK, OBJ_SEARCH_KEY, RAII_VAR, sorcery, ast_variable::value, websocket_pvt::websocket, websocket_handoff_to_channel(), websocket_request_hangup, and webchan_conf_global::write_timeout.

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 1981 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 1443 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_config()

static int load_config ( void  )

static

Definition at line 2122 of file chan_websocket.c.

{
    ast_debug(2, "Initializing Websocket Client Configuration\n");
    sorcery = ast_sorcery_open();
    if (!sorcery) {
        ast_log(LOG_ERROR, "Failed to open sorcery\n");
        return -1;
    }
 
    ast_sorcery_apply_default(sorcery, "global", "config",
        "chan_websocket.conf,criteria=type=global,single_object=yes,explicit_name=global");
 
    if (ast_sorcery_object_register(sorcery, "global", global_alloc, NULL, global_apply)) {
        ast_log(LOG_ERROR, "Failed to register chan_websocket global object with sorcery\n");
        ast_sorcery_unref(sorcery);
        sorcery = NULL;
        return -1;
    }
 
    ast_sorcery_object_field_register_nodoc(sorcery, "global", "type", "", OPT_NOOP_T, 0, 0);
    ast_sorcery_register_cust(global, control_message_format, "plain-text");
    ast_sorcery_register_int(global, webchan_conf_global, write_timeout, write_timeout,
        AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT);
 
    ast_sorcery_load(sorcery);
 
    return 0;
}

References ast_debug, AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT, ast_log, ast_sorcery_apply_default, ast_sorcery_load(), ast_sorcery_object_field_register_nodoc, ast_sorcery_object_register, ast_sorcery_open, ast_sorcery_register_cust, ast_sorcery_register_int, ast_sorcery_unref, global, global_alloc(), global_apply(), LOG_ERROR, NULL, OPT_NOOP_T, sorcery, and webchan_conf_global::write_timeout.

Referenced by load_module().

load_module()

static int load_module ( void  )

static

Function called when our module is loaded.

Definition at line 2180 of file chan_websocket.c.

{
    int res = 0;
    struct ast_websocket_protocol *protocol;
 
    res = load_config();
    if (res != 0) {
        return AST_MODULE_LOAD_DECLINE;
    }
 
    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, load_config(), 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 1017 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->unbuffered) {
        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, queue_frame_from_buffer(), QUEUE_LENGTH_MAX, SCOPED_LOCK, and websocket_pvt::unbuffered.

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 984 of file chan_websocket.c.

{
    char *command;
 
    if (payload_len == 0) {
        ast_log(LOG_WARNING, "%s: WebSocket TEXT message has 0 length\n",
            ast_channel_name(instance->channel));
        return 0;
    }
 
    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: Received: %s\n",
        ast_channel_name(instance->channel), command);
 
    return handle_command(instance, command);
}

References ast_alloca, ast_channel_name(), ast_debug, ast_log, ast_strip(), websocket_pvt::channel, handle_command(), LOG_WARNING, and MAX_TEXT_MESSAGE_LEN.

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 671 of file chan_websocket.c.

{
    struct ast_frame *duped_frame = NULL;
 
    duped_frame = create_frame_from_buffer(instance, buffer, len);
    if (!duped_frame) {
        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;
            send_event(instance, 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_LIST_INSERT_TAIL, AST_LIST_LOCK, AST_LIST_UNLOCK, websocket_pvt::channel, create_frame_from_buffer(), ast_frame::datalen, websocket_pvt::frame_queue, websocket_pvt::frame_queue_length, len(), NULL, websocket_pvt::queue_full, QUEUE_LENGTH_XOFF_LEVEL, SCOPED_LOCK, and send_event.

Referenced by handle_command(), and process_binary_message().

queue_option_frame()

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

static

Definition at line 698 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 handle_command().

read_from_ws_and_queue()

static int read_from_ws_and_queue ( struct websocket_pvt *  instance )

static

Definition at line 1142 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->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket session not found\n",
            ast_channel_name(instance->channel));
        return -1;
    }
 
    res = ast_websocket_read(instance->websocket, &payload, &payload_len,
        &opcode, &fragmented);
 
    if (res) {
        ast_debug(3, "%s: WebSocket read error\n",
            ast_channel_name(instance->channel));
        websocket_request_hangup(instance, AST_CAUSE_NETWORK_OUT_OF_ORDER, AST_WEBSOCKET_STATUS_GOING_AWAY);
        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);
    }
 
    /*
     * PINGs and PONGs will have been handled by res_http_websocket.
     * We also need to ignore CONTINUATION frames as they will be accumulated
     * by res_http_websocket until the threshold set in websocket_handoff_to_channel()
     * is reached, then it will send us a TEXT or BINARY frame.
     */
    if (opcode == AST_WEBSOCKET_OPCODE_PING || opcode == AST_WEBSOCKET_OPCODE_PONG
        || opcode == AST_WEBSOCKET_OPCODE_CONTINUATION) {
        return 0;
    }
 
    if (opcode == AST_WEBSOCKET_OPCODE_CLOSE) {
        ast_debug(3, "%s: WebSocket closed by remote\n",
            ast_channel_name(instance->channel));
        websocket_request_hangup(instance, AST_CAUSE_NORMAL, AST_WEBSOCKET_STATUS_GOING_AWAY);
        return -1;
    }
 
    if (opcode == AST_WEBSOCKET_OPCODE_BINARY) {
        /* If the application's media direction is 'in', drop any media we receive from it */
        if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_IN) {
            ast_debug(5, "%s: WebSocket dropped frame (application media direction is 'in')\n",
                ast_channel_name(instance->channel));
            return 0;
        }
    } else {
        ast_log(LOG_WARNING, "%s: WebSocket frame type %d not supported\n",
            ast_channel_name(instance->channel), (int)opcode);
        websocket_request_hangup(instance, AST_CAUSE_FAILURE, AST_WEBSOCKET_STATUS_UNSUPPORTED_DATA);
        return 0;
    }
 
    return process_binary_message(instance, payload, payload_len);
}

References AST_CAUSE_FAILURE, AST_CAUSE_NETWORK_OUT_OF_ORDER, AST_CAUSE_NORMAL, ast_channel_name(), ast_debug, ast_log, AST_WEBSOCKET_OPCODE_BINARY, AST_WEBSOCKET_OPCODE_CLOSE, AST_WEBSOCKET_OPCODE_CONTINUATION, AST_WEBSOCKET_OPCODE_PING, AST_WEBSOCKET_OPCODE_PONG, AST_WEBSOCKET_OPCODE_TEXT, ast_websocket_read(), AST_WEBSOCKET_STATUS_GOING_AWAY, AST_WEBSOCKET_STATUS_UNSUPPORTED_DATA, websocket_pvt::channel, LOG_WARNING, websocket_pvt::media_direction, NULL, process_binary_message(), process_text_message(), WEBCHAN_MEDIA_DIRECTION_IN, websocket_pvt::websocket, and websocket_request_hangup.

Referenced by webchan_read().

reload_module()

static int reload_module ( void  )

static

Definition at line 2171 of file chan_websocket.c.

{
    ast_debug(2, "Reloading chan_websocket configuration\n");
    ast_sorcery_reload(sorcery);
 
    return 0;
}

References ast_debug, ast_sorcery_reload(), and sorcery.

set_channel_timer()

static int set_channel_timer ( struct websocket_pvt *  instance )

static

Definition at line 1555 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, WS_TIMER_FDNO, 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, websocket_pvt::timer, and WS_TIMER_FDNO.

Referenced by handle_command(), and webchan_request().

set_channel_variables()

static int set_channel_variables ( struct websocket_pvt *  instance )

static

Definition at line 1576 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().

unload_module()

static int unload_module ( void  )

static

Function called when our module is unloaded.

Definition at line 2152 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;
 
    ast_sorcery_unref(sorcery);
    sorcery = NULL;
 
    return 0;
}

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

Referenced by load_module().

validate_uri_parameters()

static int validate_uri_parameters ( const char *  uri_params )

static

Definition at line 1593 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 1329 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
    enum ast_websocket_result result;
    struct webchan_conf_global *global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
    int global_write_timeout = global_cfg ? global_cfg->write_timeout : AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
 
    ao2_cleanup(global_cfg);
 
    if (!instance) {
        ast_log(LOG_WARNING, "%s: WebSocket instance not found\n",
            ast_channel_name(ast));
        ast_channel_hangupcause_set(ast, AST_CAUSE_FAILURE);
        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));
        ast_channel_hangupcause_set(ast, AST_CAUSE_FAILURE);
        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));
        ast_channel_hangupcause_set(ast, AST_CAUSE_NO_ROUTE_DESTINATION);
        return -1;
    }
 
    /*
     * If websocket_client->write_timeout was set in websocket_client.conf, it will
     * have been applied to the websocket by ast_websocket_client_connect() above.
     * If it wasn't set in websocket_client.conf, the value will be INT_MAX and
     * and ast_websocket_client_connect() will have set AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT
     * on the websocket.  However, the user may have set write_timeout in the global section
     * of chan_websocket.conf so if it wasn't set in websocket_client.conf, we'll now set
     * the websocket timeout to that.  If they  haven't set it in chan_websocket.conf either,
     * it'll default to AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT as well so the call below will
     * basically become a no-op.
     */
    if (instance->client->write_timeout == INT_MAX) {
        ast_websocket_set_timeout(instance->websocket, global_write_timeout);
    }
 
    return websocket_handoff_to_channel(instance);
}

References ao2_cleanup, AST_CAUSE_FAILURE, AST_CAUSE_NO_ROUTE_DESTINATION, ast_channel_hangupcause_set(), ast_channel_name(), ast_channel_tech_pvt(), ast_debug, AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT, ast_log, ast_sorcery_retrieve_by_id(), ast_strlen_zero(), ast_websocket_client_add_uri_params(), ast_websocket_client_connect(), ast_websocket_result_to_str(), ast_websocket_set_timeout(), AST_WS_TYPE_SERVER, websocket_pvt::channel, websocket_pvt::client, websocket_pvt::connection_id, LOG_WARNING, result, sorcery, websocket_pvt::type, websocket_pvt::uri_params, websocket_pvt::websocket, websocket_handoff_to_channel(), webchan_conf_global::write_timeout, ast_websocket_client::write_timeout, and WS_OK.

webchan_hangup()

static int webchan_hangup ( struct ast_channel *  ast )

static

Definition at line 1869 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);
 
    if (instance->websocket) {
        ast_websocket_close(instance->websocket, ast_channel_tech_hangupcause(ast) ?: 1000);
        ast_websocket_unref(instance->websocket);
        instance->websocket = NULL;
    }
    ast_channel_tech_pvt_set(ast, NULL);
 
    /* Clean up the reference from adding the instance to the channel */
    ao2_cleanup(instance);
 
    return 0;
}

References ao2_cleanup, ast_channel_name(), ast_channel_tech_hangupcause(), 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 623 of file chan_websocket.c.

{
    struct websocket_pvt *instance = NULL;
    struct ast_frame *native_frame = NULL;
    int fdno = ast_channel_fdno(ast);
 
    instance = ast_channel_tech_pvt(ast);
    if (!instance) {
        return NULL;
    }
 
    if (fdno != WS_TIMER_FDNO && fdno != WS_WEBSOCKET_FDNO) {
        return &ast_null_frame;
    }
 
    if (fdno == WS_WEBSOCKET_FDNO) {
        read_from_ws_and_queue(instance);
        if (!instance->unbuffered) {
            return &ast_null_frame;
        }
    } else {
        if (ast_timer_get_event(instance->timer) == AST_TIMING_EVENT_EXPIRED) {
            ast_timer_ack(instance->timer, 1);
        }
    }
 
    native_frame = dequeue_frame(instance);
    if (!native_frame) {
        if (instance->leftover_len > 0) {
            native_frame = create_frame_from_buffer(instance, instance->leftover_data, instance->leftover_len);
            if (native_frame) {
                ast_debug(4, "%s: Triggered with no frame available but with %d bytes in leftover_data.  Returning partial frame.\n",
                    ast_channel_name(ast), (int)instance->leftover_len);
                instance->leftover_len = 0;
                return native_frame;
            }
        }
        ast_debug(4, "%s: Triggered with no frame available and no data in leftover_data.  Returning NULL frame.\n",
            ast_channel_name(ast));
        return &ast_null_frame;
    }
 
    ast_debug(5, "%s: Dequeued %d byte frame.  Left in buffer: %d\n",
        ast_channel_name(ast), native_frame->datalen, (int)instance->leftover_len);
 
    return native_frame;
}

References ast_channel_fdno(), ast_channel_name(), ast_channel_tech_pvt(), ast_debug, ast_null_frame, ast_timer_ack(), ast_timer_get_event(), AST_TIMING_EVENT_EXPIRED, create_frame_from_buffer(), ast_frame::datalen, dequeue_frame(), websocket_pvt::leftover_data, websocket_pvt::leftover_len, NULL, read_from_ws_and_queue(), websocket_pvt::timer, websocket_pvt::unbuffered, WS_TIMER_FDNO, and WS_WEBSOCKET_FDNO.

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 1651 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) :
        (assignedids && !ast_strlen_zero(assignedids->uniqueid) ? assignedids->uniqueid : "<unknown>");
    RAII_VAR(struct webchan_conf_global *, global_cfg, NULL, ao2_cleanup);
 
    global_cfg = ast_sorcery_retrieve_by_id(sorcery, "global", "global");
 
    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])) {
        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.
         */
        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;
    }
 
    ast_debug(3, "%s: Using format %s from %s\n",
        requestor_name, ast_format_get_name(fmt),
        ast_test_flag(&opts, OPT_WS_CODEC) ? "dialstring" : "requester");
 
    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->media_direction = WEBCHAN_MEDIA_DIRECTION_BOTH;
    if (ast_test_flag(&opts, OPT_WS_MEDIA_DIRECTION)) {
        if (!strcmp("both", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            /* The default. Don't need to do anything here other than
             * ensure it is an allowed value. */
        } else if (!strcmp("out", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_OUT;
        } else if (!strcmp("in", opt_args[OPT_ARG_WS_MEDIA_DIRECTION])) {
            instance->media_direction = WEBCHAN_MEDIA_DIRECTION_IN;
        } else {
            ast_log(LOG_ERROR, "Unrecognized option for media direction: '%s'.\n",
                opt_args[OPT_ARG_WS_MEDIA_DIRECTION]);
            goto failure;
        }
    }
 
    instance->no_auto_answer = ast_test_flag(&opts, OPT_WS_NO_AUTO_ANSWER);
 
    /*
     * Passthrough requires unbuffered.  If passthrough was forced by choice of
     * codec, unbuffered will already have been set.  If passthrough was set by
     * the dialstring option, we need to force unbuffered.
     */
    if (!instance->passthrough) {
        instance->passthrough = ast_test_flag(&opts, OPT_WS_PASSTHROUGH);
        instance->unbuffered = instance->passthrough;
    }
    /*
     * If unbuffered hasn't been forced by passthrough, set it according to the
     * dialstring "u" option.
     */
    if (!instance->unbuffered) {
        instance->unbuffered = ast_test_flag(&opts, OPT_WS_UNBUFFERED);
    }
 
    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);
    }
 
    if (ast_test_flag(&opts, OPT_WS_MSG_FORMAT)) {
        instance->control_msg_format = control_msg_format_from_str(opt_args[OPT_ARG_WS_MSG_FORMAT]);
 
        if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_INVALID) {
            ast_log(LOG_WARNING, "%s: 'f/control message format' dialstring parameter value missing or invalid. "
                "Defaulting to 'plain-text'\n",
                requestor_name);
            instance->control_msg_format = WEBCHAN_CONTROL_MSG_FORMAT_PLAIN;
        }
    } else if (global_cfg) {
        instance->control_msg_format = global_cfg->control_msg_format;
    }
 
    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", requestor_name);
        goto failure;
    }
 
    /* Prevent device state caching as this channel involves ephemeral destinations or sources */
    ast_set_flag(ast_channel_flags(chan), AST_FLAG_DISABLE_DEVSTATE_CACHE);
    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 the application's media direction is 'both' or 'out', we need the channel timer. */
    if (instance->media_direction != WEBCHAN_MEDIA_DIRECTION_IN
        && 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_flags(), 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_FLAG_DISABLE_DEVSTATE_CACHE, ast_format_cache_get, ast_format_cap_alloc, ast_format_cap_append, AST_FORMAT_CAP_FLAG_DEFAULT, ast_format_cap_get_format(), ast_format_get_name(), ast_log, AST_NONSTANDARD_APP_ARGS, ast_set_flag, ast_sorcery_retrieve_by_id(), AST_STATE_DOWN, ast_strdup, ast_strdupa, ast_strings_equal(), ast_strlen_zero(), ast_test_flag, control_msg_format_from_str(), INCOMING_CONNECTION_ID, LOG_ERROR, LOG_WARNING, NULL, OPT_ARG_ARRAY_SIZE, OPT_ARG_WS_CODEC, OPT_ARG_WS_MEDIA_DIRECTION, OPT_ARG_WS_MSG_FORMAT, OPT_ARG_WS_URI_PARAM, OPT_WS_CODEC, OPT_WS_MEDIA_DIRECTION, OPT_WS_MSG_FORMAT, OPT_WS_NO_AUTO_ANSWER, OPT_WS_PASSTHROUGH, OPT_WS_UNBUFFERED, OPT_WS_URI_PARAM, options, RAII_VAR, set_channel_timer(), set_channel_variables(), sorcery, ast_assigned_ids::uniqueid, validate_uri_parameters(), WEBCHAN_CONTROL_MSG_FORMAT_INVALID, WEBCHAN_CONTROL_MSG_FORMAT_PLAIN, WEBCHAN_MEDIA_DIRECTION_BOTH, WEBCHAN_MEDIA_DIRECTION_IN, WEBCHAN_MEDIA_DIRECTION_OUT, 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 1892 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
 
    if (!instance) {
        return -1;
    }
 
    return send_event(instance, DTMF_END, digit);
}

References ast_channel_tech_pvt(), digit, and send_event.

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 1280 of file chan_websocket.c.

{
    struct websocket_pvt *instance = ast_channel_tech_pvt(ast);
    int res = 0;
 
    if (!instance || !instance->websocket) {
        ast_log(LOG_WARNING, "%s: WebSocket instance or client not found\n",
            ast_channel_name(ast));
        return -1;
    }
 
    /* The app doesn't want media right now */
    if (instance->media_direction == WEBCHAN_MEDIA_DIRECTION_OUT) {
        return 0;
    }
 
    if (f->frametype == AST_FRAME_CNG) {
        return 0;
    }
 
    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 0;
    }
 
    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;
    }
 
    res = ast_websocket_write(instance->websocket, AST_WEBSOCKET_OPCODE_BINARY,
        (char *)f->data.ptr, (uint64_t)f->datalen);
    if (res != 0) {
        ast_log(LOG_WARNING, "%s: WebSocket write failure\n", ast_channel_name(ast));
    }
 
    return res;
}

References ast_channel_name(), ast_channel_tech_pvt(), ast_format_cmp(), AST_FORMAT_CMP_NOT_EQUAL, ast_format_get_name(), AST_FRAME_CNG, 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::media_direction, websocket_pvt::native_format, ast_frame::ptr, ast_frame::subclass, WEBCHAN_MEDIA_DIRECTION_OUT, and websocket_pvt::websocket.

websocket_destructor()

static void websocket_destructor ( void *  data )

static

Definition at line 1393 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;
 
    if (instance->leftover_data) {
        ast_free(instance->leftover_data);
        instance->leftover_data = NULL;
    }
 
    ast_free(instance->uri_params);
    ast_free(instance->remote_addr);
}

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_websocket_unref(), websocket_pvt::channel, websocket_pvt::client, websocket_pvt::connection_id, websocket_pvt::frame_queue, websocket_pvt::leftover_data, websocket_pvt::native_codec, websocket_pvt::native_format, NULL, websocket_pvt::remote_addr, websocket_pvt::timer, websocket_pvt::uri_params, and websocket_pvt::websocket.

Referenced by websocket_new().

websocket_handoff_to_channel()

static int websocket_handoff_to_channel ( struct websocket_pvt *  instance )

static

Definition at line 1207 of file chan_websocket.c.

{
    int res = 0;
    int nodelay = 1;
    struct ast_sockaddr *remote_addr = ast_websocket_remote_address(instance->websocket);
 
    instance->remote_addr = ast_strdup(ast_sockaddr_stringify(remote_addr));
    ast_debug(3, "%s: WebSocket connection with %s established\n",
        ast_channel_name(instance->channel), instance->remote_addr);
 
    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));
    }
 
    /*
     * The way write timeouts are handled in iostream requires the socket to be
     * in non-blocking mode.  This is fine for reads as well because we already
     * set the websocket file descriptor on the channel and let it call
     * webchan_read() when data is available.
     */
    ast_websocket_set_nonblock(instance->websocket);
 
    /*
     * Tell res_http_websocket to accumulate incoming WebSocket CONTINUATION frames
     * into chunks of 1024 bytes and send us a TEXT or BINARY frame when the threshold
     * is reached.
     */
    ast_websocket_reconstruct_enable(instance->websocket, 1024);
 
    ast_channel_set_fd(instance->channel, WS_WEBSOCKET_FDNO, ast_websocket_fd(instance->websocket));
 
    res = send_event(instance, MEDIA_START);
    if (res != 0 ) {
        if (instance->type == AST_WS_TYPE_SERVER) {
            websocket_request_hangup(instance, AST_CAUSE_NETWORK_OUT_OF_ORDER, AST_WEBSOCKET_STATUS_GOING_AWAY);
        } else {
            /*
             * We were called by webchan_call so just need to set causes.
             * The core will hangup the channel.
             */
            ast_channel_tech_hangupcause_set(instance->channel, AST_WEBSOCKET_STATUS_GOING_AWAY);
            ast_channel_hangupcause_set(instance->channel, AST_CAUSE_NETWORK_OUT_OF_ORDER);
        }
        return -1;
    }
 
    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);
    }
 
    return 0;
}

References AST_CAUSE_NETWORK_OUT_OF_ORDER, ast_channel_hangupcause_set(), ast_channel_name(), ast_channel_set_fd(), ast_channel_tech_hangupcause_set(), AST_CONTROL_ANSWER, ast_debug, ast_log, ast_queue_control(), ast_sockaddr_stringify(), ast_strdup, ast_websocket_fd(), ast_websocket_reconstruct_enable(), ast_websocket_remote_address(), ast_websocket_set_nonblock(), AST_WEBSOCKET_STATUS_GOING_AWAY, AST_WS_TYPE_SERVER, websocket_pvt::channel, errno, LOG_WARNING, websocket_pvt::no_auto_answer, websocket_pvt::remote_addr, send_event, websocket_pvt::type, websocket_pvt::websocket, websocket_request_hangup, and WS_WEBSOCKET_FDNO.

Referenced by incoming_ws_established_cb(), and webchan_call().

websocket_new()

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

static

Definition at line 1450 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 the passthrough and unbuffered modes in this case.
     */
    if (instance->native_codec->minimum_bytes <= 10) {
        instance->passthrough = 1;
        instance->unbuffered = 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 = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_CORE, .load = load_module, .unload = unload_module, .reload = reload_module, .load_pri = AST_MODPRI_CHANNEL_DRIVER, .requires = "res_http_websocket,res_websocket_client", }

static

Definition at line 2237 of file chan_websocket.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 2237 of file chan_websocket.c.

ast_ws_server

struct ast_websocket_server* ast_ws_server

static

Definition at line 87 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 2055 of file chan_websocket.c.

                                      {
    .callback = incoming_ws_http_callback,
    .description = "Media over Websocket",
    .uri = "media",
    .has_subtree = 1,
    .data = NULL,
    .key = __FILE__,
    .no_decode_uri = 1,
};

Referenced by load_module(), and unload_module().

instances

struct ao2_container* instances = NULL

static

Definition at line 89 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().

msg_format_map

const char* msg_format_map[]

static

Initial value:

= {
    [WEBCHAN_CONTROL_MSG_FORMAT_PLAIN] = "plain-text",
    [WEBCHAN_CONTROL_MSG_FORMAT_JSON] = "json",
    [WEBCHAN_CONTROL_MSG_FORMAT_INVALID] = "invalid",
}

Definition at line 62 of file chan_websocket.c.

                                      {
    [WEBCHAN_CONTROL_MSG_FORMAT_PLAIN] = "plain-text",
    [WEBCHAN_CONTROL_MSG_FORMAT_JSON] = "json",
    [WEBCHAN_CONTROL_MSG_FORMAT_INVALID] = "invalid",
};

Referenced by control_msg_format_from_str(), and control_msg_format_to_str().

sorcery

struct ast_sorcery* sorcery = NULL

static

Definition at line 54 of file chan_websocket.c.

Referenced by __ast_sorcery_apply_config(), __ast_sorcery_apply_default(), __ast_sorcery_apply_wizard_mapping(), __ast_sorcery_insert_wizard_mapping(), __ast_sorcery_object_field_register(), __ast_sorcery_object_register(), __ast_sorcery_object_type_insert_wizard(), __ast_sorcery_object_type_remove_wizard(), __ast_sorcery_open(), __ast_sorcery_remove_wizard_mapping(), alloc_and_initialize_sorcery(), alloc_and_initialize_sorcery(), alloc_and_initialize_sorcery(), apply_list_configuration(), as_config_load(), as_config_reload(), ast_ari_asterisk_delete_object(), ast_ari_asterisk_get_object(), ast_ari_asterisk_update_object(), ast_sip_destroy_sorcery_global(), ast_sip_initialize_sorcery_auth(), ast_sip_initialize_sorcery_domain_alias(), ast_sip_initialize_sorcery_global(), ast_sip_initialize_sorcery_location(), ast_sip_initialize_sorcery_transport(), ast_sorcery_alloc(), ast_sorcery_copy(), ast_sorcery_create(), ast_sorcery_delete(), ast_sorcery_diff(), ast_sorcery_force_reload(), ast_sorcery_force_reload_object(), ast_sorcery_get_module(), ast_sorcery_get_object_type(), ast_sorcery_get_wizard_mapping(), ast_sorcery_get_wizard_mapping_count(), ast_sorcery_instance_observer_add(), ast_sorcery_instance_observer_remove(), ast_sorcery_is_stale(), ast_sorcery_load(), ast_sorcery_load_object(), ast_sorcery_object_fields_register(), ast_sorcery_object_set_congestion_levels(), ast_sorcery_object_set_copy_handler(), ast_sorcery_object_set_diff_handler(), ast_sorcery_object_unregister(), ast_sorcery_objectset_apply(), ast_sorcery_objectset_create2(), ast_sorcery_objectset_json_create(), ast_sorcery_observer_add(), ast_sorcery_observer_remove(), ast_sorcery_ref(), ast_sorcery_reload(), ast_sorcery_reload_object(), ast_sorcery_retrieve_by_fields(), ast_sorcery_retrieve_by_id(), ast_sorcery_retrieve_by_prefix(), ast_sorcery_retrieve_by_regex(), ast_sorcery_update(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), bucket_file_wizard_create(), bucket_file_wizard_delete(), bucket_file_wizard_is_stale(), bucket_file_wizard_retrieve(), bucket_file_wizard_update(), bucket_http_wizard_retrieve_id(), bucket_wizard_create(), bucket_wizard_delete(), bucket_wizard_is_stale(), bucket_wizard_retrieve(), can_reuse_registration(), create_object(), deinitialize_sorcery(), deinitialize_sorcery(), global_loaded_observer(), handle_aor(), handle_auth(), handle_auths(), handle_endpoint(), handle_export_primitives(), handle_identify(), handle_phoneprov(), handle_registrations(), incoming_ws_established_cb(), instance_created_observer(), instance_destroying_observer(), load_config(), load_module(), memory_cache_full_update(), memory_cache_populate_external(), memory_cache_populate_internal(), memory_cache_stale_check(), memory_cache_stale_check_object(), memory_cache_stale_update_full(), memory_cache_stale_update_object(), mock_retrieve_id(), object_type_loaded_observer(), object_type_registered_observer(), profile_load(), profile_reload(), reload_module(), return_sorcery_object(), sorcery_astdb_create(), sorcery_astdb_filter_objectset(), sorcery_astdb_retrieve_fields(), sorcery_astdb_retrieve_fields_common(), sorcery_astdb_retrieve_id(), sorcery_astdb_retrieve_multiple(), sorcery_astdb_retrieve_prefix(), sorcery_astdb_retrieve_regex(), sorcery_astdb_update(), sorcery_config_internal_load(), sorcery_config_load(), sorcery_config_reload(), sorcery_config_retrieve_fields(), sorcery_config_retrieve_multiple(), sorcery_config_retrieve_prefix(), sorcery_config_retrieve_regex(), sorcery_destructor(), sorcery_function_read(), sorcery_is_configuration_met(), sorcery_is_explicit_name_met(), sorcery_memory_cache_ami_populate(), sorcery_memory_cache_create(), sorcery_memory_cache_load(), sorcery_memory_cache_populate(), sorcery_memory_cache_retrieve_fields(), sorcery_memory_cache_retrieve_id(), sorcery_memory_cache_retrieve_multiple(), sorcery_memory_cache_retrieve_prefix(), sorcery_memory_cache_retrieve_regex(), sorcery_memory_cached_object_alloc(), sorcery_memory_retrieve_fields(), sorcery_memory_retrieve_multiple(), sorcery_memory_retrieve_prefix(), sorcery_memory_retrieve_regex(), sorcery_realtime_create(), sorcery_realtime_filter_objectset(), sorcery_realtime_retrieve_fields(), sorcery_realtime_retrieve_id(), sorcery_realtime_retrieve_multiple(), sorcery_realtime_retrieve_prefix(), sorcery_realtime_retrieve_regex(), sorcery_realtime_update(), sorcery_reloadable(), sorcery_test_retrieve_id(), sorcery_wizard_load(), stale_cache_update_task_data_alloc(), stale_update_task_data_alloc(), tn_config_load(), tn_config_reload(), transport_apply(), unload_module(), vs_config_load(), vs_config_reload(), webchan_call(), webchan_request(), and wizard_apply_handler().

websocket_media_direction_map

const char* websocket_media_direction_map[]

static

Initial value:

= {
    [WEBCHAN_MEDIA_DIRECTION_BOTH] = "both",
    [WEBCHAN_MEDIA_DIRECTION_OUT] = "out",
    [WEBCHAN_MEDIA_DIRECTION_IN] = "in",
}

Definition at line 81 of file chan_websocket.c.

                                                     {
    [WEBCHAN_MEDIA_DIRECTION_BOTH] = "both",
    [WEBCHAN_MEDIA_DIRECTION_OUT] = "out",
    [WEBCHAN_MEDIA_DIRECTION_IN] = "in",
};

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 }, [ 'f' ] = { .flag = OPT_WS_MSG_FORMAT , .arg_index = OPT_ARG_WS_MSG_FORMAT + 1 }, [ 'd' ] = { .flag = OPT_WS_MEDIA_DIRECTION , .arg_index = OPT_ARG_WS_MEDIA_DIRECTION + 1 }, [ 'u' ] = { .flag = OPT_WS_UNBUFFERED }, }

static

Definition at line 1649 of file chan_websocket.c.

Referenced by webchan_request().

websocket_tech

struct ast_channel_tech websocket_tech

static

Definition at line 164 of file chan_websocket.c.

                                                {
    .type = "WebSocket",
    .description = "Media over WebSocket Channel Driver",
    .requester = webchan_request,
    .call = webchan_call,
    .read = webchan_read,
    .write = webchan_write,
    .hangup = webchan_hangup,
    .send_digit_end = webchan_send_dtmf_text,
};

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