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