subprojects/gst-plugins-bad/ext/webrtc/webrtcdatachannel.c

   1 /* GStreamer
   2  * Copyright (C) 2018 Matthew Waters <matthew@centricular.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19
  20 /**
  21  * SECTION:gstwebrtc-datachannel
  22  * @short_description: RTCDataChannel object
  23  * @title: GstWebRTCDataChannel
  24  * @see_also: #GstWebRTCRTPTransceiver
  25  *
  26  * <http://w3c.github.io/webrtc-pc/#dom-rtcsctptransport>
  27  */
  28
  29 #ifdef HAVE_CONFIG_H
  30 # include "config.h"
  31 #endif
  32
  33 #include "webrtcdatachannel.h"
  34 #include <gst/app/gstappsink.h>
  35 #include <gst/app/gstappsrc.h>
  36 #include <gst/base/gstbytereader.h>
  37 #include <gst/base/gstbytewriter.h>
  38 #include <gst/sctp/sctpreceivemeta.h>
  39 #include <gst/sctp/sctpsendmeta.h>
  40
  41 #include "gstwebrtcbin.h"
  42 #include "utils.h"
  43
  44 #define GST_CAT_DEFAULT webrtc_data_channel_debug
  45 GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
  46
  47 #define webrtc_data_channel_parent_class parent_class
  48 G_DEFINE_TYPE_WITH_CODE (WebRTCDataChannel, webrtc_data_channel,
  49     GST_TYPE_WEBRTC_DATA_CHANNEL,
  50     GST_DEBUG_CATEGORY_INIT (webrtc_data_channel_debug, "webrtcdatachannel", 0,
  51         "webrtcdatachannel"););
  52
  53 typedef enum
  54 {
  55   DATA_CHANNEL_PPID_WEBRTC_CONTROL = 50,
  56   DATA_CHANNEL_PPID_WEBRTC_STRING = 51,
  57   DATA_CHANNEL_PPID_WEBRTC_BINARY_PARTIAL = 52, /* deprecated */
  58   DATA_CHANNEL_PPID_WEBRTC_BINARY = 53,
  59   DATA_CHANNEL_PPID_WEBRTC_STRING_PARTIAL = 54, /* deprecated */
  60   DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY = 56,
  61   DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY = 57,
  62 } DataChannelPPID;
  63
  64 typedef enum
  65 {
  66   CHANNEL_TYPE_RELIABLE = 0x00,
  67   CHANNEL_TYPE_RELIABLE_UNORDERED = 0x80,
  68   CHANNEL_TYPE_PARTIAL_RELIABLE_REXMIT = 0x01,
  69   CHANNEL_TYPE_PARTIAL_RELIABLE_REXMIT_UNORDERED = 0x81,
  70   CHANNEL_TYPE_PARTIAL_RELIABLE_TIMED = 0x02,
  71   CHANNEL_TYPE_PARTIAL_RELIABLE_TIMED_UNORDERED = 0x82,
  72 } DataChannelReliabilityType;
  73
  74 typedef enum
  75 {
  76   CHANNEL_MESSAGE_ACK = 0x02,
  77   CHANNEL_MESSAGE_OPEN = 0x03,
  78 } DataChannelMessage;
  79
  80 static guint16
  81 priority_type_to_uint (GstWebRTCPriorityType pri)
  82 {
  83   switch (pri) {
  84     case GST_WEBRTC_PRIORITY_TYPE_VERY_LOW:
  85       return 64;
  86     case GST_WEBRTC_PRIORITY_TYPE_LOW:
  87       return 192;
  88     case GST_WEBRTC_PRIORITY_TYPE_MEDIUM:
  89       return 384;
  90     case GST_WEBRTC_PRIORITY_TYPE_HIGH:
  91       return 768;
  92   }
  93   g_assert_not_reached ();
  94   return 0;
  95 }
  96
  97 static GstWebRTCPriorityType
  98 priority_uint_to_type (guint16 val)
  99 {
 100   if (val <= 128)
 101     return GST_WEBRTC_PRIORITY_TYPE_VERY_LOW;
 102   if (val <= 256)
 103     return GST_WEBRTC_PRIORITY_TYPE_LOW;
 104   if (val <= 512)
 105     return GST_WEBRTC_PRIORITY_TYPE_MEDIUM;
 106   return GST_WEBRTC_PRIORITY_TYPE_HIGH;
 107 }
 108
 109 static GstBuffer *
 110 construct_open_packet (WebRTCDataChannel * channel)
 111 {
 112   GstByteWriter w;
 113   gsize label_len = strlen (channel->parent.label);
 114   gsize proto_len = strlen (channel->parent.protocol);
 115   gsize size = 12 + label_len + proto_len;
 116   DataChannelReliabilityType reliability = 0;
 117   guint32 reliability_param = 0;
 118   guint16 priority;
 119   GstBuffer *buf;
 120
 121 /*
 122  *    0                   1                   2                   3
 123  *    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
 124  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 125  *   |  Message Type |  Channel Type |            Priority           |
 126  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 127  *   |                    Reliability Parameter                      |
 128  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 129  *   |         Label Length          |       Protocol Length         |
 130  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 131  *   \                                                               /
 132  *   |                             Label                             |
 133  *   /                                                               \
 134  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 135  *   \                                                               /
 136  *   |                            Protocol                           |
 137  *   /                                                               \
 138  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 139  */
 140
 141   gst_byte_writer_init_with_size (&w, size, FALSE);
 142
 143   if (!gst_byte_writer_put_uint8 (&w, (guint8) CHANNEL_MESSAGE_OPEN))
 144     g_return_val_if_reached (NULL);
 145
 146   if (!channel->parent.ordered)
 147     reliability |= 0x80;
 148   if (channel->parent.max_retransmits != -1) {
 149     reliability |= 0x01;
 150     reliability_param = channel->parent.max_retransmits;
 151   }
 152   if (channel->parent.max_packet_lifetime != -1) {
 153     reliability |= 0x02;
 154     reliability_param = channel->parent.max_packet_lifetime;
 155   }
 156
 157   priority = priority_type_to_uint (channel->parent.priority);
 158
 159   if (!gst_byte_writer_put_uint8 (&w, (guint8) reliability))
 160     g_return_val_if_reached (NULL);
 161   if (!gst_byte_writer_put_uint16_be (&w, (guint16) priority))
 162     g_return_val_if_reached (NULL);
 163   if (!gst_byte_writer_put_uint32_be (&w, (guint32) reliability_param))
 164     g_return_val_if_reached (NULL);
 165   if (!gst_byte_writer_put_uint16_be (&w, (guint16) label_len))
 166     g_return_val_if_reached (NULL);
 167   if (!gst_byte_writer_put_uint16_be (&w, (guint16) proto_len))
 168     g_return_val_if_reached (NULL);
 169   if (!gst_byte_writer_put_data (&w, (guint8 *) channel->parent.label,
 170           label_len))
 171     g_return_val_if_reached (NULL);
 172   if (!gst_byte_writer_put_data (&w, (guint8 *) channel->parent.protocol,
 173           proto_len))
 174     g_return_val_if_reached (NULL);
 175
 176   buf = gst_byte_writer_reset_and_get_buffer (&w);
 177
 178   /* send reliable and ordered */
 179   gst_sctp_buffer_add_send_meta (buf, DATA_CHANNEL_PPID_WEBRTC_CONTROL, TRUE,
 180       GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE, 0);
 181
 182   return buf;
 183 }
 184
 185 static GstBuffer *
 186 construct_ack_packet (WebRTCDataChannel * channel)
 187 {
 188   GstByteWriter w;
 189   GstBuffer *buf;
 190
 191 /*
 192  *   0                   1                   2                   3
 193  *   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
 194  *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 195  *   |  Message Type |
 196  *   +-+-+-+-+-+-+-+-+
 197  */
 198
 199   gst_byte_writer_init_with_size (&w, 1, FALSE);
 200
 201   if (!gst_byte_writer_put_uint8 (&w, (guint8) CHANNEL_MESSAGE_ACK))
 202     g_return_val_if_reached (NULL);
 203
 204   buf = gst_byte_writer_reset_and_get_buffer (&w);
 205
 206   /* send reliable and ordered */
 207   gst_sctp_buffer_add_send_meta (buf, DATA_CHANNEL_PPID_WEBRTC_CONTROL, TRUE,
 208       GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE, 0);
 209
 210   return buf;
 211 }
 212
 213 typedef void (*ChannelTask) (GstWebRTCDataChannel * channel,
 214     gpointer user_data);
 215
 216 struct task
 217 {
 218   GstWebRTCDataChannel *channel;
 219   ChannelTask func;
 220   gpointer user_data;
 221   GDestroyNotify notify;
 222 };
 223
 224 static GstStructure *
 225 _execute_task (GstWebRTCBin * webrtc, struct task *task)
 226 {
 227   if (task->func)
 228     task->func (task->channel, task->user_data);
 229
 230   return NULL;
 231 }
 232
 233 static void
 234 _free_task (struct task *task)
 235 {
 236   gst_object_unref (task->channel);
 237
 238   if (task->notify)
 239     task->notify (task->user_data);
 240   g_free (task);
 241 }
 242
 243 static void
 244 _channel_enqueue_task (WebRTCDataChannel * channel, ChannelTask func,
 245     gpointer user_data, GDestroyNotify notify)
 246 {
 247   struct task *task = g_new0 (struct task, 1);
 248
 249   task->channel = gst_object_ref (channel);
 250   task->func = func;
 251   task->user_data = user_data;
 252   task->notify = notify;
 253
 254   gst_webrtc_bin_enqueue_task (channel->webrtcbin,
 255       (GstWebRTCBinFunc) _execute_task, task, (GDestroyNotify) _free_task,
 256       NULL);
 257 }
 258
 259 static void
 260 _channel_store_error (WebRTCDataChannel * channel, GError * error)
 261 {
 262   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 263   if (error) {
 264     GST_WARNING_OBJECT (channel, "Error: %s",
 265         error ? error->message : "Unknown");
 266     if (!channel->stored_error)
 267       channel->stored_error = error;
 268     else
 269       g_clear_error (&error);
 270   }
 271   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 272 }
 273
 274 static void
 275 _emit_on_open (WebRTCDataChannel * channel, gpointer user_data)
 276 {
 277   gst_webrtc_data_channel_on_open (GST_WEBRTC_DATA_CHANNEL (channel));
 278 }
 279
 280 static void
 281 _transport_closed (WebRTCDataChannel * channel)
 282 {
 283   GError *error;
 284   gboolean both_sides_closed;
 285
 286   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 287   error = channel->stored_error;
 288   channel->stored_error = NULL;
 289
 290   both_sides_closed =
 291       channel->peer_closed && channel->parent.buffered_amount <= 0;
 292   if (both_sides_closed || error) {
 293     channel->peer_closed = FALSE;
 294   }
 295   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 296
 297   if (error) {
 298     gst_webrtc_data_channel_on_error (GST_WEBRTC_DATA_CHANNEL (channel), error);
 299     g_clear_error (&error);
 300   }
 301   if (both_sides_closed || error) {
 302     gst_webrtc_data_channel_on_close (GST_WEBRTC_DATA_CHANNEL (channel));
 303   }
 304 }
 305
 306 static void
 307 _close_sctp_stream (WebRTCDataChannel * channel, gpointer user_data)
 308 {
 309   GstPad *pad, *peer;
 310
 311   GST_INFO_OBJECT (channel, "Closing outgoing SCTP stream %i label \"%s\"",
 312       channel->parent.id, channel->parent.label);
 313
 314   pad = gst_element_get_static_pad (channel->appsrc, "src");
 315   peer = gst_pad_get_peer (pad);
 316   gst_object_unref (pad);
 317
 318   if (peer) {
 319     GstElement *sctpenc = gst_pad_get_parent_element (peer);
 320
 321     if (sctpenc) {
 322       gst_element_release_request_pad (sctpenc, peer);
 323       gst_object_unref (sctpenc);
 324     }
 325     gst_object_unref (peer);
 326   }
 327
 328   _transport_closed (channel);
 329 }
 330
 331 static void
 332 _close_procedure (WebRTCDataChannel * channel, gpointer user_data)
 333 {
 334   /* https://www.w3.org/TR/webrtc/#data-transport-closing-procedure */
 335   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 336   if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_CLOSED) {
 337     GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 338     return;
 339   } else if (channel->parent.ready_state ==
 340       GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING) {
 341     _channel_enqueue_task (channel, (ChannelTask) _transport_closed, NULL,
 342         NULL);
 343   } else if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN) {
 344     channel->parent.ready_state = GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING;
 345     GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 346     g_object_notify (G_OBJECT (channel), "ready-state");
 347
 348     GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 349     if (channel->parent.buffered_amount <= 0) {
 350       _channel_enqueue_task (channel, (ChannelTask) _close_sctp_stream,
 351           NULL, NULL);
 352     }
 353   }
 354
 355   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 356 }
 357
 358 static void
 359 _on_sctp_stream_reset (WebRTCSCTPTransport * sctp, guint stream_id,
 360     WebRTCDataChannel * channel)
 361 {
 362   if (channel->parent.id == stream_id) {
 363     GST_INFO_OBJECT (channel,
 364         "Received channel close for SCTP stream %i label \"%s\"",
 365         channel->parent.id, channel->parent.label);
 366
 367     GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 368     channel->peer_closed = TRUE;
 369     GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 370
 371     _channel_enqueue_task (channel, (ChannelTask) _close_procedure,
 372         GUINT_TO_POINTER (stream_id), NULL);
 373   }
 374 }
 375
 376 static void
 377 webrtc_data_channel_close (GstWebRTCDataChannel * channel)
 378 {
 379   _close_procedure (WEBRTC_DATA_CHANNEL (channel), NULL);
 380 }
 381
 382 static GstFlowReturn
 383 _parse_control_packet (WebRTCDataChannel * channel, guint8 * data,
 384     gsize size, GError ** error)
 385 {
 386   GstByteReader r;
 387   guint8 message_type;
 388   gchar *label = NULL;
 389   gchar *proto = NULL;
 390
 391   if (!data)
 392     g_return_val_if_reached (GST_FLOW_ERROR);
 393   if (size < 1)
 394     g_return_val_if_reached (GST_FLOW_ERROR);
 395
 396   gst_byte_reader_init (&r, data, size);
 397
 398   if (!gst_byte_reader_get_uint8 (&r, &message_type))
 399     g_return_val_if_reached (GST_FLOW_ERROR);
 400
 401   if (message_type == CHANNEL_MESSAGE_ACK) {
 402     /* all good */
 403     GST_INFO_OBJECT (channel, "Received channel ack");
 404     return GST_FLOW_OK;
 405   } else if (message_type == CHANNEL_MESSAGE_OPEN) {
 406     guint8 reliability;
 407     guint32 reliability_param;
 408     guint16 priority, label_len, proto_len;
 409     const guint8 *src;
 410     GstBuffer *buffer;
 411     GstFlowReturn ret;
 412
 413     GST_INFO_OBJECT (channel, "Received channel open");
 414
 415     if (channel->parent.negotiated) {
 416       g_set_error (error, GST_WEBRTC_BIN_ERROR,
 417           GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 418           "Data channel was signalled as negotiated already");
 419       g_return_val_if_reached (GST_FLOW_ERROR);
 420     }
 421
 422     if (channel->opened)
 423       return GST_FLOW_OK;
 424
 425     if (!gst_byte_reader_get_uint8 (&r, &reliability))
 426       goto parse_error;
 427     if (!gst_byte_reader_get_uint16_be (&r, &priority))
 428       goto parse_error;
 429     if (!gst_byte_reader_get_uint32_be (&r, &reliability_param))
 430       goto parse_error;
 431     if (!gst_byte_reader_get_uint16_be (&r, &label_len))
 432       goto parse_error;
 433     if (!gst_byte_reader_get_uint16_be (&r, &proto_len))
 434       goto parse_error;
 435
 436     label = g_new0 (gchar, (gsize) label_len + 1);
 437     proto = g_new0 (gchar, (gsize) proto_len + 1);
 438
 439     if (!gst_byte_reader_get_data (&r, label_len, &src))
 440       goto parse_error;
 441     memcpy (label, src, label_len);
 442     label[label_len] = '\0';
 443     if (!gst_byte_reader_get_data (&r, proto_len, &src))
 444       goto parse_error;
 445     memcpy (proto, src, proto_len);
 446     proto[proto_len] = '\0';
 447
 448     g_free (channel->parent.label);
 449     channel->parent.label = label;
 450     g_free (channel->parent.protocol);
 451     channel->parent.protocol = proto;
 452     channel->parent.priority = priority_uint_to_type (priority);
 453     channel->parent.ordered = !(reliability & 0x80);
 454     if (reliability & 0x01) {
 455       channel->parent.max_retransmits = reliability_param;
 456       channel->parent.max_packet_lifetime = -1;
 457     } else if (reliability & 0x02) {
 458       channel->parent.max_retransmits = -1;
 459       channel->parent.max_packet_lifetime = reliability_param;
 460     } else {
 461       channel->parent.max_retransmits = -1;
 462       channel->parent.max_packet_lifetime = -1;
 463     }
 464     channel->opened = TRUE;
 465
 466     GST_INFO_OBJECT (channel, "Received channel open for SCTP stream %i "
 467         "label \"%s\" protocol %s ordered %s", channel->parent.id,
 468         channel->parent.label, channel->parent.protocol,
 469         channel->parent.ordered ? "true" : "false");
 470
 471     _channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
 472
 473     GST_INFO_OBJECT (channel, "Sending channel ack");
 474     buffer = construct_ack_packet (channel);
 475
 476     GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 477     channel->parent.buffered_amount += gst_buffer_get_size (buffer);
 478     GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 479
 480     ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
 481     if (ret != GST_FLOW_OK) {
 482       g_set_error (error, GST_WEBRTC_BIN_ERROR,
 483           GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 484           "Could not send ack packet");
 485       return ret;
 486     }
 487
 488     return ret;
 489   } else {
 490     g_set_error (error, GST_WEBRTC_BIN_ERROR,
 491         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 492         "Unknown message type in control protocol");
 493     return GST_FLOW_ERROR;
 494   }
 495
 496 parse_error:
 497   {
 498     g_free (label);
 499     g_free (proto);
 500     g_set_error (error, GST_WEBRTC_BIN_ERROR,
 501         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "Failed to parse packet");
 502     g_return_val_if_reached (GST_FLOW_ERROR);
 503   }
 504 }
 505
 506 static void
 507 on_sink_eos (GstAppSink * sink, gpointer user_data)
 508 {
 509 }
 510
 511 struct map_info
 512 {
 513   GstBuffer *buffer;
 514   GstMapInfo map_info;
 515 };
 516
 517 static void
 518 buffer_unmap_and_unref (struct map_info *info)
 519 {
 520   gst_buffer_unmap (info->buffer, &info->map_info);
 521   gst_buffer_unref (info->buffer);
 522   g_free (info);
 523 }
 524
 525 static void
 526 _emit_have_data (WebRTCDataChannel * channel, GBytes * data)
 527 {
 528   gst_webrtc_data_channel_on_message_data (GST_WEBRTC_DATA_CHANNEL (channel),
 529       data);
 530 }
 531
 532 static void
 533 _emit_have_string (GstWebRTCDataChannel * channel, gchar * str)
 534 {
 535   gst_webrtc_data_channel_on_message_string (GST_WEBRTC_DATA_CHANNEL (channel),
 536       str);
 537 }
 538
 539 static GstFlowReturn
 540 _data_channel_have_sample (WebRTCDataChannel * channel, GstSample * sample,
 541     GError ** error)
 542 {
 543   GstSctpReceiveMeta *receive;
 544   GstBuffer *buffer;
 545   GstFlowReturn ret = GST_FLOW_OK;
 546
 547   GST_LOG_OBJECT (channel, "Received sample %" GST_PTR_FORMAT, sample);
 548
 549   g_return_val_if_fail (channel->sctp_transport != NULL, GST_FLOW_ERROR);
 550
 551   buffer = gst_sample_get_buffer (sample);
 552   if (!buffer) {
 553     g_set_error (error, GST_WEBRTC_BIN_ERROR,
 554         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "No buffer to handle");
 555     return GST_FLOW_ERROR;
 556   }
 557   receive = gst_sctp_buffer_get_receive_meta (buffer);
 558   if (!receive) {
 559     g_set_error (error, GST_WEBRTC_BIN_ERROR,
 560         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 561         "No SCTP Receive meta on the buffer");
 562     return GST_FLOW_ERROR;
 563   }
 564
 565   switch (receive->ppid) {
 566     case DATA_CHANNEL_PPID_WEBRTC_CONTROL:{
 567       GstMapInfo info = GST_MAP_INFO_INIT;
 568       if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
 569         g_set_error (error, GST_WEBRTC_BIN_ERROR,
 570             GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 571             "Failed to map received buffer");
 572         ret = GST_FLOW_ERROR;
 573       } else {
 574         ret = _parse_control_packet (channel, info.data, info.size, error);
 575         gst_buffer_unmap (buffer, &info);
 576       }
 577       break;
 578     }
 579     case DATA_CHANNEL_PPID_WEBRTC_STRING:
 580     case DATA_CHANNEL_PPID_WEBRTC_STRING_PARTIAL:{
 581       GstMapInfo info = GST_MAP_INFO_INIT;
 582       if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
 583         g_set_error (error, GST_WEBRTC_BIN_ERROR,
 584             GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 585             "Failed to map received buffer");
 586         ret = GST_FLOW_ERROR;
 587       } else {
 588         gchar *str = g_strndup ((gchar *) info.data, info.size);
 589         _channel_enqueue_task (channel, (ChannelTask) _emit_have_string, str,
 590             g_free);
 591         gst_buffer_unmap (buffer, &info);
 592       }
 593       break;
 594     }
 595     case DATA_CHANNEL_PPID_WEBRTC_BINARY:
 596     case DATA_CHANNEL_PPID_WEBRTC_BINARY_PARTIAL:{
 597       struct map_info *info = g_new0 (struct map_info, 1);
 598       if (!gst_buffer_map (buffer, &info->map_info, GST_MAP_READ)) {
 599         g_set_error (error, GST_WEBRTC_BIN_ERROR,
 600             GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 601             "Failed to map received buffer");
 602         ret = GST_FLOW_ERROR;
 603       } else {
 604         GBytes *data = g_bytes_new_with_free_func (info->map_info.data,
 605             info->map_info.size, (GDestroyNotify) buffer_unmap_and_unref, info);
 606         info->buffer = gst_buffer_ref (buffer);
 607         _channel_enqueue_task (channel, (ChannelTask) _emit_have_data, data,
 608             (GDestroyNotify) g_bytes_unref);
 609       }
 610       break;
 611     }
 612     case DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY:
 613       _channel_enqueue_task (channel, (ChannelTask) _emit_have_data, NULL,
 614           NULL);
 615       break;
 616     case DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY:
 617       _channel_enqueue_task (channel, (ChannelTask) _emit_have_string, NULL,
 618           NULL);
 619       break;
 620     default:
 621       g_set_error (error, GST_WEBRTC_BIN_ERROR,
 622           GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 623           "Unknown SCTP PPID %u received", receive->ppid);
 624       ret = GST_FLOW_ERROR;
 625       break;
 626   }
 627
 628   return ret;
 629 }
 630
 631 static GstFlowReturn
 632 on_sink_preroll (GstAppSink * sink, gpointer user_data)
 633 {
 634   WebRTCDataChannel *channel = user_data;
 635   GstSample *sample = gst_app_sink_pull_preroll (sink);
 636   GstFlowReturn ret;
 637
 638   if (sample) {
 639     /* This sample also seems to be provided by the sample callback
 640        ret = _data_channel_have_sample (channel, sample); */
 641     ret = GST_FLOW_OK;
 642     gst_sample_unref (sample);
 643   } else if (gst_app_sink_is_eos (sink)) {
 644     ret = GST_FLOW_EOS;
 645   } else {
 646     ret = GST_FLOW_ERROR;
 647   }
 648
 649   if (ret != GST_FLOW_OK) {
 650     _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
 651   }
 652
 653   return ret;
 654 }
 655
 656 static GstFlowReturn
 657 on_sink_sample (GstAppSink * sink, gpointer user_data)
 658 {
 659   WebRTCDataChannel *channel = user_data;
 660   GstSample *sample = gst_app_sink_pull_sample (sink);
 661   GstFlowReturn ret;
 662   GError *error = NULL;
 663
 664   if (sample) {
 665     ret = _data_channel_have_sample (channel, sample, &error);
 666     gst_sample_unref (sample);
 667   } else if (gst_app_sink_is_eos (sink)) {
 668     ret = GST_FLOW_EOS;
 669   } else {
 670     ret = GST_FLOW_ERROR;
 671   }
 672
 673   if (error)
 674     _channel_store_error (channel, error);
 675
 676   if (ret != GST_FLOW_OK) {
 677     _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
 678   }
 679
 680   return ret;
 681 }
 682
 683 static GstAppSinkCallbacks sink_callbacks = {
 684   on_sink_eos,
 685   on_sink_preroll,
 686   on_sink_sample,
 687 };
 688
 689 void
 690 webrtc_data_channel_start_negotiation (WebRTCDataChannel * channel)
 691 {
 692   GstBuffer *buffer;
 693
 694   g_return_if_fail (!channel->parent.negotiated);
 695   g_return_if_fail (channel->parent.id != -1);
 696   g_return_if_fail (channel->sctp_transport != NULL);
 697
 698   buffer = construct_open_packet (channel);
 699
 700   GST_INFO_OBJECT (channel, "Sending channel open for SCTP stream %i "
 701       "label \"%s\" protocol %s ordered %s", channel->parent.id,
 702       channel->parent.label, channel->parent.protocol,
 703       channel->parent.ordered ? "true" : "false");
 704
 705   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 706   channel->parent.buffered_amount += gst_buffer_get_size (buffer);
 707   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 708
 709   if (gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc),
 710           buffer) == GST_FLOW_OK) {
 711     channel->opened = TRUE;
 712     _channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
 713   } else {
 714     GError *error = NULL;
 715     g_set_error (&error, GST_WEBRTC_BIN_ERROR,
 716         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 717         "Failed to send DCEP open packet");
 718     _channel_store_error (channel, error);
 719     _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
 720   }
 721 }
 722
 723 static void
 724 _get_sctp_reliability (WebRTCDataChannel * channel,
 725     GstSctpSendMetaPartiallyReliability * reliability, guint * rel_param)
 726 {
 727   if (channel->parent.max_retransmits != -1) {
 728     *reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_RTX;
 729     *rel_param = channel->parent.max_retransmits;
 730   } else if (channel->parent.max_packet_lifetime != -1) {
 731     *reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_TTL;
 732     *rel_param = channel->parent.max_packet_lifetime;
 733   } else {
 734     *reliability = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE;
 735     *rel_param = 0;
 736   }
 737 }
 738
 739 static gboolean
 740 _is_within_max_message_size (WebRTCDataChannel * channel, gsize size)
 741 {
 742   return size <= channel->sctp_transport->max_message_size;
 743 }
 744
 745 static void
 746 webrtc_data_channel_send_data (GstWebRTCDataChannel * base_channel,
 747     GBytes * bytes)
 748 {
 749   WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
 750   GstSctpSendMetaPartiallyReliability reliability;
 751   guint rel_param;
 752   guint32 ppid;
 753   GstBuffer *buffer;
 754   GstFlowReturn ret;
 755
 756   if (!bytes) {
 757     buffer = gst_buffer_new ();
 758     ppid = DATA_CHANNEL_PPID_WEBRTC_BINARY_EMPTY;
 759   } else {
 760     gsize size;
 761     guint8 *data;
 762
 763     data = (guint8 *) g_bytes_get_data (bytes, &size);
 764     g_return_if_fail (data != NULL);
 765     if (!_is_within_max_message_size (channel, size)) {
 766       GError *error = NULL;
 767       g_set_error (&error, GST_WEBRTC_BIN_ERROR,
 768           GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 769           "Requested to send data that is too large");
 770       _channel_store_error (channel, error);
 771       _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL,
 772           NULL);
 773       return;
 774     }
 775
 776     buffer = gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY, data, size,
 777         0, size, g_bytes_ref (bytes), (GDestroyNotify) g_bytes_unref);
 778     ppid = DATA_CHANNEL_PPID_WEBRTC_BINARY;
 779   }
 780
 781   _get_sctp_reliability (channel, &reliability, &rel_param);
 782   gst_sctp_buffer_add_send_meta (buffer, ppid, channel->parent.ordered,
 783       reliability, rel_param);
 784
 785   GST_LOG_OBJECT (channel, "Sending data using buffer %" GST_PTR_FORMAT,
 786       buffer);
 787
 788   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 789   channel->parent.buffered_amount += gst_buffer_get_size (buffer);
 790   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 791
 792   ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
 793
 794   if (ret != GST_FLOW_OK) {
 795     GError *error = NULL;
 796     g_set_error (&error, GST_WEBRTC_BIN_ERROR,
 797         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "Failed to send data");
 798     _channel_store_error (channel, error);
 799     _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
 800   }
 801 }
 802
 803 static void
 804 webrtc_data_channel_send_string (GstWebRTCDataChannel * base_channel,
 805     const gchar * str)
 806 {
 807   WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (base_channel);
 808   GstSctpSendMetaPartiallyReliability reliability;
 809   guint rel_param;
 810   guint32 ppid;
 811   GstBuffer *buffer;
 812   GstFlowReturn ret;
 813
 814   if (!channel->parent.negotiated)
 815     g_return_if_fail (channel->opened);
 816   g_return_if_fail (channel->sctp_transport != NULL);
 817
 818   if (!str) {
 819     buffer = gst_buffer_new ();
 820     ppid = DATA_CHANNEL_PPID_WEBRTC_STRING_EMPTY;
 821   } else {
 822     gsize size = strlen (str);
 823     gchar *str_copy = g_strdup (str);
 824
 825     if (!_is_within_max_message_size (channel, size)) {
 826       GError *error = NULL;
 827       g_set_error (&error, GST_WEBRTC_BIN_ERROR,
 828           GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE,
 829           "Requested to send a string that is too large");
 830       _channel_store_error (channel, error);
 831       _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL,
 832           NULL);
 833       return;
 834     }
 835
 836     buffer =
 837         gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY, str_copy,
 838         size, 0, size, str_copy, g_free);
 839     ppid = DATA_CHANNEL_PPID_WEBRTC_STRING;
 840   }
 841
 842   _get_sctp_reliability (channel, &reliability, &rel_param);
 843   gst_sctp_buffer_add_send_meta (buffer, ppid, channel->parent.ordered,
 844       reliability, rel_param);
 845
 846   GST_TRACE_OBJECT (channel, "Sending string using buffer %" GST_PTR_FORMAT,
 847       buffer);
 848
 849   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 850   channel->parent.buffered_amount += gst_buffer_get_size (buffer);
 851   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 852
 853   ret = gst_app_src_push_buffer (GST_APP_SRC (channel->appsrc), buffer);
 854
 855   if (ret != GST_FLOW_OK) {
 856     GError *error = NULL;
 857     g_set_error (&error, GST_WEBRTC_BIN_ERROR,
 858         GST_WEBRTC_BIN_ERROR_DATA_CHANNEL_FAILURE, "Failed to send string");
 859     _channel_store_error (channel, error);
 860     _channel_enqueue_task (channel, (ChannelTask) _close_procedure, NULL, NULL);
 861   }
 862 }
 863
 864 static void
 865 _on_sctp_notify_state_unlocked (GObject * sctp_transport,
 866     WebRTCDataChannel * channel)
 867 {
 868   GstWebRTCSCTPTransportState state;
 869
 870   g_object_get (sctp_transport, "state", &state, NULL);
 871   if (state == GST_WEBRTC_SCTP_TRANSPORT_STATE_CONNECTED) {
 872     if (channel->parent.negotiated)
 873       _channel_enqueue_task (channel, (ChannelTask) _emit_on_open, NULL, NULL);
 874   }
 875 }
 876
 877 static void
 878 _on_sctp_notify_state (GObject * sctp_transport, GParamSpec * pspec,
 879     WebRTCDataChannel * channel)
 880 {
 881   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 882   _on_sctp_notify_state_unlocked (sctp_transport, channel);
 883   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 884 }
 885
 886 static void
 887 _emit_low_threshold (WebRTCDataChannel * channel, gpointer user_data)
 888 {
 889   gst_webrtc_data_channel_on_buffered_amount_low (GST_WEBRTC_DATA_CHANNEL
 890       (channel));
 891 }
 892
 893 static GstPadProbeReturn
 894 on_appsrc_data (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
 895 {
 896   WebRTCDataChannel *channel = user_data;
 897   guint64 prev_amount;
 898   guint64 size = 0;
 899
 900   if (GST_PAD_PROBE_INFO_TYPE (info) & (GST_PAD_PROBE_TYPE_BUFFER)) {
 901     GstBuffer *buffer = GST_PAD_PROBE_INFO_BUFFER (info);
 902     size = gst_buffer_get_size (buffer);
 903   } else if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_BUFFER_LIST) {
 904     GstBufferList *list = GST_PAD_PROBE_INFO_BUFFER_LIST (info);
 905     size = gst_buffer_list_calculate_size (list);
 906   }
 907
 908   if (size > 0) {
 909     GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
 910     prev_amount = channel->parent.buffered_amount;
 911     channel->parent.buffered_amount -= size;
 912     GST_TRACE_OBJECT (channel, "checking low-threshold: prev %"
 913         G_GUINT64_FORMAT " low-threshold %" G_GUINT64_FORMAT " buffered %"
 914         G_GUINT64_FORMAT, prev_amount,
 915         channel->parent.buffered_amount_low_threshold,
 916         channel->parent.buffered_amount);
 917     if (prev_amount >= channel->parent.buffered_amount_low_threshold
 918         && channel->parent.buffered_amount <
 919         channel->parent.buffered_amount_low_threshold) {
 920       _channel_enqueue_task (channel, (ChannelTask) _emit_low_threshold, NULL,
 921           NULL);
 922     }
 923
 924     if (channel->parent.ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_CLOSING
 925         && channel->parent.buffered_amount <= 0) {
 926       _channel_enqueue_task (channel, (ChannelTask) _close_sctp_stream, NULL,
 927           NULL);
 928     }
 929     GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
 930   }
 931
 932   return GST_PAD_PROBE_OK;
 933 }
 934
 935 static void
 936 gst_webrtc_data_channel_constructed (GObject * object)
 937 {
 938   WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (object);
 939   GstPad *pad;
 940   GstCaps *caps;
 941
 942   caps = gst_caps_new_any ();
 943
 944   channel->appsrc = gst_element_factory_make ("appsrc", NULL);
 945   gst_object_ref_sink (channel->appsrc);
 946   pad = gst_element_get_static_pad (channel->appsrc, "src");
 947
 948   channel->src_probe = gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_DATA_BOTH,
 949       (GstPadProbeCallback) on_appsrc_data, channel, NULL);
 950
 951   channel->appsink = gst_element_factory_make ("appsink", NULL);
 952   gst_object_ref_sink (channel->appsink);
 953   g_object_set (channel->appsink, "sync", FALSE, "async", FALSE, "caps", caps,
 954       NULL);
 955   gst_app_sink_set_callbacks (GST_APP_SINK (channel->appsink), &sink_callbacks,
 956       channel, NULL);
 957
 958   gst_object_unref (pad);
 959   gst_caps_unref (caps);
 960 }
 961
 962 static void
 963 gst_webrtc_data_channel_finalize (GObject * object)
 964 {
 965   WebRTCDataChannel *channel = WEBRTC_DATA_CHANNEL (object);
 966
 967   if (channel->src_probe) {
 968     GstPad *pad = gst_element_get_static_pad (channel->appsrc, "src");
 969     gst_pad_remove_probe (pad, channel->src_probe);
 970     gst_object_unref (pad);
 971     channel->src_probe = 0;
 972   }
 973
 974   if (channel->sctp_transport)
 975     g_signal_handlers_disconnect_by_data (channel->sctp_transport, channel);
 976   g_clear_object (&channel->sctp_transport);
 977
 978   g_clear_object (&channel->appsrc);
 979   g_clear_object (&channel->appsink);
 980
 981   G_OBJECT_CLASS (parent_class)->finalize (object);
 982 }
 983
 984 static void
 985 webrtc_data_channel_class_init (WebRTCDataChannelClass * klass)
 986 {
 987   GObjectClass *gobject_class = (GObjectClass *) klass;
 988   GstWebRTCDataChannelClass *channel_class =
 989       (GstWebRTCDataChannelClass *) klass;
 990
 991   gobject_class->constructed = gst_webrtc_data_channel_constructed;
 992   gobject_class->finalize = gst_webrtc_data_channel_finalize;
 993
 994   channel_class->send_data = webrtc_data_channel_send_data;
 995   channel_class->send_string = webrtc_data_channel_send_string;
 996   channel_class->close = webrtc_data_channel_close;
 997 }
 998
 999 static void
