net/quic/quic_data_stream.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/quic/quic_data_stream.h"
   6
   7 #include "base/logging.h"
   8 #include "net/quic/quic_session.h"
   9 #include "net/quic/quic_utils.h"
  10 #include "net/quic/quic_write_blocked_list.h"
  11
  12 using base::StringPiece;
  13 using std::min;
  14
  15 namespace net {
  16
  17 #define ENDPOINT (session()->is_server() ? "Server: " : " Client: ")
  18
  19 namespace {
  20
  21 // This is somewhat arbitrary.  It's possible, but unlikely, we will either fail
  22 // to set a priority client-side, or cancel a stream before stripping the
  23 // priority from the wire server-side.  In either case, start out with a
  24 // priority in the middle.
  25 QuicPriority kDefaultPriority = 3;
  26
  27 }  // namespace
  28
  29 QuicDataStream::QuicDataStream(QuicStreamId id, QuicSession* session)
  30     : ReliableQuicStream(id, session),
  31       visitor_(nullptr),
  32       headers_decompressed_(false),
  33       priority_(kDefaultPriority) {
  34   DCHECK_NE(kCryptoStreamId, id);
  35   // Don't receive any callbacks from the sequencer until headers
  36   // are complete.
  37   sequencer()->SetBlockedUntilFlush();
  38 }
  39
  40 QuicDataStream::~QuicDataStream() {
  41 }
  42
  43 size_t QuicDataStream::WriteHeaders(
  44     const SpdyHeaderBlock& header_block,
  45     bool fin,
  46     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
  47   size_t bytes_written = session()->WriteHeaders(
  48       id(), header_block, fin, priority_, ack_notifier_delegate);
  49   if (fin) {
  50     // TODO(rch): Add test to ensure fin_sent_ is set whenever a fin is sent.
  51     set_fin_sent(true);
  52     CloseWriteSide();
  53   }
  54   return bytes_written;
  55 }
  56
  57 size_t QuicDataStream::Readv(const struct iovec* iov, size_t iov_len) {
  58   if (FinishedReadingHeaders()) {
  59     // If the headers have been read, simply delegate to the sequencer's
  60     // Readv method.
  61     return sequencer()->Readv(iov, iov_len);
  62   }
  63   // Otherwise, copy decompressed header data into |iov|.
  64   size_t bytes_consumed = 0;
  65   size_t iov_index = 0;
  66   while (iov_index < iov_len &&
  67          decompressed_headers_.length() > bytes_consumed) {
  68     size_t bytes_to_read = min(iov[iov_index].iov_len,
  69                                decompressed_headers_.length() - bytes_consumed);
  70     char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base);
  71     memcpy(iov_ptr,
  72            decompressed_headers_.data() + bytes_consumed, bytes_to_read);
  73     bytes_consumed += bytes_to_read;
  74     ++iov_index;
  75   }
  76   decompressed_headers_.erase(0, bytes_consumed);
  77   if (FinishedReadingHeaders()) {
  78     sequencer()->FlushBufferedFrames();
  79   }
  80   return bytes_consumed;
  81 }
  82
  83 int QuicDataStream::GetReadableRegions(iovec* iov, size_t iov_len) {
  84   if (FinishedReadingHeaders()) {
  85     return sequencer()->GetReadableRegions(iov, iov_len);
  86   }
  87   if (iov_len == 0) {
  88     return 0;
  89   }
  90   iov[0].iov_base = static_cast<void*>(
  91       const_cast<char*>(decompressed_headers_.data()));
  92   iov[0].iov_len = decompressed_headers_.length();
  93   return 1;
  94 }
  95
  96 bool QuicDataStream::IsDoneReading() const {
  97   if (!headers_decompressed_ || !decompressed_headers_.empty()) {
  98     return false;
  99   }
 100   return sequencer()->IsClosed();
 101 }
 102
 103 bool QuicDataStream::HasBytesToRead() const {
 104   return !decompressed_headers_.empty() || sequencer()->HasBytesToRead();
 105 }
 106
 107 void QuicDataStream::set_priority(QuicPriority priority) {
 108   DCHECK_EQ(0u, stream_bytes_written());
 109   priority_ = priority;
 110 }
 111
 112 QuicPriority QuicDataStream::EffectivePriority() const {
 113   return priority();
 114 }
 115
 116 uint32 QuicDataStream::ProcessRawData(const char* data, uint32 data_len) {
 117   if (!FinishedReadingHeaders()) {
 118     LOG(DFATAL) << "ProcessRawData called before headers have been finished";
 119     return 0;
 120   }
 121   return ProcessData(data, data_len);
 122 }
 123
 124 const IPEndPoint& QuicDataStream::GetPeerAddress() {
 125   return session()->peer_address();
 126 }
 127
 128 bool QuicDataStream::GetSSLInfo(SSLInfo* ssl_info) {
 129   return session()->GetSSLInfo(ssl_info);
 130 }
 131
 132 uint32 QuicDataStream::ProcessHeaderData() {
 133   if (decompressed_headers_.empty()) {
 134     return 0;
 135   }
 136
 137   size_t bytes_processed = ProcessData(decompressed_headers_.data(),
 138                                        decompressed_headers_.length());
 139   if (bytes_processed == decompressed_headers_.length()) {
 140     decompressed_headers_.clear();
 141   } else {
 142     decompressed_headers_ = decompressed_headers_.erase(0, bytes_processed);
 143   }
 144   return bytes_processed;
 145 }
 146
 147 void QuicDataStream::OnStreamHeaders(StringPiece headers_data) {
 148   headers_data.AppendToString(&decompressed_headers_);
 149   ProcessHeaderData();
 150 }
 151
 152 void QuicDataStream::OnStreamHeadersPriority(QuicPriority priority) {
 153   DCHECK(session()->connection()->is_server());
 154   set_priority(priority);
 155 }
 156
 157 void QuicDataStream::OnStreamHeadersComplete(bool fin, size_t frame_len) {
 158   headers_decompressed_ = true;
 159   if (fin) {
 160     sequencer()->OnStreamFrame(QuicStreamFrame(id(), fin, 0, IOVector()));
 161   }
 162   ProcessHeaderData();
 163   if (FinishedReadingHeaders()) {
 164     sequencer()->FlushBufferedFrames();
 165   }
 166 }
 167
 168 void QuicDataStream::OnClose() {
 169   ReliableQuicStream::OnClose();
 170
 171   if (visitor_) {
 172     Visitor* visitor = visitor_;
 173     // Calling Visitor::OnClose() may result the destruction of the visitor,
 174     // so we need to ensure we don't call it again.
 175     visitor_ = nullptr;
 176     visitor->OnClose(this);
 177   }
 178 }
 179
 180 bool QuicDataStream::FinishedReadingHeaders() {
 181   return headers_decompressed_ && decompressed_headers_.empty();
 182 }
 183
 184 }  // namespace net