net/spdy/bidirectional_stream_spdy_job.cc

// Copyright 2015 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/spdy/bidirectional_stream_spdy_job.h"

#include "base/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/time/time.h"
#include "base/timer/timer.h"
#include "net/http/bidirectional_stream_request_info.h"
#include "net/spdy/spdy_buffer.h"
#include "net/spdy/spdy_header_block.h"
#include "net/spdy/spdy_http_utils.h"
#include "net/spdy/spdy_stream.h"

namespace net {

namespace {

// Time to wait in millisecond to notify |delegate_| of data received.
// Handing small chunks of data to the caller creates measurable overhead.
// So buffer data in short time-spans and send a single read notification.
const int kBufferTimeMs = 1;

}  // namespace

BidirectionalStreamSpdyJob::BidirectionalStreamSpdyJob(
    const base::WeakPtr<SpdySession>& spdy_session)
    : spdy_session_(spdy_session),
      request_info_(nullptr),
      delegate_(nullptr),
      negotiated_protocol_(kProtoUnknown),
      more_read_data_pending_(false),
      read_buffer_len_(0),
      stream_closed_(false),
      closed_stream_status_(ERR_FAILED),
      closed_stream_received_bytes_(0),
      closed_stream_sent_bytes_(0),
      weak_factory_(this) {}

BidirectionalStreamSpdyJob::~BidirectionalStreamSpdyJob() {
  if (stream_) {
    stream_->DetachDelegate();
    DCHECK(!stream_);
  }
}

void BidirectionalStreamSpdyJob::Start(
    const BidirectionalStreamRequestInfo* request_info,
    const BoundNetLog& net_log,
    BidirectionalStreamJob::Delegate* delegate,
    scoped_ptr<base::Timer> timer) {
  DCHECK(!stream_);
  DCHECK(timer);

  delegate_ = delegate;
  timer_ = std::move(timer);

  if (!spdy_session_) {
    delegate_->OnFailed(ERR_CONNECTION_CLOSED);
    return;
  }

  request_info_ = request_info;

  int rv = stream_request_.StartRequest(
      SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url,
      request_info_->priority, net_log,
      base::Bind(&BidirectionalStreamSpdyJob::OnStreamInitialized,
                 weak_factory_.GetWeakPtr()));
  if (rv != ERR_IO_PENDING)
    OnStreamInitialized(rv);
}

int BidirectionalStreamSpdyJob::ReadData(IOBuffer* buf, int buf_len) {
  if (stream_)
    DCHECK(!stream_->IsIdle());

  DCHECK(buf);
  DCHECK(buf_len);
  DCHECK(!timer_->IsRunning()) << "There should be only one ReadData in flight";

  // If there is data buffered, complete the IO immediately.
  if (!read_data_queue_.IsEmpty()) {
    return read_data_queue_.Dequeue(buf->data(), buf_len);
  } else if (stream_closed_) {
    return closed_stream_status_;
  }
  // Read will complete asynchronously and Delegate::OnReadCompleted will be
  // called upon completion.
  read_buffer_ = buf;
  read_buffer_len_ = buf_len;
  return ERR_IO_PENDING;
}

void BidirectionalStreamSpdyJob::SendData(IOBuffer* data,
                                          int length,
                                          bool end_stream) {
  DCHECK(!stream_closed_);
  DCHECK(stream_);

  stream_->SendData(data, length,
                    end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
}

void BidirectionalStreamSpdyJob::Cancel() {
  if (!stream_)
    return;
  // Cancels the stream and detaches the delegate so it doesn't get called back.
  stream_->DetachDelegate();
  DCHECK(!stream_);
}

NextProto BidirectionalStreamSpdyJob::GetProtocol() const {
  return negotiated_protocol_;
}

int64_t BidirectionalStreamSpdyJob::GetTotalReceivedBytes() const {
  if (stream_closed_)
    return closed_stream_received_bytes_;

  if (!stream_)
    return 0;

  return stream_->raw_received_bytes();
}

int64_t BidirectionalStreamSpdyJob::GetTotalSentBytes() const {
  if (stream_closed_)
    return closed_stream_sent_bytes_;

  if (!stream_)
    return 0;

  return stream_->raw_sent_bytes();
}

void BidirectionalStreamSpdyJob::OnRequestHeadersSent() {
  DCHECK(stream_);

  negotiated_protocol_ = stream_->GetProtocol();
  delegate_->OnHeadersSent();
}

SpdyResponseHeadersStatus BidirectionalStreamSpdyJob::OnResponseHeadersUpdated(
    const SpdyHeaderBlock& response_headers) {
  DCHECK(stream_);

  delegate_->OnHeadersReceived(response_headers);
  return RESPONSE_HEADERS_ARE_COMPLETE;
}

void BidirectionalStreamSpdyJob::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
  DCHECK(stream_);
  DCHECK(!stream_closed_);

  // If |buffer| is null, BidirectionalStreamSpdyJob::OnClose will be invoked by
  // SpdyStream to indicate the end of stream.
  if (!buffer)
    return;

  // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust
  // recv window size accordingly.
  read_data_queue_.Enqueue(std::move(buffer));
  if (read_buffer_) {
    // Handing small chunks of data to the caller creates measurable overhead.
    // So buffer data in short time-spans and send a single read notification.
    ScheduleBufferedRead();
  }
}

void BidirectionalStreamSpdyJob::OnDataSent() {
  DCHECK(stream_);
  DCHECK(!stream_closed_);

  delegate_->OnDataSent();
}

void BidirectionalStreamSpdyJob::OnTrailers(const SpdyHeaderBlock& trailers) {
  DCHECK(stream_);
  DCHECK(!stream_closed_);

  delegate_->OnTrailersReceived(trailers);
}

void BidirectionalStreamSpdyJob::OnClose(int status) {
  DCHECK(stream_);

  stream_closed_ = true;
  closed_stream_status_ = status;
  closed_stream_received_bytes_ = stream_->raw_received_bytes();
  closed_stream_sent_bytes_ = stream_->raw_sent_bytes();
  stream_.reset();

  if (status != OK) {
    delegate_->OnFailed(status);
    return;
  }
  // Complete any remaining read, as all data has been buffered.
  // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will
  // do nothing.
  timer_->Stop();
  DoBufferedRead();
}

void BidirectionalStreamSpdyJob::SendRequestHeaders() {
  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
  HttpRequestInfo http_request_info;
  http_request_info.url = request_info_->url;
  http_request_info.method = request_info_->method;
  http_request_info.extra_headers = request_info_->extra_headers;

  CreateSpdyHeadersFromHttpRequest(
      http_request_info, http_request_info.extra_headers,
      stream_->GetProtocolVersion(), true, headers.get());
  stream_->SendRequestHeaders(std::move(headers),
                              request_info_->end_stream_on_headers
                                  ? NO_MORE_DATA_TO_SEND
                                  : MORE_DATA_TO_SEND);
}

void BidirectionalStreamSpdyJob::OnStreamInitialized(int rv) {
  DCHECK_NE(ERR_IO_PENDING, rv);
  if (rv == OK) {
    stream_ = stream_request_.ReleaseStream();
    stream_->SetDelegate(this);
    SendRequestHeaders();
    return;
  }
  delegate_->OnFailed(rv);
}

void BidirectionalStreamSpdyJob::ScheduleBufferedRead() {
  // If there is already a scheduled DoBufferedRead, don't issue
  // another one. Mark that we have received more data and return.
  if (timer_->IsRunning()) {
    more_read_data_pending_ = true;
    return;
  }

  more_read_data_pending_ = false;
  timer_->Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kBufferTimeMs),
                base::Bind(&BidirectionalStreamSpdyJob::DoBufferedRead,
                           weak_factory_.GetWeakPtr()));
}

void BidirectionalStreamSpdyJob::DoBufferedRead() {
  DCHECK(!timer_->IsRunning());
  // Check to see that the stream has not errored out.
  DCHECK(stream_ || stream_closed_);
  DCHECK(!stream_closed_ || closed_stream_status_ == OK);

  // When |more_read_data_pending_| is true, it means that more data has arrived
  // since started waiting. Wait a little longer and continue to buffer.
  if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) {
    ScheduleBufferedRead();
    return;
  }

  int rv = 0;
  if (read_buffer_) {
    rv = ReadData(read_buffer_.get(), read_buffer_len_);
    DCHECK_NE(ERR_IO_PENDING, rv);
    read_buffer_ = nullptr;
    read_buffer_len_ = 0;
    delegate_->OnDataRead(rv);
  }
}

bool BidirectionalStreamSpdyJob::ShouldWaitForMoreBufferedData() const {
  if (stream_closed_)
    return false;
  DCHECK_GT(read_buffer_len_, 0);
  return read_data_queue_.GetTotalSize() <
         static_cast<size_t>(read_buffer_len_);
}

}  // namespace net