net/quic/quic_http_stream.cc

// Copyright (c) 2012 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_http_stream.h"

#include <utility>

#include "base/auto_reset.h"
#include "base/callback_helpers.h"
#include "base/metrics/histogram_macros.h"
#include "base/strings/string_split.h"
#include "base/strings/stringprintf.h"
#include "net/base/load_flags.h"
#include "net/base/net_errors.h"
#include "net/http/http_response_headers.h"
#include "net/http/http_util.h"
#include "net/quic/quic_client_promised_info.h"
#include "net/quic/quic_http_utils.h"
#include "net/quic/quic_utils.h"
#include "net/quic/spdy_utils.h"
#include "net/spdy/spdy_frame_builder.h"
#include "net/spdy/spdy_framer.h"
#include "net/spdy/spdy_http_utils.h"
#include "net/ssl/ssl_info.h"

namespace net {

namespace {

std::unique_ptr<base::Value> NetLogQuicPushStreamCallback(
    QuicStreamId stream_id,
    const GURL* url,
    NetLogCaptureMode capture_mode) {
  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
  dict->SetInteger("stream_id", stream_id);
  dict->SetString("url", url->spec());
  return std::move(dict);
}

}  // namespace

QuicHttpStream::QuicHttpStream(
    const base::WeakPtr<QuicChromiumClientSession>& session)
    : next_state_(STATE_NONE),
      session_(session),
      session_error_(OK),
      was_handshake_confirmed_(session->IsCryptoHandshakeConfirmed()),
      stream_(nullptr),
      request_info_(nullptr),
      request_body_stream_(nullptr),
      priority_(MINIMUM_PRIORITY),
      response_info_(nullptr),
      response_status_(OK),
      response_headers_received_(false),
      headers_bytes_received_(0),
      headers_bytes_sent_(0),
      closed_stream_received_bytes_(0),
      closed_stream_sent_bytes_(0),
      user_buffer_len_(0),
      quic_connection_error_(QUIC_NO_ERROR),
      port_migration_detected_(false),
      found_promise_(false),
      push_handle_(nullptr),
      in_loop_(false),
      weak_factory_(this) {
  DCHECK(session_);
  session_->AddObserver(this);
}

QuicHttpStream::~QuicHttpStream() {
  CHECK(!in_loop_);
  Close(false);
  if (session_)
    session_->RemoveObserver(this);
}

bool QuicHttpStream::CheckVary(const SpdyHeaderBlock& client_request,
                               const SpdyHeaderBlock& promise_request,
                               const SpdyHeaderBlock& promise_response) {
  HttpResponseInfo promise_response_info;

  HttpRequestInfo promise_request_info;
  ConvertHeaderBlockToHttpRequestHeaders(promise_request,
                                         &promise_request_info.extra_headers);
  HttpRequestInfo client_request_info;
  ConvertHeaderBlockToHttpRequestHeaders(client_request,
                                         &client_request_info.extra_headers);

  if (!SpdyHeadersToHttpResponse(promise_response, &promise_response_info)) {
    DLOG(WARNING) << "Invalid headers";
    return false;
  }

  HttpVaryData vary_data;
  if (!vary_data.Init(promise_request_info,
                      *promise_response_info.headers.get())) {
    // Promise didn't contain valid vary info, so URL match was sufficient.
    return true;
  }
  // Now compare the client request for matching.
  return vary_data.MatchesRequest(client_request_info,
                                  *promise_response_info.headers.get());
}

void QuicHttpStream::OnRendezvousResult(QuicSpdyStream* stream) {
  push_handle_ = nullptr;
  if (stream) {
    stream_ = static_cast<QuicChromiumClientStream*>(stream);
    stream_->SetDelegate(this);
  }
  // callback_ should be non-null in the case of asynchronous
  // rendezvous; i.e. |Try()| returned QUIC_PENDING.
  if (!callback_.is_null()) {
    if (stream) {
      next_state_ = STATE_OPEN;
      stream_net_log_.AddEvent(
          NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM,
          base::Bind(&NetLogQuicPushStreamCallback, stream_->id(),
                     &request_info_->url));
      session_->net_log().AddEvent(
          NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM,
          base::Bind(&NetLogQuicPushStreamCallback, stream_->id(),
                     &request_info_->url));
      DoCallback(OK);
      return;
    }
    // rendezvous has failed so proceed as with a non-push request.
    next_state_ = STATE_REQUEST_STREAM;
    OnIOComplete(OK);
  }
}

int QuicHttpStream::InitializeStream(const HttpRequestInfo* request_info,
                                     RequestPriority priority,
                                     const BoundNetLog& stream_net_log,
                                     const CompletionCallback& callback) {
  DCHECK(!stream_);
  if (!session_)
    return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED
                                    : ERR_QUIC_HANDSHAKE_FAILED;

  stream_net_log.AddEvent(
      NetLog::TYPE_HTTP_STREAM_REQUEST_BOUND_TO_QUIC_SESSION,
      session_->net_log().source().ToEventParametersCallback());

  stream_net_log_ = stream_net_log;
  request_info_ = request_info;
  request_time_ = base::Time::Now();
  priority_ = priority;

  bool success = session_->GetSSLInfo(&ssl_info_);
  DCHECK(success);
  DCHECK(ssl_info_.cert.get());

  std::string url(request_info->url.spec());
  QuicClientPromisedInfo* promised =
      session_->push_promise_index()->GetPromised(url);
  if (promised) {
    found_promise_ = true;
    stream_net_log_.AddEvent(
        NetLog::TYPE_QUIC_HTTP_STREAM_PUSH_PROMISE_RENDEZVOUS,
        base::Bind(&NetLogQuicPushStreamCallback, promised->id(),
                   &request_info_->url));
    session_->net_log().AddEvent(
        NetLog::TYPE_QUIC_HTTP_STREAM_PUSH_PROMISE_RENDEZVOUS,
        base::Bind(&NetLogQuicPushStreamCallback, promised->id(),
                   &request_info_->url));
    return OK;
  }

  next_state_ = STATE_REQUEST_STREAM;
  int rv = DoLoop(OK);
  if (rv == ERR_IO_PENDING)
    callback_ = callback;

  return rv;
}

void QuicHttpStream::OnStreamReady(int rv) {
  DCHECK(rv == OK || !stream_);
  if (rv == OK) {
    stream_->SetDelegate(this);
    if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) {
      stream_->DisableConnectionMigration();
    }
    if (response_info_) {
      // This happens in the case of a asynchronous push rendezvous
      // that ultimately fails (e.g. vary failure).  |response_info_|
      // non-null implies that |DoStreamRequest()| was called via
      // |SendRequest()|.
      next_state_ = STATE_SET_REQUEST_PRIORITY;
      rv = DoLoop(OK);
    }
  } else if (!was_handshake_confirmed_) {
    rv = ERR_QUIC_HANDSHAKE_FAILED;
  }
  if (rv != ERR_IO_PENDING) {
    DoCallback(rv);
  }
}

