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
12using base::StringPiece;
13using std::min;
14
15namespace net {
16
17#define ENDPOINT (session()->is_server() ? "Server: " : " Client: ")
18
19namespace {
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.
25QuicPriority kDefaultPriority = 3;
26
27}  // namespace
28
29QuicDataStream::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
43QuicDataStream::~QuicDataStream() {
44}
45
46size_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
60size_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
86int 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
99bool QuicDataStream::IsDoneReading() const {
100  if (!headers_decompressed_ || !decompressed_headers_.empty()) {
101    return false;
102  }
103  return sequencer()->IsClosed();
104}
105
106bool QuicDataStream::HasBytesToRead() const {
107  return !decompressed_headers_.empty() || sequencer()->HasBytesToRead();
108}
109
110void QuicDataStream::set_priority(QuicPriority priority) {
111  DCHECK_EQ(0u, stream_bytes_written());
112  priority_ = priority;
113}
114
115QuicPriority QuicDataStream::EffectivePriority() const {
116  return priority();
117}
118
119uint32 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
127const IPEndPoint& QuicDataStream::GetPeerAddress() {
128  return session()->peer_address();
129}
130
131bool QuicDataStream::GetSSLInfo(SSLInfo* ssl_info) {
132  return session()->GetSSLInfo(ssl_info);
133}
134
135uint32 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
150void QuicDataStream::OnStreamHeaders(StringPiece headers_data) {
151  headers_data.AppendToString(&decompressed_headers_);
152  ProcessHeaderData();
153}
154
155void QuicDataStream::OnStreamHeadersPriority(QuicPriority priority) {
156  DCHECK(session()->connection()->is_server());
157  set_priority(priority);
158}
159
160void 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
171void 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
183bool QuicDataStream::FinishedReadingHeaders() {
184  return headers_decompressed_ && decompressed_headers_.empty();
185}
186
187}  // namespace net
188