Asterisk
/
asterisk
mirror of https://github.com/asterisk/asterisk
1
0
Fork 0
Code Issues Releases Wiki Activity

chan_websocket: Add "unbuffered" mode and fix a few bugs.

Browse Source 
Besides the User and Developer notes below, checks have been added to
reject a START_MEDIA_BUFFERING buffering command if it's already
active and to reject a STOP_MEDIA_BUFFERING command if a corresponding
START_MEDIA_BUFFERING command was never issued.

Resolves: #2115

UserNote: A new `unbuffered` chan_websocket mode has been added which
sends frames to the core as they're received from the WebSocket and your
app. It's similar to `passthrough` mode where your app is responsible
for proper framing and timing except that, assuming you're not using an
unbufferable codec (opus, speex, g729), in unbuffered mode, you can
still use START_MEDIA_BUFFERING to send bulk media and have
chan_websocket frame and time it. When you send STOP_MEDIA_BUFFERING,
the channel will go back into unbuffered mode. Unbuffered mode is most
useful when the media your app normally sends to Asterisk is already
properly framed and timed but you may occasionally need to send bulk
media. The mode is set with the `u` dialstring option.

DeveloperNote: Two new fields have been added to the chan_websocket
MEDIA_START event for both the plain text and JSON formats:
    `passthrough: true/false`
    `unbuffered: true/false`
These indicate the mode the channel was created with and correspond
to the `p` and `u` dialstring options.

DeveloperNote: The MARK_MEDIA chan_websocket command can now be used
by an app when in "passthrough" mode.
pull/1848/merge
George Joseph 3 weeks ago
parent 5eb9c9ce25
commit e80ed05877
2 changed files with 101 additions and 22 deletions
Show Stats Download Patch File Download Diff File

109
channels/chan_websocket.c
Unescape View File

