diff --git a/channels/chan_websocket.c b/channels/chan_websocket.c
index 3d6292288c..724c9263fb 100644
--- a/channels/chan_websocket.c
+++ b/channels/chan_websocket.c
@@ -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)
diff --git a/channels/chan_websocket_doc.xml b/channels/chan_websocket_doc.xml
index 63e171a115..ca836a026e 100644
--- a/channels/chan_websocket_doc.xml
+++ b/channels/chan_websocket_doc.xml
@@ -108,6 +108,20 @@
automatically set for the opus, speex and g729 codecs.
+
+ 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.
+
+
This option allows you to add additional parameters to the
outbound URI. The format is: