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   DCHECK(FinishedReadingHeaders());
  62   return sequencer()->Readv(iov, iov_len);
  63 }
  64
  65 int QuicDataStream::GetReadableRegions(iovec* iov, size_t iov_len) const {
  66   DCHECK(FinishedReadingHeaders());
  67   return sequencer()->GetReadableRegions(iov, iov_len);
  68 }
  69
  70 bool QuicDataStream::IsDoneReading() const {
  71   if (!headers_decompressed_ || !decompressed_headers_.empty()) {
  72     return false;
  73   }
  74   return sequencer()->IsClosed();
  75 }
  76
  77 bool QuicDataStream::HasBytesToRead() const {
  78   return !decompressed_headers_.empty() || sequencer()->HasBytesToRead();
  79 }
  80
  81 void QuicDataStream::MarkHeadersConsumed(size_t bytes_consumed) {
  82   decompressed_headers_.erase(0, bytes_consumed);
  83   if (FinishedReadingHeaders()) {
  84     sequencer()->FlushBufferedFrames();
  85   }
  86 }
  87
  88 void QuicDataStream::set_priority(QuicPriority priority) {
  89   DCHECK_EQ(0u, stream_bytes_written());
  90   priority_ = priority;
  91 }
  92
  93 QuicPriority QuicDataStream::EffectivePriority() const {
  94   return priority();
  95 }
  96
  97 uint32 QuicDataStream::ProcessRawData(const char* data, uint32 data_len) {
  98   if (!FinishedReadingHeaders()) {
  99     LOG(DFATAL) << "ProcessRawData called before headers have been finished";
 100     return 0;
 101   }
 102   return ProcessData(data, data_len);
 103 }
 104
 105 void QuicDataStream::OnStreamHeaders(StringPiece headers_data) {
 106   headers_data.AppendToString(&decompressed_headers_);
 107 }
 108
 109 void QuicDataStream::OnStreamHeadersPriority(QuicPriority priority) {
 110   DCHECK_EQ(Perspective::IS_SERVER, session()->connection()->perspective());
 111   set_priority(priority);
 112 }
 113
 114 void QuicDataStream::OnStreamHeadersComplete(bool fin, size_t frame_len) {
 115   headers_decompressed_ = true;
 116   if (fin) {
 117     sequencer()->OnStreamFrame(QuicStreamFrame(id(), fin, 0, StringPiece()));
 118   }
 119   if (FinishedReadingHeaders()) {
 120     sequencer()->FlushBufferedFrames();
 121   }
 122 }
 123
 124 void QuicDataStream::OnClose() {
 125   ReliableQuicStream::OnClose();
 126
 127   if (visitor_) {
 128     Visitor* visitor = visitor_;
 129     // Calling Visitor::OnClose() may result the destruction of the visitor,
 130     // so we need to ensure we don't call it again.
 131     visitor_ = nullptr;
 132     visitor->OnClose(this);
 133   }
 134 }
 135
 136 bool QuicDataStream::FinishedReadingHeaders() const {
 137   return headers_decompressed_ && decompressed_headers_.empty();
 138 }
 139
 140 }  // namespace net