mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-09-19 18:40:31 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/ext/webrtc/webrtcdatachannel.c
Johan Sternerup 8dbdfad914 webrtcbin: Support closing of data channels 
Support for closing WebRTC data channels as described in RFC
8831 (section 6.7) now fully supported. This means that we can now
reuse data channels that have been closed properly. Previously, an
application that created a lot of short-lived on-demand data channels
would quickly exhaust resources held by lingering non-closed data
channels.

We now use a one-to-one style socket interface to SCTP just like the
Google implementation (i.e. SOCK_STREAM instead of SOCK_SEQPACKET, see
RFC 6458). For some reason the socket interface to use was made
optional through a property "use-sock-stream" even though code wasn't
written to handle the SOCK_SEQPACKET style. Specifically the
SCTP_RESET_STREAMS command wouldn't work without passing the correct
assocation id. Changing the default interface to use from
SOCK_SEQPACKET to SOCK_STREAM now means we don't have to bother about
the association id as there is only one association per socket. For
the SCTP_RESET_STREAMS command we set it to SCTP_ALL_ASSOC just to
match the Google implementation.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/2186>
2021-05-12 03:02:27 +00:00

1048 lines
32 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);
#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"););
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;
}
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);
}
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;
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->appsrc, "src");
peer = gst_pad_get_peer (pad);
gst_object_unref (pad);
if (peer) {
GstElement *sctpenc = gst_pad_get_parent_element (peer);
if (sctpenc) {
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");
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
if (channel->parent.buffered_amount <= 0) {
_channel_enqueue_task (channel, (ChannelTask) _close_sctp_stream,
NULL, NULL);
}
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
static void
_on_sctp_stream_reset (GstWebRTCSCTPTransport * 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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
"Could not send ack packet");
return ret;
}
return ret;
} else {
g_set_error (error, GST_WEBRTC_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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_BIN_ERROR,
GST_WEBRTC_BIN_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);
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_BIN_ERROR,
GST_WEBRTC_BIN_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 void
webrtc_data_channel_send_data (GstWebRTCDataChannel * base_channel,
GBytes * bytes)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
GstSctpSendMetaPartiallyReliability reliability;
guint rel_param;
guint32 ppid;
GstBuffer *buffer;
GstFlowReturn ret;
if (!bytes) {
buffer = gst_buffer_new ();
ppid = DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY;
} else {
gsize size;
guint8 *data;
data = (guint8 *) g_bytes_get_data (bytes, &size);
g_return_if_fail (data != NULL);
if (!_is_within_max_message_size (channel, size)) {
GError *error = NULL;
g_set_error (&error, GST_WEBRTC_BIN_ERROR,
GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
"Requested to send data that is too large");
_channel_store_error (channel, error);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL,
NULL);
return;
}
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);
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) {
GError *error = NULL;
g_set_error (&error, GST_WEBRTC_BIN_ERROR,
GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "Failed to send data");
_channel_store_error (channel, error);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
}
}
static void
webrtc_data_channel_send_string (GstWebRTCDataChannel * base_channel,
const gchar * str)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
GstSctpSendMetaPartiallyReliability reliability;
guint rel_param;
guint32 ppid;
GstBuffer *buffer;
GstFlowReturn ret;
if (!channel->parent.negotiated)
g_return_if_fail (channel->opened);
g_return_if_fail (channel->sctp_transport != NULL);
if (!str) {
buffer = gst_buffer_new ();
ppid = DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY;
} else {
gsize size = strlen (str);
gchar *str_copy = g_strdup (str);
if (!_is_within_max_message_size (channel, size)) {
GError *error = NULL;
g_set_error (&error, GST_WEBRTC_BIN_ERROR,
GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
"Requested to send a string that is too large");
_channel_store_error (channel, error);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL,
NULL);
return;
}
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);
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) {
GError *error = NULL;
g_set_error (&error, GST_WEBRTC_BIN_ERROR,
GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "Failed to send string");
_channel_store_error (channel, error);
_channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
}
}
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 void
_on_sctp_notify_state (GObject * sctp_transport, GParamSpec * pspec,
WebRTCDataChannel * channel)
{
GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
_on_sctp_notify_state_unlocked (sctp_transport, channel);
GST_WEBRTC_DATA_CHANNEL_UNLOCK (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);
}
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);
}
if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING
&& channel->parent.buffered_amount <= 0) {
_channel_enqueue_task (channel, (ChannelTask) _close_sctp_stream, NULL,
NULL);
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
return GST_PAD_PROBE_OK;
}
static void
gst_webrtc_data_channel_constructed (GObject * object)
{
WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (object);
GstPad *pad;
GstCaps *caps;
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->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);
gst_object_unref (pad);
gst_caps_unref (caps);
}
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->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)
{
}
static void
_data_channel_set_sctp_transport (WebRTCDataChannel * channel,
GstWebRTCSCTPTransport * 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_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);
_on_sctp_notify_state_unlocked (G_OBJECT (sctp), channel);
}
GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
}
void
webrtc_data_channel_link_to_sctp (WebRTCDataChannel * channel,
GstWebRTCSCTPTransport * 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->appsrc, "src",
channel->sctp_transport->sctpenc, pad_name))
g_warn_if_reached ();
g_free (pad_name);
}
}
}
Reference in a new issue View git blame Copy permalink