bool QuicHttpStream::CancelPromiseIfHasBody() {
  if (!request_body_stream_)
    return false;
  // Method type or request with body ineligble for push.
  this->push_handle_->Cancel();
  this->push_handle_ = nullptr;
  next_state_ = STATE_REQUEST_STREAM;
  return true;
}

int QuicHttpStream::HandlePromise() {
  QuicAsyncStatus push_status = session_->push_promise_index()->Try(
      request_headers_, this, &this->push_handle_);

  switch (push_status) {
    case QUIC_FAILURE:
      // Push rendezvous failed.
      next_state_ = STATE_REQUEST_STREAM;
      break;
    case QUIC_SUCCESS:
      next_state_ = STATE_OPEN;
      if (!CancelPromiseIfHasBody()) {
        stream_net_log_.AddEvent(
            NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM,
            base::Bind(&NetLogQuicPushStreamCallback, stream_->id(),
                       &request_info_->url));
        session_->net_log().AddEvent(
            NetLog::TYPE_QUIC_HTTP_STREAM_ADOPTED_PUSH_STREAM,
            base::Bind(&NetLogQuicPushStreamCallback, stream_->id(),
                       &request_info_->url));
        // Avoid the call to |DoLoop()| below, which would reset
        // next_state_ to STATE_NONE.
        return OK;
      }

      break;
    case QUIC_PENDING:
      if (!CancelPromiseIfHasBody()) {
        // Have a promise but the promised stream doesn't exist yet.
        // Still have to do validation before accepting the promised
        // stream for sure.
        return ERR_IO_PENDING;
      }
      break;
  }
  return DoLoop(OK);
}

int QuicHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
                                HttpResponseInfo* response,
                                const CompletionCallback& callback) {
  CHECK(!request_body_stream_);
  CHECK(!response_info_);
  CHECK(!callback.is_null());
  CHECK(response);

  // TODO(rch): remove this once we figure out why channel ID is not being
  // sent when it should be.
  HostPortPair origin = HostPortPair::FromURL(request_info_->url);
  if (origin.Equals(HostPortPair("accounts.google.com", 443)) &&
      request_headers.HasHeader(HttpRequestHeaders::kCookie)) {
    UMA_HISTOGRAM_BOOLEAN("Net.QuicSession.CookieSentToAccountsOverChannelId",
                          ssl_info_.channel_id_sent);
  }
  if ((!found_promise_ && !stream_) || !session_) {
    return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED
                                    : ERR_QUIC_HANDSHAKE_FAILED;
  }

  // Store the serialized request headers.
  CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers,
                                   /*direct=*/true, &request_headers_);

  // Store the request body.
  request_body_stream_ = request_info_->upload_data_stream;
  if (request_body_stream_) {
    // TODO(rch): Can we be more precise about when to allocate
    // raw_request_body_buf_. Removed the following check. DoReadRequestBody()
    // was being called even if we didn't yet allocate raw_request_body_buf_.
    //   && (request_body_stream_->size() ||
    //       request_body_stream_->is_chunked()))
    // Use 10 packets as the body buffer size to give enough space to
    // help ensure we don't often send out partial packets.
    raw_request_body_buf_ =
        new IOBufferWithSize(static_cast<size_t>(10 * kMaxPacketSize));
    // The request body buffer is empty at first.
    request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), 0);
  }

  // Store the response info.
  response_info_ = response;

  int rv;

  if (found_promise_) {
    rv = HandlePromise();
  } else {
    next_state_ = STATE_SET_REQUEST_PRIORITY;
    rv = DoLoop(OK);
  }

  if (rv == ERR_IO_PENDING)
    callback_ = callback;

  return rv > 0 ? OK : rv;
}

UploadProgress QuicHttpStream::GetUploadProgress() const {
  if (!request_body_stream_)
    return UploadProgress();

  return UploadProgress(request_body_stream_->position(),
                        request_body_stream_->size());
}

int QuicHttpStream::ReadResponseHeaders(const CompletionCallback& callback) {
  CHECK(!callback.is_null());

  if (stream_ == nullptr)
    return response_status_;

  // Check if we already have the response headers. If so, return synchronously.
  if (response_headers_received_)
    return OK;

  // Still waiting for the response, return IO_PENDING.
  CHECK(callback_.is_null());
  callback_ = callback;
  return ERR_IO_PENDING;
}

int QuicHttpStream::ReadResponseBody(IOBuffer* buf,
                                     int buf_len,
                                     const CompletionCallback& callback) {
  if (!stream_) {
    // If the stream is already closed, there is no body to read.
    return response_status_;
  }

  int rv = ReadAvailableData(buf, buf_len);
  if (rv != ERR_IO_PENDING)
    return rv;

  CHECK(callback_.is_null());
  CHECK(!user_buffer_.get());
  CHECK_EQ(0, user_buffer_len_);

  callback_ = callback;
  user_buffer_ = buf;
  user_buffer_len_ = buf_len;
  return ERR_IO_PENDING;
}

