// Copyright 2013 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "net/quic/quic_data_stream.h"
#include "base/logging.h"
#include "net/quic/quic_session.h"
#include "net/quic/quic_utils.h"
#include "net/quic/quic_write_blocked_list.h"
using base::StringPiece;
using std::min;
namespace net {
#define ENDPOINT (session()->is_server() ? "Server: " : " Client: ")
namespace {
// This is somewhat arbitrary. It's possible, but unlikely, we will either fail
// to set a priority client-side, or cancel a stream before stripping the
// priority from the wire server-side. In either case, start out with a
// priority in the middle.
QuicPriority kDefaultPriority = 3;
} // namespace
QuicDataStream::QuicDataStream(QuicStreamId id,
QuicSession* session)
: ReliableQuicStream(id, session),
visitor_(NULL),
headers_decompressed_(false),
priority_(kDefaultPriority),
decompression_failed_(false),
priority_parsed_(false) {
DCHECK_NE(kCryptoStreamId, id);
// Don't receive any callbacks from the sequencer until headers
// are complete.
sequencer()->SetBlockedUntilFlush();
}
QuicDataStream::~QuicDataStream() {
}
size_t QuicDataStream::WriteHeaders(
const SpdyHeaderBlock& header_block,
bool fin,
QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
size_t bytes_written = session()->WriteHeaders(
id(), header_block, fin, ack_notifier_delegate);
if (fin) {
// TODO(rch): Add test to ensure fin_sent_ is set whenever a fin is sent.
set_fin_sent(true);
CloseWriteSide();
}
return bytes_written;
}
size_t QuicDataStream::Readv(const struct iovec* iov, size_t iov_len) {
if (FinishedReadingHeaders()) {
// If the headers have been read, simply delegate to the sequencer's
// Readv method.
return sequencer()->Readv(iov, iov_len);
}
// Otherwise, copy decompressed header data into |iov|.
size_t bytes_consumed = 0;
size_t iov_index = 0;
while (iov_index < iov_len &&
decompressed_headers_.length() > bytes_consumed) {
size_t bytes_to_read = min(iov[iov_index].iov_len,
decompressed_headers_.length() - bytes_consumed);
char* iov_ptr = static_cast<char*>(iov[iov_index].iov_base);
memcpy(iov_ptr,
decompressed_headers_.data() + bytes_consumed, bytes_to_read);
bytes_consumed += bytes_to_read;
++iov_index;
}
decompressed_headers_.erase(0, bytes_consumed);
if (FinishedReadingHeaders()) {
sequencer()->FlushBufferedFrames();
}
return bytes_consumed;
}
int QuicDataStream::GetReadableRegions(iovec* iov, size_t iov_len) {
if (FinishedReadingHeaders()) {
return sequencer()->GetReadableRegions(iov, iov_len);
}
if (iov_len == 0) {
return 0;
}
iov[0].iov_base = static_cast<void*>(
const_cast<char*>(decompressed_headers_.data()));
iov[0].iov_len = decompressed_headers_.length();
return 1;
}
bool QuicDataStream::IsDoneReading() const {
if (!headers_decompressed_ || !decompressed_headers_.empty()) {
return false;
}
return sequencer()->IsClosed();
}
bool QuicDataStream::HasBytesToRead() const {
return !decompressed_headers_.empty() || sequencer()->HasBytesToRead();
}
void QuicDataStream::set_priority(QuicPriority priority) {
DCHECK_EQ(0u, stream_bytes_written());
priority_ = priority;
}
QuicPriority QuicDataStream::EffectivePriority() const {
return priority();
}
uint32 QuicDataStream::ProcessRawData(const char* data, uint32 data_len) {
if (!FinishedReadingHeaders()) {
LOG(DFATAL) << "ProcessRawData called before headers have been finished";
return 0;
}
return ProcessData(data, data_len);
}
const IPEndPoint& QuicDataStream::GetPeerAddress() {
return session()->peer_address();
}
bool QuicDataStream::GetSSLInfo(SSLInfo* ssl_info) {
return session()->GetSSLInfo(ssl_info);
}
uint32 QuicDataStream::ProcessHeaderData() {
if (decompressed_headers_.empty()) {
return 0;
}
size_t bytes_processed = ProcessData(decompressed_headers_.data(),
decompressed_headers_.length());
if (bytes_processed == decompressed_headers_.length()) {
decompressed_headers_.clear();
} else {
decompressed_headers_ = decompressed_headers_.erase(0, bytes_processed);
}
return bytes_processed;
}
void QuicDataStream::OnStreamHeaders(StringPiece headers_data) {
headers_data.AppendToString(&decompressed_headers_);
ProcessHeaderData();
}
void QuicDataStream::OnStreamHeadersPriority(QuicPriority priority) {
DCHECK(session()->connection()->is_server());
set_priority(priority);
}
void QuicDataStream::OnStreamHeadersComplete(bool fin, size_t frame_len) {
headers_decompressed_ = true;
if (fin) {
sequencer()->OnStreamFrame(QuicStreamFrame(id(), fin, 0, IOVector()));
}
ProcessHeaderData();
if (FinishedReadingHeaders()) {
sequencer()->FlushBufferedFrames();
}
}
void QuicDataStream::OnClose() {
ReliableQuicStream::OnClose();
if (visitor_) {
Visitor* visitor = visitor_;
// Calling Visitor::OnClose() may result the destruction of the visitor,
// so we need to ensure we don't call it again.
visitor_ = NULL;
visitor->OnClose(this);
}
}
bool QuicDataStream::FinishedReadingHeaders() {
return headers_decompressed_ && decompressed_headers_.empty();
}
} // namespace net