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