void QuicHttpStream::Close(bool not_reusable) {
  // Note: the not_reusable flag has no meaning for SPDY streams.
  if (stream_) {
    stream_->SetDelegate(nullptr);
    stream_->Reset(QUIC_STREAM_CANCELLED);
    ResetStream();
    response_status_ = was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED
                                                : ERR_QUIC_HANDSHAKE_FAILED;
  }
}

HttpStream* QuicHttpStream::RenewStreamForAuth() {
  return nullptr;
}

bool QuicHttpStream::IsResponseBodyComplete() const {
  return next_state_ == STATE_OPEN && !stream_;
}

bool QuicHttpStream::IsConnectionReused() const {
  // TODO(rch): do something smarter here.
  return stream_ && stream_->id() > 1;
}

void QuicHttpStream::SetConnectionReused() {
  // QUIC doesn't need an indicator here.
}

bool QuicHttpStream::CanReuseConnection() const {
  // QUIC streams aren't considered reusable.
  return false;
}

int64_t QuicHttpStream::GetTotalReceivedBytes() const {
  // TODO(sclittle): Currently, this only includes headers and response body
  // bytes. Change this to include QUIC overhead as well.
  int64_t total_received_bytes = headers_bytes_received_;
  if (stream_) {
    total_received_bytes += stream_->stream_bytes_read();
  } else {
    total_received_bytes += closed_stream_received_bytes_;
  }
  return total_received_bytes;
}

int64_t QuicHttpStream::GetTotalSentBytes() const {
  // TODO(sclittle): Currently, this only includes request headers and body
  // bytes. Change this to include QUIC overhead as well.
  int64_t total_sent_bytes = headers_bytes_sent_;
  if (stream_) {
    total_sent_bytes += stream_->stream_bytes_written();
  } else {
    total_sent_bytes += closed_stream_sent_bytes_;
  }
  return total_sent_bytes;
}

bool QuicHttpStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
  // TODO(mmenke):  Figure out what to do here.
  return true;
}

void QuicHttpStream::GetSSLInfo(SSLInfo* ssl_info) {
  *ssl_info = ssl_info_;
}

void QuicHttpStream::GetSSLCertRequestInfo(
    SSLCertRequestInfo* cert_request_info) {
  DCHECK(stream_);
  NOTIMPLEMENTED();
}

bool QuicHttpStream::GetRemoteEndpoint(IPEndPoint* endpoint) {
  if (!session_)
    return false;

  *endpoint = session_->peer_address();
  return true;
}

Error QuicHttpStream::GetSignedEKMForTokenBinding(crypto::ECPrivateKey* key,
                                                  std::vector<uint8_t>* out) {
  return session_->GetTokenBindingSignature(key, out);
}

void QuicHttpStream::Drain(HttpNetworkSession* session) {
  NOTREACHED();
  Close(false);
  delete this;
}

void QuicHttpStream::PopulateNetErrorDetails(NetErrorDetails* details) {
  details->connection_info = HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3;
  if (was_handshake_confirmed_)
    details->quic_connection_error = quic_connection_error_;
  if (session_) {
    session_->PopulateNetErrorDetails(details);
  } else {
    details->quic_port_migration_detected = port_migration_detected_;
  }
}

void QuicHttpStream::SetPriority(RequestPriority priority) {
  priority_ = priority;
}

void QuicHttpStream::OnHeadersAvailable(const SpdyHeaderBlock& headers,
                                        size_t frame_len) {
  headers_bytes_received_ += frame_len;

  // QuicHttpStream ignores trailers.
  if (response_headers_received_) {
    if (stream_->IsDoneReading()) {
      // Close the read side. If the write side has been closed, this will
      // invoke QuicHttpStream::OnClose to reset the stream.
      stream_->OnFinRead();
    }
    return;
  }

  int rv = ProcessResponseHeaders(headers);
  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
    DoCallback(rv);
  }
}

