Home | History | Annotate | Download | only in quic 
      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,
     30                                QuicSession* session)
     31     : ReliableQuicStream(id, session),
     32       visitor_(NULL),
     33       headers_decompressed_(false),
     34       priority_(kDefaultPriority),
     35       decompression_failed_(false),
     36       priority_parsed_(false) {
     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 = session()->WriteHeaders(
     51       id(), header_block, fin, 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 const IPEndPoint& QuicDataStream::GetPeerAddress() {
    128   return session()->peer_address();
    129 }
    130 
    131 bool QuicDataStream::GetSSLInfo(SSLInfo* ssl_info) {
    132   return session()->GetSSLInfo(ssl_info);
    133 }
    134 
    135 uint32 QuicDataStream::ProcessHeaderData() {
    136   if (decompressed_headers_.empty()) {
    137     return 0;
    138   }
    139 
    140   size_t bytes_processed = ProcessData(decompressed_headers_.data(),
    141                                        decompressed_headers_.length());
    142   if (bytes_processed == decompressed_headers_.length()) {
    143     decompressed_headers_.clear();
    144   } else {
    145     decompressed_headers_ = decompressed_headers_.erase(0, bytes_processed);
    146   }
    147   return bytes_processed;
    148 }
    149 
    150 void QuicDataStream::OnStreamHeaders(StringPiece headers_data) {
    151   headers_data.AppendToString(&decompressed_headers_);
    152   ProcessHeaderData();
    153 }
    154 
    155 void QuicDataStream::OnStreamHeadersPriority(QuicPriority priority) {
    156   DCHECK(session()->connection()->is_server());
    157   set_priority(priority);
    158 }
    159 
    160 void QuicDataStream::OnStreamHeadersComplete(bool fin, size_t frame_len) {
    161   headers_decompressed_ = true;
    162   if (fin) {
    163     sequencer()->OnStreamFrame(QuicStreamFrame(id(), fin, 0, IOVector()));
    164   }
    165   ProcessHeaderData();
    166   if (FinishedReadingHeaders()) {
    167     sequencer()->FlushBufferedFrames();
    168   }
    169 }
    170 
    171 void QuicDataStream::OnClose() {
    172   ReliableQuicStream::OnClose();
    173 
    174   if (visitor_) {
    175     Visitor* visitor = visitor_;
    176     // Calling Visitor::OnClose() may result the destruction of the visitor,
    177     // so we need to ensure we don't call it again.
    178     visitor_ = NULL;
    179     visitor->OnClose(this);
    180   }
    181 }
    182 
    183 bool QuicDataStream::FinishedReadingHeaders() {
    184   return headers_decompressed_ && decompressed_headers_.empty();
    185 }
    186 
    187 }  // namespace net
    188