Refactoring on QUIC

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.
  */
 
@@ skipping 43 matching lines @@
                   << *data_channel_id;
     return false;
   }
   size_t remaining_bytes = byte_buffer.Length();
   *bytes_read = len - remaining_bytes;
   return true;
 }
 
 QuicDataChannel::QuicDataChannel(rtc::Thread* signaling_thread,
                                  rtc::Thread* worker_thread,
+                                 rtc::Thread* network_thread,
                                  const std::string& label,
                                  const DataChannelInit& config)
     : signaling_thread_(signaling_thread),
       worker_thread_(worker_thread),
+      network_thread_(network_thread),
       id_(config.id),
       state_(kConnecting),
       buffered_amount_(0),
       next_message_id_(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_FROM_HERE, rtc::Bind(&QuicDataChannel::Send_w, this, buffer));
+  return network_thread_->Invoke<bool>(
+      RTC_FROM_HERE, rtc::Bind(&QuicDataChannel::Send_n, this, buffer));
 }
 
-bool QuicDataChannel::Send_w(const DataBuffer& buffer) {
-  RTC_DCHECK(worker_thread_->IsCurrent());
+bool QuicDataChannel::Send_n(const DataBuffer& buffer) {
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   // Encode and send the header containing the data channel ID and message ID.
   rtc::CopyOnWriteBuffer header;
   WriteQuicDataChannelMessageHeader(id_, ++next_message_id_, &header);
   RTC_DCHECK(quic_transport_channel_);
   cricket::ReliableQuicStream* stream =
       quic_transport_channel_->CreateQuicStream();
   RTC_DCHECK(stream);
 
   // Send the header with a FIN if the message is empty.
@@ skipping 139 matching lines @@
       this, &QuicDataChannel::OnReadyToSend);
   quic_transport_channel_->SignalClosed.connect(
       this, &QuicDataChannel::OnConnectionClosed);
   if (quic_transport_channel_->writable()) {
     return kOpen;
   }
   return kConnecting;
 }
 
 void QuicDataChannel::OnIncomingMessage(Message&& message) {
-  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   RTC_DCHECK(message.stream);
   if (!observer_) {
     LOG(LS_WARNING) << "QUIC data channel " << id_
                     << " received a message but has no observer.";
     message.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 (message.stream->fin_received()) {
@@ skipping 18 matching lines @@
                                              &QuicDataChannel::OnDataReceived);
   // The QUIC stream will be removed from |incoming_quic_messages_| once it
   // closes.
   message.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(network_thread_->IsCurrent());
   RTC_DCHECK(data);
   const auto& kv = incoming_quic_messages_.find(stream_id);
   if (kv == incoming_quic_messages_.end()) {
     RTC_DCHECK(false);
     return;
   }
   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>(
       RTC_FROM_HERE, 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(network_thread_->IsCurrent());
   RTC_DCHECK(channel == quic_transport_channel_);
   LOG(LS_INFO) << "QuicTransportChannel is ready to send";
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, signaling_thread_,
       rtc::Bind(&QuicDataChannel::SetState_s, this, kOpen));
 }
 
 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);
 }
 
 void QuicDataChannel::OnIncomingQueuedStreamClosed(net::QuicStreamId stream_id,
                                                    int error) {
-  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK(network_thread_->IsCurrent());
   LOG(LS_VERBOSE) << "Incoming queued stream " << stream_id << " is closed.";
   incoming_quic_messages_.erase(stream_id);
 }
 
 void QuicDataChannel::OnConnectionClosed() {
   RTC_DCHECK(worker_thread_->IsCurrent());
   invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_,
                              rtc::Bind(&QuicDataChannel::Close, this));
 }
 
@@ skipping 29 matching lines @@
 
 size_t QuicDataChannel::GetNumWriteBlockedStreams() const {
   return write_blocked_quic_streams_.size();
 }
 
 size_t QuicDataChannel::GetNumIncomingStreams() const {
   return incoming_quic_messages_.size();
 }
 
 }  // namespace webrtc