void QuicHttpStream::OnDataAvailable() {
  if (callback_.is_null()) {
    // Data is available, but can't be delivered
    return;
  }

  CHECK(user_buffer_.get());
  CHECK_NE(0, user_buffer_len_);
  int rv = ReadAvailableData(user_buffer_.get(), user_buffer_len_);
  if (rv == ERR_IO_PENDING) {
    // This was a spurrious notification. Wait for the next one.
    return;
  }

  CHECK(!callback_.is_null());
  user_buffer_ = nullptr;
  user_buffer_len_ = 0;
  DoCallback(rv);
}

void QuicHttpStream::OnClose() {
  if (stream_->connection_error() != QUIC_NO_ERROR ||
      stream_->stream_error() != QUIC_STREAM_NO_ERROR) {
    response_status_ = was_handshake_confirmed_ ? ERR_QUIC_PROTOCOL_ERROR
                                                : ERR_QUIC_HANDSHAKE_FAILED;
  } else if (!response_headers_received_) {
    response_status_ = ERR_ABORTED;
  }

  quic_connection_error_ = stream_->connection_error();
  ResetStream();
  if (in_loop_) {
    // If already in DoLoop(), |callback_| will be handled when DoLoop() exits.
    return;
  }
  if (!callback_.is_null()) {
    DoCallback(response_status_);
  }
}

void QuicHttpStream::OnError(int error) {
  ResetStream();
  response_status_ =
      was_handshake_confirmed_ ? error : ERR_QUIC_HANDSHAKE_FAILED;
  if (!callback_.is_null())
    DoCallback(response_status_);
}

bool QuicHttpStream::HasSendHeadersComplete() {
  return next_state_ > STATE_SEND_HEADERS_COMPLETE;
}

void QuicHttpStream::OnCryptoHandshakeConfirmed() {
  was_handshake_confirmed_ = true;
  if (next_state_ == STATE_WAIT_FOR_CONFIRMATION_COMPLETE) {
    int rv = DoLoop(OK);

    if (rv != ERR_IO_PENDING && !callback_.is_null()) {
      DoCallback(rv);
    }
  }
}

void QuicHttpStream::OnSessionClosed(int error, bool port_migration_detected) {
  Close(false);
  session_error_ = error;
  port_migration_detected_ = port_migration_detected;
  session_.reset();
}

void QuicHttpStream::OnIOComplete(int rv) {
  rv = DoLoop(rv);

  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
    DoCallback(rv);
  }
}

void QuicHttpStream::DoCallback(int rv) {
  CHECK_NE(rv, ERR_IO_PENDING);
  CHECK(!callback_.is_null());
  CHECK(!in_loop_);

  // The client callback can do anything, including destroying this class,
  // so any pending callback must be issued after everything else is done.
  base::ResetAndReturn(&callback_).Run(rv);
}

int QuicHttpStream::DoLoop(int rv) {
  CHECK(!in_loop_);
  base::AutoReset<bool> auto_reset_in_loop(&in_loop_, true);
  do {
    State state = next_state_;
    next_state_ = STATE_NONE;
    switch (state) {
      case STATE_REQUEST_STREAM:
        rv = DoStreamRequest();
        break;
      case STATE_SET_REQUEST_PRIORITY:
        rv = DoSetRequestPriority();
        break;
      case STATE_WAIT_FOR_CONFIRMATION:
        CHECK_EQ(OK, rv);
        rv = DoWaitForConfirmation();
        break;
      case STATE_WAIT_FOR_CONFIRMATION_COMPLETE:
        rv = DoWaitForConfirmationComplete(rv);
        break;
      case STATE_SEND_HEADERS:
        CHECK_EQ(OK, rv);
        rv = DoSendHeaders();
        break;
      case STATE_SEND_HEADERS_COMPLETE:
        rv = DoSendHeadersComplete(rv);
        break;
      case STATE_READ_REQUEST_BODY:
        CHECK_EQ(OK, rv);
        rv = DoReadRequestBody();
        break;
      case STATE_READ_REQUEST_BODY_COMPLETE:
        rv = DoReadRequestBodyComplete(rv);
        break;
      case STATE_SEND_BODY:
        CHECK_EQ(OK, rv);
        rv = DoSendBody();
        break;
      case STATE_SEND_BODY_COMPLETE:
        rv = DoSendBodyComplete(rv);
        break;
      case STATE_OPEN:
        CHECK_EQ(OK, rv);
        break;
      default:
        NOTREACHED() << "next_state_: " << next_state_;
        break;
    }
  } while (next_state_ != STATE_NONE && next_state_ != STATE_OPEN &&
           rv != ERR_IO_PENDING);

  return rv;
}