1000 webrtc_data_channel_init (WebRTCDataChannel * channel)
1001 {
1002 }
1003
1004 static void
1005 _data_channel_set_sctp_transport (WebRTCDataChannel * channel,
1006     WebRTCSCTPTransport * sctp)
1007 {
1008   g_return_if_fail (GST_IS_WEBRTC_DATA_CHANNEL (channel));
1009   g_return_if_fail (GST_IS_WEBRTC_SCTP_TRANSPORT (sctp));
1010
1011   GST_WEBRTC_DATA_CHANNEL_LOCK (channel);
1012   if (channel->sctp_transport)
1013     g_signal_handlers_disconnect_by_data (channel->sctp_transport, channel);
1014
1015   gst_object_replace ((GstObject **) & channel->sctp_transport,
1016       GST_OBJECT (sctp));
1017
1018   if (sctp) {
1019     g_signal_connect (sctp, "stream-reset", G_CALLBACK (_on_sctp_stream_reset),
1020         channel);
1021     g_signal_connect (sctp, "notify::state", G_CALLBACK (_on_sctp_notify_state),
1022         channel);
1023   }
1024   GST_WEBRTC_DATA_CHANNEL_UNLOCK (channel);
1025 }
1026
1027 void
1028 webrtc_data_channel_link_to_sctp (WebRTCDataChannel * channel,
1029     WebRTCSCTPTransport * sctp_transport)
1030 {
1031   if (sctp_transport && !channel->sctp_transport) {
1032     gint id;
1033
1034     g_object_get (channel, "id", &id, NULL);
1035
1036     if (sctp_transport->association_established && id != -1) {
1037       gchar *pad_name;
1038
1039       _data_channel_set_sctp_transport (channel, sctp_transport);
1040       pad_name = g_strdup_printf ("sink_%u", id);
1041       if (!gst_element_link_pads (channel->appsrc, "src",
1042               channel->sctp_transport->sctpenc, pad_name))
1043         g_warn_if_reached ();
1044       g_free (pad_name);
1045
1046       _on_sctp_notify_state_unlocked (G_OBJECT (sctp_transport), channel);
1047     }
1048   }
1049 }