mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-10-01 08:12:30 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/subprojects/gst-plugins-bad/ext/webrtc/webrtcdatachannel.c
Daniel Moberg 4c1de25e9d webrtc: do not tear down data channel before data is flushed 
Current implementation can in some cases detect
that all data is sent but in reality it is not,
leading to a push to an unlinked pad.
This is a race between the probe used to track data sent and a
call to close.

This patch sends an EOS before starting the close procedure
and then waits for the EOS event to come through to the
src pad before commencing with tear down.
This ensures that any queued data before EOS is flushed.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/4462>
2023-04-27 00:40:18 +00:00

1215 lines
37 KiB
C
Raw Blame History

/* GStreamer
* Copyright (C) 2018 Matthew Waters <matthew@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:gstwebrtc-datachannel
* @short_description: RTCDataChannel object
* @title: GstWebRTCDataChannel
* @see_also: #GstWebRTCRTPTransceiver
*
* <http://w3c.github.io/webrtc-pc/#dom-rtcsctptransport>
*/
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include "webrtcdatachannel.h"
#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <gst/base/gstbytereader.h>
#include <gst/base/gstbytewriter.h>
#include <gst/sctp/sctpreceivemeta.h>
#include <gst/sctp/sctpsendmeta.h>
#include "gstwebrtcbin.h"
#include "utils.h"
#define GST_CAT_DEFAULT webrtc_data_channel_debug
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
static void _close_procedure (WebRTCDataChannel * channel, gpointer user_data);
typedef void (*ChannelTask) (GstWebRTCDataChannel * channel,
gpointer user_data);
struct task
{
GstWebRTCDataChannel *channel;
ChannelTask func;
gpointer user_data;
GDestroyNotify notify;
};
static GstStructure *
_execute_task (GstWebRTCBin * webrtc, struct task *task)
{
if (task->func)
task->func (task->channel, task->user_data);
return NULL;
}
static void
_free_task (struct task *task)
{
gst_object_unref (task->channel);
if (task->notify)
task->notify (task->user_data);
g_free (task);
}
static void
_channel_enqueue_task (WebRTCDataChannel * channel, ChannelTask func,
gpointer user_data, GDestroyNotify notify)
{
struct task *task = g_new0 (struct task, 1);
task->channel = gst_object_ref (channel);
task->func = func;
task->user_data = user_data;
task->notify = notify;
gst_webrtc_bin_enqueue_task (channel->webrtcbin,
(GstWebRTCBinFunc) _execute_task, task, (GDestroyNotify) _free_task,
NULL);
}
static void
_channel_store_error (WebRTCDataChannel * channel, GError * error)
{
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (error) {
GST_WARNING_OBJECT (channel, "Error: %s",
error ? error->message : "Unknown");
if (!channel->stored_error)
channel->stored_error = error;
else
g_clear_error (&error);
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
struct _WebRTCErrorIgnoreBin
{
GstBin bin;
WebRTCDataChannel *data_channel;
};
G_DEFINE_TYPE (WebRTCErrorIgnoreBin, webrtc_error_ignore_bin, GST_TYPE_BIN);
static void
webrtc_error_ignore_bin_handle_message (GstBin * bin, GstMessage * message)
{
WebRTCErrorIgnoreBin *self = WEBRTC_ERROR_IGNORE_BIN (bin);
switch (GST_MESSAGE_TYPE (message)) {
case GST_MESSAGE_ERROR:{
GError *error = NULL;
gst_message_parse_error (message, &error, NULL);
GST_DEBUG_OBJECT (bin, "handling error message from internal element");
_channel_store_error (self->data_channel, error);
_channel_enqueue_task (self->data_channel, (ChannelTask) _close_procedure,
NULL, NULL);
break;
}
default:
GST_BIN_CLASS (webrtc_error_ignore_bin_parent_class)->handle_message (bin,
message);
break;
}
}
static void
webrtc_error_ignore_bin_class_init (WebRTCErrorIgnoreBinClass * klass)
{
GstBinClass *bin_class = (GstBinClass *) klass;
bin_class->handle_message = webrtc_error_ignore_bin_handle_message;
}
static void
webrtc_error_ignore_bin_init (WebRTCErrorIgnoreBin * bin)
{
}
static GstElement *
webrtc_error_ignore_bin_new (WebRTCDataChannel * data_channel,
GstElement * other)
{
WebRTCErrorIgnoreBin *self;
GstPad *pad;
self = g_object_new (webrtc_error_ignore_bin_get_type (), NULL);
self->data_channel = data_channel;
gst_bin_add (GST_BIN (self), other);
pad = gst_element_get_static_pad (other, "src");
if (pad) {
GstPad *ghost_pad = gst_ghost_pad_new ("src", pad);
gst_element_add_pad (GST_ELEMENT (self), ghost_pad);
gst_clear_object (&pad);
}
pad = gst_element_get_static_pad (other, "sink");
if (pad) {
GstPad *ghost_pad = gst_ghost_pad_new ("sink", pad);
gst_element_add_pad (GST_ELEMENT (self), ghost_pad);
gst_clear_object (&pad);
}
return (GstElement *) self;
}
#define webrtc_data_channel_parent_class parent_class
G_DEFINE_TYPE_WITH_CODE (WebRTCDataChannel, webrtc_data_channel,
GST_TYPE_WEBRTC_DATA_CHANNEL,
GST_DEBUG_CATEGORY_INIT (webrtc_data_channel_debug, "webrtcdatachannel", 0,
"webrtcdatachannel"););
G_LOCK_DEFINE_STATIC (outstanding_channels_lock);
static GList *outstanding_channels;
typedef enum
{
DATA_CHANNEL_PPID_WEBRTC_CONTROL = 50,
DATA_CHANNEL_PPID_WEBRTC_STRING = 51,
DATA_CHANNEL_PPID_WEBRTC_BINARY_PARTIAL = 52, /* deprecated */
DATA_CHANNEL_PPID_WEBRTC_BINARY = 53,
DATA_CHANNEL_PPID_WEBRTC_STRING_PARTIAL = 54, /* deprecated */
DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY = 56,
DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY = 57,
} DataChannelPPID;
typedef enum
{
CHANNEL_TYPE_RELIABLE = 0x00,
CHANNEL_TYPE_RELIABLE_UNORDERED = 0x80,
CHANNEL_TYPE_PARTIAL_RELIABLE_REXMIT = 0x01,
CHANNEL_TYPE_PARTIAL_RELIABLE_REXMIT_UNORDERED = 0x81,
CHANNEL_TYPE_PARTIAL_RELIABLE_TIMED = 0x02,
CHANNEL_TYPE_PARTIAL_RELIABLE_TIMED_UNORDERED = 0x82,
} DataChannelReliabilityType;
typedef enum
{
CHANNEL_MESSAGE_ACK = 0x02,
CHANNEL_MESSAGE_OPEN = 0x03,
} DataChannelMessage;
static guint16
priority_type_to_uint (GstWebRTCPriorityType pri)
{
switch (pri) {
case GST_WEBRTC_PRIORITY_TYPE_VERY_LOW:
return 64;
case GST_WEBRTC_PRIORITY_TYPE_LOW:
return 192;
case GST_WEBRTC_PRIORITY_TYPE_MEDIUM:
return 384;
case GST_WEBRTC_PRIORITY_TYPE_HIGH:
return 768;
}
g_assert_not_reached ();
return 0;
}
static GstWebRTCPriorityType
priority_uint_to_type (guint16 val)
{
if (val <= 128)
return GST_WEBRTC_PRIORITY_TYPE_VERY_LOW;
if (val <= 256)
return GST_WEBRTC_PRIORITY_TYPE_LOW;
if (val <= 512)
return GST_WEBRTC_PRIORITY_TYPE_MEDIUM;
return GST_WEBRTC_PRIORITY_TYPE_HIGH;
}
static GstBuffer *
construct_open_packet (WebRTCDataChannel * channel)
{
GstByteWriter w;
gsize label_len = strlen (channel->parent.label);
gsize proto_len = strlen (channel->parent.protocol);
gsize size = 12 + label_len + proto_len;
DataChannelReliabilityType reliability = 0;
guint32 reliability_param = 0;
guint16 priority;
GstBuffer *buf;
/*
* 0 1 2 3
* 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | Message Type | Channel Type | Priority |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | Reliability Parameter |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | Label Length | Protocol Length |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* \ /
* | Label |
* / \
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* \ /
* | Protocol |
* / \
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
gst_byte_writer_init_with_size (&w, size, FALSE);
if (!gst_byte_writer_put_uint8 (&w, (guint8) CHANNEL_MESSAGE_OPEN))
g_return_val_if_reached (NULL);
if (!channel->parent.ordered)
reliability |= 0x80;
if (channel->parent.max_retransmits != -1) {
reliability |= 0x01;
reliability_param = channel->parent.max_retransmits;
}
if (channel->parent.max_packet_lifetime != -1) {
reliability |= 0x02;
reliability_param = channel->parent.max_packet_lifetime;
}
priority = priority_type_to_uint (channel->parent.priority);
if (!gst_byte_writer_put_uint8 (&w, (guint8) reliability))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_uint16_be (&w, (guint16) priority))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_uint32_be (&w, (guint32) reliability_param))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_uint16_be (&w, (guint16) label_len))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_uint16_be (&w, (guint16) proto_len))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_data (&w, (guint8 *) channel->parent.label,
label_len))
g_return_val_if_reached (NULL);
if (!gst_byte_writer_put_data (&w, (guint8 *) channel->parent.protocol,
proto_len))
g_return_val_if_reached (NULL);
buf = gst_byte_writer_reset_and_get_buffer (&w);
/* send reliable and ordered */
gst_sctp_buffer_add_send_meta (buf, DATA_CHANNEL_PPID_WEBRTC_CONTROL, TRUE,
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE, 0);
return buf;
}
static GstBuffer *
construct_ack_packet (WebRTCDataChannel * channel)
{
GstByteWriter w;
GstBuffer *buf;
/*
* 0 1 2 3
* 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | Message Type |
* +-+-+-+-+-+-+-+-+
*/
gst_byte_writer_init_with_size (&w, 1, FALSE);
if (!gst_byte_writer_put_uint8 (&w, (guint8) CHANNEL_MESSAGE_ACK))
g_return_val_if_reached (NULL);
buf = gst_byte_writer_reset_and_get_buffer (&w);
/* send reliable and ordered */
gst_sctp_buffer_add_send_meta (buf, DATA_CHANNEL_PPID_WEBRTC_CONTROL, TRUE,
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE, 0);
return buf;
}
static void
_emit_on_open (WebRTCDataChannel * channel, gpointer user_data)
{
gst_webrtc_data_channel_on_open (GST_WEBRTC_DATA_CHANNEL (channel));
}
static void
_transport_closed (WebRTCDataChannel * channel)
{
GError *error;
gboolean both_sides_closed;
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
error = channel->stored_error;
channel->stored_error = NULL;
GST_TRACE_OBJECT (channel, "transport closed, peer closed %u error %p "
"buffered %" G_GUINT64_FORMAT, channel->peer_closed, error,
channel->parent.buffered_amount);
both_sides_closed =
channel->peer_closed && channel->parent.buffered_amount <= 0;
if (both_sides_closed || error) {
channel->peer_closed = FALSE;
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
if (error) {
gst_webrtc_data_channel_on_error (GST_WEBRTC_DATA_CHANNEL (channel), error);
g_clear_error (&error);
}
if (both_sides_closed || error) {
gst_webrtc_data_channel_on_close (GST_WEBRTC_DATA_CHANNEL (channel));
}
}
static void
_close_sctp_stream (WebRTCDataChannel * channel, gpointer user_data)
{
GstPad *pad, *peer;
GST_INFO_OBJECT (channel, "Closing outgoing SCTP stream %i label \"%s\"",
channel->parent.id, channel->parent.label);
pad = gst_element_get_static_pad (channel->src_bin, "src");
peer = gst_pad_get_peer (pad);
gst_object_unref (pad);
if (peer) {
GstElement *sctpenc = gst_pad_get_parent_element (peer);
if (sctpenc) {
GST_TRACE_OBJECT (channel, "removing sctpenc pad %" GST_PTR_FORMAT, peer);
gst_element_release_request_pad (sctpenc, peer);
gst_object_unref (sctpenc);
}
gst_object_unref (peer);
}
_transport_closed (channel);
}
static void
_close_procedure (WebRTCDataChannel * channel, gpointer user_data)
{
/* https://www.w3.org/TR/webrtc/#data-transport-closing-procedure */
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_CLOSED) {
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
return;
} else if (channel->parent.ready_state ==
GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING) {
_channel_enqueue_task (channel, (ChannelTask) _transport_closed, NULL,
NULL);
} else if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN) {
channel->parent.ready_state = GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING;
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_object_notify (G_OBJECT (channel), "ready-state");
/* Make sure that all data enqueued gets properly sent before data channel is closed. */
GstFlowReturn ret =
gst_app_src_end_of_stream (GST_APP_SRC (WEBRTC_DATA_CHANNEL
(channel)->appsrc));
if (ret != GST_FLOW_OK) {
GST_WARNING_OBJECT (channel, "Send end of stream returned %i, %s", ret,
gst_flow_get_name (ret));
}
return;
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
static void
_on_sctp_stream_reset (WebRTCSCTPTransport * sctp, guint stream_id,
WebRTCDataChannel * channel)
{
if (channel->parent.id == stream_id) {
GST_INFO_OBJECT (channel,
"Received channel close for SCTP stream %i label \"%s\"",
channel->parent.id, channel->parent.label);
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
channel->peer_closed = TRUE;
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure,
GUINT_TO_POINTER (stream_id), NULL);
}
}
static void
webrtc_data_channel_close (GstWebRTCDataChannel * channel)
{
_close_procedure (WEBRTC_DATA_CHANNEL (channel), NULL);
}
static GstFlowReturn
_parse_control_packet (WebRTCDataChannel * channel, guint8 * data,
gsize size, GError ** error)
{
GstByteReader r;
guint8 message_type;
gchar *label = NULL;
gchar *proto = NULL;
if (!data)
g_return_val_if_reached (GST_FLOW_ERROR);
if (size < 1)
g_return_val_if_reached (GST_FLOW_ERROR);
gst_byte_reader_init (&r, data, size);
if (!gst_byte_reader_get_uint8 (&r, &message_type))
g_return_val_if_reached (GST_FLOW_ERROR);
if (message_type == CHANNEL_MESSAGE_ACK) {
/* all good */
GST_INFO_OBJECT (channel, "Received channel ack");
return GST_FLOW_OK;
} else if (message_type == CHANNEL_MESSAGE_OPEN) {
guint8 reliability;
guint32 reliability_param;
guint16 priority, label_len, proto_len;
const guint8 *src;
GstBuffer *buffer;
GstFlowReturn ret;
GST_INFO_OBJECT (channel, "Received channel open");
if (channel->parent.negotiated) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Data channel was signalled as negotiated already");
g_return_val_if_reached (GST_FLOW_ERROR);
}
if (channel->opened)
return GST_FLOW_OK;
if (!gst_byte_reader_get_uint8 (&r, &reliability))
goto parse_error;
if (!gst_byte_reader_get_uint16_be (&r, &priority))
goto parse_error;
if (!gst_byte_reader_get_uint32_be (&r, &reliability_param))
goto parse_error;
if (!gst_byte_reader_get_uint16_be (&r, &label_len))
goto parse_error;
if (!gst_byte_reader_get_uint16_be (&r, &proto_len))
goto parse_error;
label = g_new0 (gchar, (gsize) label_len + 1);
proto = g_new0 (gchar, (gsize) proto_len + 1);
if (!gst_byte_reader_get_data (&r, label_len, &src))
goto parse_error;
memcpy (label, src, label_len);
label[label_len] = '\0';
if (!gst_byte_reader_get_data (&r, proto_len, &src))
goto parse_error;
memcpy (proto, src, proto_len);
proto[proto_len] = '\0';
g_free (channel->parent.label);
channel->parent.label = label;
g_free (channel->parent.protocol);
channel->parent.protocol = proto;
channel->parent.priority = priority_uint_to_type (priority);
channel->parent.ordered = !(reliability & 0x80);
if (reliability & 0x01) {
channel->parent.max_retransmits = reliability_param;
channel->parent.max_packet_lifetime = -1;
} else if (reliability & 0x02) {
channel->parent.max_retransmits = -1;
channel->parent.max_packet_lifetime = reliability_param;
} else {
channel->parent.max_retransmits = -1;
channel->parent.max_packet_lifetime = -1;
}
channel->opened = TRUE;
GST_INFO_OBJECT (channel, "Received channel open for SCTP stream %i "
"label \"%s\" protocol %s ordered %s", channel->parent.id,
channel->parent.label, channel->parent.protocol,
channel->parent.ordered ? "true" : "false");
_channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
GST_INFO_OBJECT (channel, "Sending channel ack");
buffer = construct_ack_packet (channel);
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
channel->parent.buffered_amount += gst_buffer_get_size (buffer);
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
if (ret != GST_FLOW_OK) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE, "Could not send ack packet");
GST_WARNING_OBJECT (channel, "push returned %i, %s", ret,
gst_flow_get_name (ret));
return ret;
}
return ret;
} else {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Unknown message type in control protocol");
return GST_FLOW_ERROR;
}
parse_error:
{
g_free (label);
g_free (proto);
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE, "Failed to parse packet");
g_return_val_if_reached (GST_FLOW_ERROR);
}
}
static void
on_sink_eos (GstAppSink * sink, gpointer user_data)
{
}
struct map_info
{
GstBuffer *buffer;
GstMapInfo map_info;
};
static void
buffer_unmap_and_unref (struct map_info *info)
{
gst_buffer_unmap (info->buffer, &info->map_info);
gst_buffer_unref (info->buffer);
g_free (info);
}
static void
_emit_have_data (WebRTCDataChannel * channel, GBytes * data)
{
gst_webrtc_data_channel_on_message_data (GST_WEBRTC_DATA_CHANNEL (channel),
data);
}
static void
_emit_have_string (GstWebRTCDataChannel * channel, gchar * str)
{
gst_webrtc_data_channel_on_message_string (GST_WEBRTC_DATA_CHANNEL (channel),
str);
}
static GstFlowReturn
_data_channel_have_sample (WebRTCDataChannel * channel, GstSample * sample,
GError ** error)
{
GstSctpReceiveMeta *receive;
GstBuffer *buffer;
GstFlowReturn ret = GST_FLOW_OK;
GST_LOG_OBJECT (channel, "Received sample %" GST_PTR_FORMAT, sample);
g_return_val_if_fail (channel->sctp_transport != NULL, GST_FLOW_ERROR);
buffer = gst_sample_get_buffer (sample);
if (!buffer) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE, "No buffer to handle");
return GST_FLOW_ERROR;
}
receive = gst_sctp_buffer_get_receive_meta (buffer);
if (!receive) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"No SCTP Receive meta on the buffer");
return GST_FLOW_ERROR;
}
switch (receive->ppid) {
case DATA_CHANNEL_PPID_WEBRTC_CONTROL:{
GstMapInfo info = GST_MAP_INFO_INIT;
if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Failed to map received buffer");
ret = GST_FLOW_ERROR;
} else {
ret = _parse_control_packet (channel, info.data, info.size, error);
gst_buffer_unmap (buffer, &info);
}
break;
}
case DATA_CHANNEL_PPID_WEBRTC_STRING:
case DATA_CHANNEL_PPID_WEBRTC_STRING_PARTIAL:{
GstMapInfo info = GST_MAP_INFO_INIT;
if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Failed to map received buffer");
ret = GST_FLOW_ERROR;
} else {
gchar *str = g_strndup ((gchar *) info.data, info.size);
_channel_enqueue_task (channel, (ChannelTask) _emit_have_string, str,
g_free);
gst_buffer_unmap (buffer, &info);
}
break;
}
case DATA_CHANNEL_PPID_WEBRTC_BINARY:
case DATA_CHANNEL_PPID_WEBRTC_BINARY_PARTIAL:{
struct map_info *info = g_new0 (struct map_info, 1);
if (!gst_buffer_map (buffer, &info->map_info, GST_MAP_READ)) {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Failed to map received buffer");
ret = GST_FLOW_ERROR;
} else {
GBytes *data = g_bytes_new_with_free_func (info->map_info.data,
info->map_info.size, (GDestroyNotify) buffer_unmap_and_unref, info);
info->buffer = gst_buffer_ref (buffer);
_channel_enqueue_task (channel, (ChannelTask) _emit_have_data, data,
(GDestroyNotify) g_bytes_unref);
}
break;
}
case DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY:
_channel_enqueue_task (channel, (ChannelTask) _emit_have_data, NULL,
NULL);
break;
case DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY:
_channel_enqueue_task (channel, (ChannelTask) _emit_have_string, NULL,
NULL);
break;
default:
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Unknown SCTP PPID %u received", receive->ppid);
ret = GST_FLOW_ERROR;
break;
}
return ret;
}
static GstFlowReturn
on_sink_preroll (GstAppSink * sink, gpointer user_data)
{
WebRTCDataChannel *channel = user_data;
GstSample *sample = gst_app_sink_pull_preroll (sink);
GstFlowReturn ret;
if (sample) {
/* This sample also seems to be provided by the sample callback
ret = _data_channel_have_sample (channel, sample); */
ret = GST_FLOW_OK;
gst_sample_unref (sample);
} else if (gst_app_sink_is_eos (sink)) {
ret = GST_FLOW_EOS;
} else {
ret = GST_FLOW_ERROR;
}
if (ret != GST_FLOW_OK) {
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
}
return ret;
}
static GstFlowReturn
on_sink_sample (GstAppSink * sink, gpointer user_data)
{
WebRTCDataChannel *channel = user_data;
GstSample *sample = gst_app_sink_pull_sample (sink);
GstFlowReturn ret;
GError *error = NULL;
if (sample) {
ret = _data_channel_have_sample (channel, sample, &error);
gst_sample_unref (sample);
} else if (gst_app_sink_is_eos (sink)) {
ret = GST_FLOW_EOS;
} else {
ret = GST_FLOW_ERROR;
}
if (error)
_channel_store_error (channel, error);
if (ret != GST_FLOW_OK) {
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
}
return ret;
}
static GstAppSinkCallbacks sink_callbacks = {
on_sink_eos,
on_sink_preroll,
on_sink_sample,
};
void
webrtc_data_channel_start_negotiation (WebRTCDataChannel * channel)
{
GstBuffer *buffer;
g_return_if_fail (!channel->parent.negotiated);
g_return_if_fail (channel->parent.id != -1);
g_return_if_fail (channel->sctp_transport != NULL);
buffer = construct_open_packet (channel);
GST_INFO_OBJECT (channel, "Sending channel open for SCTP stream %i "
"label \"%s\" protocol %s ordered %s", channel->parent.id,
channel->parent.label, channel->parent.protocol,
channel->parent.ordered ? "true" : "false");
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
channel->parent.buffered_amount += gst_buffer_get_size (buffer);
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_object_notify (G_OBJECT (&channel->parent), "buffered-amount");
if (gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc),
buffer) == GST_FLOW_OK) {
channel->opened = TRUE;
_channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
} else {
GError *error = NULL;
g_set_error (&error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE,
"Failed to send DCEP open packet");
_channel_store_error (channel, error);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
}
}
static void
_get_sctp_reliability (WebRTCDataChannel * channel,
GstSctpSendMetaPartiallyReliability * reliability, guint * rel_param)
{
if (channel->parent.max_retransmits != -1) {
*reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_RTX;
*rel_param = channel->parent.max_retransmits;
} else if (channel->parent.max_packet_lifetime != -1) {
*reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_TTL;
*rel_param = channel->parent.max_packet_lifetime;
} else {
*reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE;
*rel_param = 0;
}
}
static gboolean
_is_within_max_message_size (WebRTCDataChannel * channel, gsize size)
{
return size <= channel->sctp_transport->max_message_size;
}
static gboolean
webrtc_data_channel_send_data (GstWebRTCDataChannel * base_channel,
GBytes * bytes, GError ** error)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
GstSctpSendMetaPartiallyReliability reliability;
guint rel_param;
guint32 ppid;
GstBuffer *buffer;
gsize size = 0;
GstFlowReturn ret;
if (!bytes) {
buffer = gst_buffer_new ();
ppid = DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY;
} else {
guint8 *data;
data = (guint8 *) g_bytes_get_data (bytes, &size);
g_return_val_if_fail (data != NULL, FALSE);
if (!_is_within_max_message_size (channel, size)) {
g_set_error (error, GST_WEBRTC_ERROR, GST_WEBRTC_ERROR_TYPE_ERROR,
"Requested to send data that is too large");
return FALSE;
}
buffer = gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY, data, size,
0, size, g_bytes_ref (bytes), (GDestroyNotify) g_bytes_unref);
ppid = DATA_CHANNEL_PPID_WEBRTC_BINARY;
}
_get_sctp_reliability (channel, &reliability, &rel_param);
gst_sctp_buffer_add_send_meta (buffer, ppid, channel->parent.ordered,
reliability, rel_param);
GST_LOG_OBJECT (channel, "Sending data using buffer %" GST_PTR_FORMAT,
buffer);
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN) {
channel->parent.buffered_amount += size;
} else {
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_INVALID_STATE, "channel is not open");
gst_buffer_unref (buffer);
return FALSE;
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
if (ret == GST_FLOW_OK) {
g_object_notify (G_OBJECT (&channel->parent), "buffered-amount");
} else {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE, "Failed to send data");
GST_WARNING_OBJECT (channel, "push returned %i, %s", ret,
gst_flow_get_name (ret));
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
channel->parent.buffered_amount -= size;
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
return FALSE;
}
return TRUE;
}
static gboolean
webrtc_data_channel_send_string (GstWebRTCDataChannel * base_channel,
const gchar * str, GError ** error)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
GstSctpSendMetaPartiallyReliability reliability;
guint rel_param;
guint32 ppid;
GstBuffer *buffer;
gsize size = 0;
GstFlowReturn ret;
if (!channel->parent.negotiated)
g_return_val_if_fail (channel->opened, FALSE);
g_return_val_if_fail (channel->sctp_transport != NULL, FALSE);
if (!str) {
buffer = gst_buffer_new ();
ppid = DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY;
} else {
gchar *str_copy;
size = strlen (str);
if (!_is_within_max_message_size (channel, size)) {
g_set_error (error, GST_WEBRTC_ERROR, GST_WEBRTC_ERROR_TYPE_ERROR,
"Requested to send a string that is too large");
return FALSE;
}
str_copy = g_strdup (str);
buffer =
gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY, str_copy,
size, 0, size, str_copy, g_free);
ppid = DATA_CHANNEL_PPID_WEBRTC_STRING;
}
_get_sctp_reliability (channel, &reliability, &rel_param);
gst_sctp_buffer_add_send_meta (buffer, ppid, channel->parent.ordered,
reliability, rel_param);
GST_TRACE_OBJECT (channel, "Sending string using buffer %" GST_PTR_FORMAT,
buffer);
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN) {
channel->parent.buffered_amount += size;
} else {
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_INVALID_STATE, "channel is not open");
gst_buffer_unref (buffer);
return FALSE;
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
if (ret == GST_FLOW_OK) {
g_object_notify (G_OBJECT (&channel->parent), "buffered-amount");
} else {
g_set_error (error, GST_WEBRTC_ERROR,
GST_WEBRTC_ERROR_DATA_CHANNEL_FAILURE, "Failed to send string");
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
channel->parent.buffered_amount -= size;
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
return FALSE;
}
return TRUE;
}
static void
_on_sctp_notify_state_unlocked (GObject * sctp_transport,
WebRTCDataChannel * channel)
{
GstWebRTCSCTPTransportState state;
g_object_get (sctp_transport, "state", &state, NULL);
if (state == GST_WEBRTC_SCTP_TRANSPORT_STATE_CONNECTED) {
if (channel->parent.negotiated)
_channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
}
}
static WebRTCDataChannel *
ensure_channel_alive (WebRTCDataChannel * channel)
{
/* ghetto impl of, does the channel still exist?.
* Needed because g_signal_handler_disconnect*() will not disconnect any
* running functions and _finalize() implementation can complete and
* invalidate channel */
G_LOCK (outstanding_channels_lock);
if (g_list_find (outstanding_channels, channel)) {
g_object_ref (channel);
} else {
G_UNLOCK (outstanding_channels_lock);
return NULL;
}
G_UNLOCK (outstanding_channels_lock);
return channel;
}
static void
_on_sctp_notify_state (GObject * sctp_transport, GParamSpec * pspec,
WebRTCDataChannel * channel)
{
if (!(channel = ensure_channel_alive (channel)))
return;
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
_on_sctp_notify_state_unlocked (sctp_transport, channel);
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_object_unref (channel);
}
static void
_emit_low_threshold (WebRTCDataChannel * channel, gpointer user_data)
{
gst_webrtc_data_channel_on_buffered_amount_low (GST_WEBRTC_DATA_CHANNEL
(channel));
}
static GstPadProbeReturn
on_appsrc_data (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
WebRTCDataChannel *channel = user_data;
guint64 prev_amount;
guint64 size = 0;
if (GST_PAD_PROBE_INFO_TYPE (info) & (GST_PAD_PROBE_TYPE_BUFFER)) {
GstBuffer *buffer = GST_PAD_PROBE_INFO_BUFFER (info);
size = gst_buffer_get_size (buffer);
} else if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
GstBufferList *list = GST_PAD_PROBE_INFO_BUFFER_LIST (info);
size = gst_buffer_list_calculate_size (list);
} else if (GST_PAD_PROBE_INFO_TYPE (info) &
GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
if (GST_EVENT_TYPE (event) == GST_EVENT_EOS
&& channel->parent.ready_state ==
GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING) {
_channel_enqueue_task (channel, (ChannelTask) _close_sctp_stream, NULL,
NULL);
return GST_PAD_PROBE_DROP;
}
}
if (size > 0) {
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
prev_amount = channel->parent.buffered_amount;
channel->parent.buffered_amount -= size;
GST_TRACE_OBJECT (channel, "checking low-threshold: prev %"
G_GUINT64_FORMAT " low-threshold %" G_GUINT64_FORMAT " buffered %"
G_GUINT64_FORMAT, prev_amount,
channel->parent.buffered_amount_low_threshold,
channel->parent.buffered_amount);
if (prev_amount >= channel->parent.buffered_amount_low_threshold
&& channel->parent.buffered_amount <=
channel->parent.buffered_amount_low_threshold) {
_channel_enqueue_task (channel, (ChannelTask) _emit_low_threshold, NULL,
NULL);
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
g_object_notify (G_OBJECT (&channel->parent), "buffered-amount");
}
return GST_PAD_PROBE_OK;
}
static void
gst_webrtc_data_channel_constructed (GObject * object)
{
WebRTCDataChannel *channel;
GstPad *pad;
GstCaps *caps;
G_OBJECT_CLASS (parent_class)->constructed (object);
channel = WEBRTC_DATA_CHANNEL (object);
GST_DEBUG ("New channel %p constructed", channel);
caps = gst_caps_new_any ();
channel->appsrc = gst_element_factory_make ("appsrc", NULL);
gst_object_ref_sink (channel->appsrc);
pad = gst_element_get_static_pad (channel->appsrc, "src");
channel->src_probe = gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_DATA_BOTH,
(GstPadProbeCallback) on_appsrc_data, channel, NULL);
channel->src_bin = webrtc_error_ignore_bin_new (channel, channel->appsrc);
channel->appsink = gst_element_factory_make ("appsink", NULL);
gst_object_ref_sink (channel->appsink);
g_object_set (channel->appsink, "sync", FALSE, "async", FALSE, "caps", caps,
NULL);
gst_app_sink_set_callbacks (GST_APP_SINK (channel->appsink), &sink_callbacks,
channel, NULL);
channel->sink_bin = webrtc_error_ignore_bin_new (channel, channel->appsink);
gst_object_unref (pad);
gst_caps_unref (caps);
}
static void
gst_webrtc_data_channel_dispose (GObject * object)
{
G_LOCK (outstanding_channels_lock);
outstanding_channels = g_list_remove (outstanding_channels, object);
G_UNLOCK (outstanding_channels_lock);
G_OBJECT_CLASS (parent_class)->dispose (object);
}
static void
gst_webrtc_data_channel_finalize (GObject * object)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (object);
if (channel->src_probe) {
GstPad *pad = gst_element_get_static_pad (channel->appsrc, "src");
gst_pad_remove_probe (pad, channel->src_probe);
gst_object_unref (pad);
channel->src_probe = 0;
}
if (channel->sctp_transport)
g_signal_handlers_disconnect_by_data (channel->sctp_transport, channel);
g_clear_object (&channel->sctp_transport);
g_clear_object (&channel->appsrc);
g_clear_object (&channel->appsink);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
webrtc_data_channel_class_init (WebRTCDataChannelClass * klass)
{
GObjectClass *gobject_class = (GObjectClass *) klass;
GstWebRTCDataChannelClass *channel_class =
(GstWebRTCDataChannelClass *) klass;
gobject_class->constructed = gst_webrtc_data_channel_constructed;
gobject_class->dispose = gst_webrtc_data_channel_dispose;
gobject_class->finalize = gst_webrtc_data_channel_finalize;
channel_class->send_data = webrtc_data_channel_send_data;
channel_class->send_string = webrtc_data_channel_send_string;
channel_class->close = webrtc_data_channel_close;
}
static void
webrtc_data_channel_init (WebRTCDataChannel * channel)
{
G_LOCK (outstanding_channels_lock);
outstanding_channels = g_list_prepend (outstanding_channels, channel);
G_UNLOCK (outstanding_channels_lock);
}
static void
_data_channel_set_sctp_transport (WebRTCDataChannel * channel,
WebRTCSCTPTransport * sctp)
{
g_return_if_fail (GST_IS_WEBRTC_DATA_CHANNEL (channel));
g_return_if_fail (GST_IS_WEBRTC_SCTP_TRANSPORT (sctp));
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (channel->sctp_transport)
g_signal_handlers_disconnect_by_data (channel->sctp_transport, channel);
GST_TRACE_OBJECT (channel, "set sctp %p", sctp);
gst_object_replace ((GstObject **) & channel->sctp_transport,
GST_OBJECT (sctp));
if (sctp) {
g_signal_connect (sctp, "stream-reset", G_CALLBACK (_on_sctp_stream_reset),
channel);
g_signal_connect (sctp, "notify::state", G_CALLBACK (_on_sctp_notify_state),
channel);
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
void
webrtc_data_channel_link_to_sctp (WebRTCDataChannel * channel,
WebRTCSCTPTransport * sctp_transport)
{
if (sctp_transport && !channel->sctp_transport) {
gint id;
g_object_get (channel, "id", &id, NULL);
if (sctp_transport->association_established && id != -1) {
gchar *pad_name;
_data_channel_set_sctp_transport (channel, sctp_transport);
pad_name = g_strdup_printf ("sink_%u", id);
if (!gst_element_link_pads (channel->src_bin, "src",
channel->sctp_transport->sctpenc, pad_name))
g_warn_if_reached ();
g_free (pad_name);
_on_sctp_notify_state_unlocked (G_OBJECT (sctp_transport), channel);
}
}
}
Reference in a new issue View git blame Copy permalink