Added a net::BidirectionalStream to expose a bidirectional streaming interface

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/base/request_priority.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 {
+
+const int64 kBufferTimeMs = 1;
+
+BidirectionalStreamSpdyJob::BidirectionalStreamSpdyJob(
+    const base::WeakPtr<SpdySession>& spdy_session)
+    : spdy_session_(spdy_session),
+      stream_closed_(false),
+      closed_stream_status_(ERR_FAILED),
+      more_read_data_pending_(false),
+      negotiated_protocol_(kProtoUnknown),
+      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,
+    RequestPriority priority,
+    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, 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 (!stream_closed_)
+    DCHECK(stream_);
+
+  // If there is data buffered, complete the IO immediately.
+  if (!data_queue_.IsEmpty()) {
+    return 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.
+  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.
+  DCHECK_EQ(OK, status);
+  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(static_cast<Error>(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 data_queue_.GetTotalSize() < static_cast<size_t>(read_buffer_len_);
+}
+
+}  // namespace net