@ -105,6 +105,8 @@ struct websocket_pvt {
enum webchan_control_msg_format control_msg_format;
int no_auto_answer;
int passthrough;
int unbuffered;
int last_unbuffered;
int optimal_frame_size;
int bulk_media_in_progress;
int report_queue_drained;
@ -228,7 +230,7 @@ static char *_create_event_MEDIA_START(struct websocket_pvt *instance)
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:o }",
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),
@ -236,6 +238,8 @@ static char *_create_event_MEDIA_START(struct websocket_pvt *instance)
"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))
);
@ -245,14 +249,16 @@ static char *_create_event_MEDIA_START(struct websocket_pvt *instance)
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",
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
"ptime", instance->native_codec->default_ms,
"passthrough", instance->passthrough ? "true" : "false",
"unbuffered", instance->unbuffered ? "true" : "false"
);
}
@ -456,11 +462,13 @@ static __attribute__ ((format (gnu_printf, 2, 3))) char *_create_event_ERROR(
(_res); \
})
/*
* Reminder... This function gets called by webchan_read which is
* triggered by the channel timer firing. It always gets called
* every 20ms (or whatever the timer is set to) even if there are
* no frames in the queue.
/*!
* \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)
{
@ -593,8 +601,18 @@ static struct ast_frame *create_frame_from_buffer(struct websocket_pvt *instance
*
* 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. We'll return a null frame.
* 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)
{
@ -607,16 +625,19 @@ static struct ast_frame *webchan_read(struct ast_channel *ast)
return NULL;
}
if (fdno == WS_WEBSOCKET_FDNO) {
read_from_ws_and_queue(instance);
return &ast_null_frame;
}
if (fdno != WS_TIMER_FDNO) {
if (fdno != WS_TIMER_FDNO && fdno != WS_WEBSOCKET_FDNO) {
return &ast_null_frame;
}
if (ast_timer_get_event(instance->timer) == AST_TIMING_EVENT_EXPIRED) {
ast_timer_ack(instance->timer, 1);
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);
@ -624,18 +645,18 @@ static struct ast_frame *webchan_read(struct ast_channel *ast)
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: WebSocket read timer fired with no frame available but with %d bytes in leftover_data. Returning partial frame.\n",
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: WebSocket read timer fired with no frame available and no data in leftover_data. Returning NULL frame.\n",
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: WebSocket read timer fired. Dequeued %d byte frame. Left in buffer: %d\n",
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;
@ -705,6 +726,16 @@ static int queue_option_frame(struct websocket_pvt *instance,
} \
})
#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) { \
@ -759,8 +790,15 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
} 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)) {
@ -778,11 +816,17 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
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) {
@ -803,7 +847,6 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
SCOPED_LOCK(frame_queue_lock, &instance->frame_queue, AST_LIST_LOCK,
AST_LIST_UNLOCK);
ERROR_ON_PASSTHROUGH_MODE_RTN(instance, command);
ERROR_ON_INVALID_MEDIA_DIRECTION_RTN(instance, command, WEBCHAN_MEDIA_DIRECTION_IN);
if (instance->control_msg_format == WEBCHAN_CONTROL_MSG_FORMAT_JSON) {
@ -826,6 +869,7 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
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))) {
@ -834,6 +878,7 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
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)) {
@ -848,6 +893,7 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
} 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;
@ -855,6 +901,7 @@ static int handle_command(struct websocket_pvt *instance, char *buffer)
} 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;
@ -982,7 +1029,7 @@ static int process_binary_message(struct websocket_pvt *instance,
next_frame_ptr = payload;
instance->bytes_read += payload_len;
if (instance->passthrough) {
if (instance->unbuffered) {
res = queue_frame_from_buffer(instance, payload, payload_len);
return res;
}
@ -1421,10 +1468,11 @@ static struct websocket_pvt* websocket_new(const char *chan_name,
* It's not possible for us to re-time or re-frame media if the data
* stream can't be broken up on arbitrary byte boundaries. This is usually
* indicated by the codec's minimum_bytes being small (10 bytes or less).
* We need to force passthrough mode in this case.
* 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 =
@ -1537,6 +1585,7 @@ enum {
OPT_WS_PASSTHROUGH = (1 << 3),
OPT_WS_MSG_FORMAT = (1 << 4),
OPT_WS_MEDIA_DIRECTION = (1 << 5),
OPT_WS_UNBUFFERED = (1 << 6),
};
enum {
@ -1546,6 +1595,7 @@ enum {
OPT_ARG_WS_PASSTHROUGH,
OPT_ARG_WS_MSG_FORMAT,
OPT_ARG_WS_MEDIA_DIRECTION,
OPT_ARG_WS_UNBUFFERED,
OPT_ARG_ARRAY_SIZE
};
@ -1556,6 +1606,7 @@ AST_APP_OPTIONS(websocket_options, BEGIN_OPTIONS
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,
@ -1649,8 +1700,22 @@ static struct ast_channel *webchan_request(const char *type,
}
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)

14
channels/chan_websocket_doc.xml
Unescape View File

@ -108,6 +108,20 @@
automatically set for the opus, speex and g729 codecs.
</para>
</enum>
<enum name="u - Unbuffered mode">
<para>In unbuffered mode, the channel driver won't attempt
to re-frame or re-time media coming in over the websocket from
the remote app. In this case, the remote app is fully responsible
for correctly framing and timing media sent to Asterisk. This is
useful for when the app is getting its media from a source that's
already framed and timed. The difference between unbuffered and
passthrough modes is that, if you're not using an unbufferable
codec (opus, speex and g729), you can still use START_MEDIA_BUFFERING
to switch to buffered mode to send a file, then when you call
STOP_MEDIA_BUFFERING, the channel will go back into unbuffered
mode.
</para>
</enum>
<enum name="v(uri_parameters) - Add parameters to the outbound URI">
<para>This option allows you to add additional parameters to the
outbound URI. The format is:

Write Preview
Loading…
Cancel
Save