int QuicHttpStream::DoStreamRequest() {
  // TODO(rtenneti) Bug: b/28676259 - a temporary fix until we find out why
  // |session_| could be a nullptr. Delete |null_session| check and histogram if
  // session is never a nullptr.
  bool null_session = session_ == nullptr;
  if (null_session) {
    UMA_HISTOGRAM_BOOLEAN("Net.QuicHttpStream::DoStreamRequest.IsNullSession",
                          null_session);
    return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED
                                    : ERR_QUIC_HANDSHAKE_FAILED;
  }
  int rv = stream_request_.StartRequest(
      session_, &stream_,
      base::Bind(&QuicHttpStream::OnStreamReady, weak_factory_.GetWeakPtr()));
  if (rv == OK) {
    stream_->SetDelegate(this);
    if (request_info_->load_flags & LOAD_DISABLE_CONNECTION_MIGRATION) {
      stream_->DisableConnectionMigration();
    }
    if (response_info_) {
      next_state_ = STATE_SET_REQUEST_PRIORITY;
    }
  } else if (rv != ERR_IO_PENDING && !was_handshake_confirmed_) {
    rv = ERR_QUIC_HANDSHAKE_FAILED;
  }
  return rv;
}

int QuicHttpStream::DoSetRequestPriority() {
  // Set priority according to request
  DCHECK(stream_);
  DCHECK(response_info_);
  SpdyPriority priority = ConvertRequestPriorityToQuicPriority(priority_);
  stream_->SetPriority(priority);
  next_state_ = STATE_WAIT_FOR_CONFIRMATION;
  return OK;
}

int QuicHttpStream::DoWaitForConfirmation() {
  next_state_ = STATE_WAIT_FOR_CONFIRMATION_COMPLETE;
  if (!session_->IsCryptoHandshakeConfirmed() &&
      request_info_->method == "POST") {
    return ERR_IO_PENDING;
  }
  return OK;
}

int QuicHttpStream::DoWaitForConfirmationComplete(int rv) {
  if (rv < 0)
    return rv;

  next_state_ = STATE_SEND_HEADERS;
  return OK;
}

int QuicHttpStream::DoSendHeaders() {
  if (!stream_)
    return ERR_UNEXPECTED;

  // Log the actual request with the URL Request's net log.
  stream_net_log_.AddEvent(
      NetLog::TYPE_HTTP_TRANSACTION_QUIC_SEND_REQUEST_HEADERS,
      base::Bind(&QuicRequestNetLogCallback, stream_->id(), &request_headers_,
                 priority_));
  bool has_upload_data = request_body_stream_ != nullptr;

  next_state_ = STATE_SEND_HEADERS_COMPLETE;
  size_t frame_len = stream_->WriteHeaders(std::move(request_headers_),
                                           !has_upload_data, nullptr);
  headers_bytes_sent_ += frame_len;

  request_headers_ = SpdyHeaderBlock();
  return static_cast<int>(frame_len);
}

int QuicHttpStream::DoSendHeadersComplete(int rv) {
  if (rv < 0)
    return rv;

  // If the stream is already closed, don't read the request the body.
  if (!stream_)
    return response_status_;

  next_state_ = request_body_stream_ ? STATE_READ_REQUEST_BODY : STATE_OPEN;

  return OK;
}

