Add QuicDataChannel and QuicDataTransport classes

webrtc/api/quicdatachannel.cc

+/*
+ *  Copyright 2016 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include "webrtc/api/quicdatachannel.h"
+
+#include "webrtc/base/bind.h"
+#include "webrtc/base/bytebuffer.h"
+#include "webrtc/base/copyonwritebuffer.h"
+#include "webrtc/base/logging.h"
+#include "webrtc/p2p/quic/quictransportchannel.h"
+#include "webrtc/p2p/quic/reliablequicstream.h"
+
+namespace webrtc {
+
+QuicDataChannel::QuicDataChannel(
+    rtc::Thread* signaling_thread,
+    rtc::Thread* worker_thread,
+    const std::string& label,
+    const DataChannelInit& config,
+    const QuicMessageDispatcher& message_dispatcher)
+    : signaling_thread_(signaling_thread),
+      worker_thread_(worker_thread),
+      message_dispatcher_(message_dispatcher),
+      id_(config.id),
+      state_(kConnecting),
+      buffered_amount_(0),
+      num_sent_messages_(0),
+      label_(label),
+      protocol_(config.protocol) {}
+
+QuicDataChannel::~QuicDataChannel() {}
+
+void QuicDataChannel::RegisterObserver(DataChannelObserver* observer) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  observer_ = observer;
+}
+
+void QuicDataChannel::UnregisterObserver() {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  observer_ = nullptr;
+}
+
+bool QuicDataChannel::Send(const DataBuffer& buffer) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ != kOpen) {
+    LOG(LS_ERROR) << "QUIC data channel " << id_
+                  << " is not open so cannot send.";
+    return false;
+  }
+  return worker_thread_->Invoke<bool>(
+      rtc::Bind(&QuicDataChannel::Send_w, this, buffer));
+}
+
+bool QuicDataChannel::Send_w(const DataBuffer& buffer) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  if (buffer.size() == 0) {
+    LOG(LS_WARNING) << "QUIC data channel " << id_
+                    << " refuses to send an empty message.";

[pthatcher1, 2016/04/14 17:21:56]

I think this was just a limitation of SCTP.  I don't think we need to emulate
it.  In other words, we can support empty messages, so let's do so.

[mikescarlett, 2016/04/14 22:16:50]

Ok I'll remove that.

+    return true;
+  }
+  // Send the header containing the data channel ID and message ID.
+  rtc::CopyOnWriteBuffer header;
+  message_dispatcher_.EncodeHeader(id_, ++num_sent_messages_, &header);

[pthatcher1, 2016/04/14 17:21:56]

It's strange to me that a class called MessageDispatcher has a method called
EncodeHeader.  Why not just make EncodeHeader a function separate from any
class?

[mikescarlett, 2016/04/14 22:16:50]

Done.

+  RTC_DCHECK(quic_transport_channel_);
+  cricket::ReliableQuicStream* stream =
+      quic_transport_channel_->CreateQuicStream();
+  RTC_DCHECK(stream);
+  uint64_t old_queued_bytes = stream->queued_data_bytes();
+  stream->Write(header.data<char>(), header.size(), false);

[pthatcher1, 2016/04/14 17:21:56]

Don't you need to check the Write call with the header?  What if it
blocks/fails?

[mikescarlett, 2016/04/14 22:16:50]

I added a check.

+  // Send the message with FIN == true, which signals to the remote peer that
+  // there is no more data after this message.

[pthatcher1, 2016/04/14 17:21:56]

Can you add a line like this?

bool fin = true;

[mikescarlett, 2016/04/14 22:16:50]

Done.

+  rtc::StreamResult result =
+      stream->Write(buffer.data.data<char>(), buffer.size(), true);
+  if (result == rtc::SR_SUCCESS) {
+    // The message is sent and we don't need this QUIC stream.
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " successfully wrote data for QUIC data channel " << id_;
+    stream->Close();
+    return true;
+  }
+  // TODO(mikescarlett): Register the ReliableQuicStream's priority to the
+  // QuicWriteBlockedList so that the QUIC session doesn't drop messages when
+  // the QUIC transport channel becomes unwritable.
+  if (result == rtc::SR_BLOCK) {
+    // The QUIC stream is write blocked, so the message will be queued by the
+    // QUIC session. If this is due to the QUIC not being writable, it will be
+    // sent once QUIC becomes writable again. Otherwise it may be due to
+    // exceeding the QUIC flow control limit, in which case the remote peer's
+    // QUIC session will tell the QUIC stream to send more data.
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " is write blocked for QUIC data channel " << id_;
+    invoker_.AsyncInvoke<void>(
+        signaling_thread_, rtc::Bind(&QuicDataChannel::OnBufferedAmountChange_s,
+                                     this, buffered_amount_));
+    buffered_amount_ += stream->queued_data_bytes() - old_queued_bytes;
+    stream->SignalQueuedBytesWritten.connect(
+        this, &QuicDataChannel::OnQueuedBytesWritten);
+    write_blocked_quic_streams_[stream->id()] = stream;
+    // The QUIC stream will be removed from |write_blocked_quic_streams_| once
+    // it closes.
+    stream->SignalClosed.connect(this,
+                                 &QuicDataChannel::OnWriteBlockedStreamClosed);
+    return true;
+  }
+  LOG(LS_ERROR) << "Stream " << stream->id()
+                << " failed to write for QUIC data channel " << id_
+                << ". Unexpected result: " << result;
+  return false;
+}
+
+void QuicDataChannel::OnQueuedBytesWritten(net::QuicStreamId stream_id,
+                                           uint64_t queued_bytes_written) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_, rtc::Bind(&QuicDataChannel::OnBufferedAmountChange_s,
+                                   this, buffered_amount_));
+  buffered_amount_ -= queued_bytes_written;
+  const auto& kv = write_blocked_quic_streams_.find(stream_id);
+  RTC_DCHECK(kv != write_blocked_quic_streams_.end());
+  cricket::ReliableQuicStream* stream = kv->second;
+  // True if the QUIC stream is done sending data.
+  if (stream->fin_sent()) {
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " successfully wrote data for QUIC data channel " << id_;
+    stream->Close();
+  }
+}
+
+void QuicDataChannel::Close() {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ == kClosed || state_ == kClosing) {
+    return;
+  }
+  LOG(LS_INFO) << "Closing QUIC data channel.";
+  SetState_s(kClosing);
+  worker_thread_->Invoke<void>(rtc::Bind(&QuicDataChannel::Close_w, this));
+}
+
+void QuicDataChannel::Close_w() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  for (auto& kv : incoming_quic_streams_) {
+    Message& message = kv.second;
+    cricket::ReliableQuicStream* stream = message.stream;
+    stream->Close();
+  }
+
+  for (auto& kv : write_blocked_quic_streams_) {
+    cricket::ReliableQuicStream* stream = kv.second;
+    stream->Close();
+  }
+
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_,
+      rtc::Bind(&QuicDataChannel::SetState_s, this, kClosed));
+}
+
+void QuicDataChannel::SetTransportChannel(
+    cricket::QuicTransportChannel* channel) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  RTC_DCHECK(channel);
+  RTC_DCHECK(!quic_transport_channel_);
+  quic_transport_channel_ = channel;
+  LOG(LS_INFO) << "Setting QuicTransportChannel for QUIC data channel " << id_;
+  worker_thread_->Invoke<void>(
+      rtc::Bind(&QuicDataChannel::ConnectTransportChannel_w, this));
+  if (quic_transport_channel_->writable()) {
+    SetState_s(kOpen);
+  }
+}
+
+void QuicDataChannel::ConnectTransportChannel_w() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  quic_transport_channel_->SignalReadyToSend.connect(
+      this, &QuicDataChannel::OnReadyToSend);
+  quic_transport_channel_->SignalClosed.connect(
+      this, &QuicDataChannel::OnConnectionClosed);
+}
+
+void QuicDataChannel::OnIncomingStream(uint64_t message_id,
+                                       const char* first_bytes,
+                                       size_t len,
+                                       cricket::ReliableQuicStream* stream) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(first_bytes);
+  RTC_DCHECK(stream);
+  if (!observer_) {
+    LOG(LS_WARNING) << "QUIC data channel " << id_
+                    << " received a message but has no observer.";
+    stream->Close();
+    return;
+  }
+  // A FIN is received if the message fits into a single QUIC stream frame and
+  // the remote peer is done sending.
+  if (stream->fin_received()) {
+    LOG(LS_INFO) << "Stream " << stream->id()
+                 << " has finished receiving data for QUIC data channel "
+                 << id_;
+    DataBuffer final_message(rtc::CopyOnWriteBuffer(first_bytes, len), false);
+    invoker_.AsyncInvoke<void>(signaling_thread_,
+                               rtc::Bind(&QuicDataChannel::OnMessage_s, this,
+                                         std::move(final_message)));
+    stream->Close();
+    return;
+  }
+  // Otherwise the message is divided across multiple QUIC stream frames, so
+  // queue the data. OnDataReceived() will be called each time the remaining
+  // QUIC stream frames arrive.
+  LOG(LS_INFO) << "QUIC data channel " << id_
+               << " is queuing incoming data for stream " << stream->id();
+  rtc::CopyOnWriteBuffer received_data;
+  if (len > 0) {
+    received_data.AppendData(first_bytes, len);
+  }
+  Message message;
+  message.stream = stream;
+  message.buffer = std::move(received_data);
+  incoming_quic_streams_[stream->id()] = std::move(message);
+  stream->SignalDataReceived.connect(this, &QuicDataChannel::OnDataReceived);
+  // The QUIC stream will be removed from |incoming_quic_streams_| once it
+  // closes.
+  stream->SignalClosed.connect(this,
+                               &QuicDataChannel::OnIncomingQueuedStreamClosed);
+}
+
+void QuicDataChannel::OnDataReceived(net::QuicStreamId stream_id,
+                                     const char* data,
+                                     size_t len) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(data);
+  const auto& kv = incoming_quic_streams_.find(stream_id);
+  RTC_DCHECK(kv != incoming_quic_streams_.end());
+  Message& message = kv->second;
+  cricket::ReliableQuicStream* stream = message.stream;
+  rtc::CopyOnWriteBuffer& received_data = message.buffer;
+  // If the QUIC stream has not received a FIN, then the remote peer is not
+  // finished sending data.
+  if (!stream->fin_received()) {
+    received_data.AppendData(data, len);
+    return;
+  }
+  // Otherwise we are done receiving and can provide the data channel observer
+  // with the message.
+  LOG(LS_INFO) << "Stream " << stream_id
+               << " has finished receiving data for QUIC data channel " << id_;
+  received_data.AppendData(data, len);
+  DataBuffer final_message(std::move(received_data), false);
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_,
+      rtc::Bind(&QuicDataChannel::OnMessage_s, this, std::move(final_message)));
+  // Once the stream is closed, OnDataReceived will not fire for the stream.
+  stream->Close();
+}
+
+void QuicDataChannel::OnReadyToSend(cricket::TransportChannel* channel) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(channel == quic_transport_channel_);
+  LOG(LS_INFO) << "QuicTransportChannel is ready to send";
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_, rtc::Bind(&QuicDataChannel::SetState_s, this, kOpen));
+}
+
+// Called when a write blocked QUIC stream that has been added to
+// |write_blocked_quic_streams_| is closed.
+void QuicDataChannel::OnWriteBlockedStreamClosed(net::QuicStreamId stream_id,
+                                                 int error) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  LOG(LS_VERBOSE) << "Write blocked stream " << stream_id << " is closed.";
+  write_blocked_quic_streams_.erase(stream_id);
+}
+
+// Called when an incoming QUIC stream that has been added to
+// |incoming_quic_streams_| is closed.
+void QuicDataChannel::OnIncomingQueuedStreamClosed(net::QuicStreamId stream_id,
+                                                   int error) {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  LOG(LS_VERBOSE) << "Incoming queued stream " << stream_id << " is closed.";
+  incoming_quic_streams_.erase(stream_id);
+}
+
+void QuicDataChannel::OnConnectionClosed() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  invoker_.AsyncInvoke<void>(
+      signaling_thread_,
+      rtc::Bind(&QuicDataChannel::SetState_s, this, kClosed));
+}
+
+void QuicDataChannel::OnMessage_s(const DataBuffer& received_data) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (observer_) {
+    observer_->OnMessage(received_data);
+  }
+}
+
+void QuicDataChannel::SetState_s(DataState state) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (state_ == state || state_ == kClosed) {
+    return;
+  }
+  if (state_ == kClosing && state != kClosed) {
+    return;
+  }
+  LOG(LS_INFO) << "Setting state to " << state << " for QUIC data channel "
+               << id_;
+  state_ = state;
+  if (observer_) {
+    observer_->OnStateChange();
+  }
+}
+
+void QuicDataChannel::OnBufferedAmountChange_s(uint64_t buffered_amount) {
+  RTC_DCHECK(signaling_thread_->IsCurrent());
+  if (observer_) {
+    observer_->OnBufferedAmountChange(buffered_amount);
+  }
+}
+
+size_t QuicDataChannel::GetNumWriteBlockedStreams() const {
+  return write_blocked_quic_streams_.size();
+}
+
+size_t QuicDataChannel::GetNumIncomingStreams() const {
+  return incoming_quic_streams_.size();
+}
+
+}  // namespace webrtc