int QuicHttpStream::DoReadRequestBody() {
  next_state_ = STATE_READ_REQUEST_BODY_COMPLETE;
  return request_body_stream_->Read(
      raw_request_body_buf_.get(), raw_request_body_buf_->size(),
      base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
}

int QuicHttpStream::DoReadRequestBodyComplete(int rv) {
  // If the stream is already closed, don't continue.
  if (!stream_)
    return response_status_;

  // |rv| is the result of read from the request body from the last call to
  // DoSendBody().
  if (rv < 0) {
    stream_->SetDelegate(nullptr);
    stream_->Reset(QUIC_ERROR_PROCESSING_STREAM);
    ResetStream();
    return rv;
  }

  request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), rv);
  if (rv == 0) {  // Reached the end.
    DCHECK(request_body_stream_->IsEOF());
  }

  next_state_ = STATE_SEND_BODY;
  return OK;
}

int QuicHttpStream::DoSendBody() {
  if (!stream_)
    return ERR_UNEXPECTED;

  CHECK(request_body_stream_);
  CHECK(request_body_buf_.get());
  const bool eof = request_body_stream_->IsEOF();
  int len = request_body_buf_->BytesRemaining();
  if (len > 0 || eof) {
    next_state_ = STATE_SEND_BODY_COMPLETE;
    base::StringPiece data(request_body_buf_->data(), len);
    return stream_->WriteStreamData(
        data, eof,
        base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
  }

  next_state_ = STATE_OPEN;
  return OK;
}

int QuicHttpStream::DoSendBodyComplete(int rv) {
  if (rv < 0)
    return rv;

  // If the stream is already closed, don't continue.
  if (!stream_)
    return response_status_;

  request_body_buf_->DidConsume(request_body_buf_->BytesRemaining());

  if (!request_body_stream_->IsEOF()) {
    next_state_ = STATE_READ_REQUEST_BODY;
    return OK;
  }

  next_state_ = STATE_OPEN;
  return OK;
}

int QuicHttpStream::ProcessResponseHeaders(const SpdyHeaderBlock& headers) {
  if (!SpdyHeadersToHttpResponse(headers, response_info_)) {
    DLOG(WARNING) << "Invalid headers";
    return ERR_QUIC_PROTOCOL_ERROR;
  }
  // Put the peer's IP address and port into the response.
  IPEndPoint address = session_->peer_address();
  response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
  response_info_->connection_info =
      HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3;
  response_info_->vary_data.Init(*request_info_,
                                 *response_info_->headers.get());
  response_info_->was_npn_negotiated = true;
  response_info_->npn_negotiated_protocol = "quic/1+spdy/3";
  response_info_->response_time = base::Time::Now();
  response_info_->request_time = request_time_;
  response_headers_received_ = true;

  return OK;
}

int QuicHttpStream::ReadAvailableData(IOBuffer* buf, int buf_len) {
  int rv = stream_->Read(buf, buf_len);
  // TODO(rtenneti): Temporary fix for crbug.com/585591. Added a check for null
  // |stream_| to fix crash bug. Delete |stream_| check and histogram after fix
  // is merged.
  bool null_stream = stream_ == nullptr;
  UMA_HISTOGRAM_BOOLEAN("Net.QuicReadAvailableData.NullStream", null_stream);
  if (null_stream)
    return rv;
  if (stream_->IsDoneReading()) {
    stream_->SetDelegate(nullptr);
    stream_->OnFinRead();
    ResetStream();
  }
  return rv;
}

void QuicHttpStream::ResetStream() {
  if (push_handle_) {
    push_handle_->Cancel();
    push_handle_ = nullptr;
  }
  if (!stream_)
    return;
  closed_stream_received_bytes_ = stream_->stream_bytes_read();
  closed_stream_sent_bytes_ = stream_->stream_bytes_written();
  stream_ = nullptr;

  // If |request_body_stream_| is non-NULL, Reset it, to abort any in progress
  // read.
  if (request_body_stream_)
    request_body_stream_->Reset();
}

}  // namespace net