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

net/spdy/bidirectional_spdy_stream.cc

Index: net/spdy/bidirectional_spdy_stream.cc
diff --git a/net/spdy/bidirectional_spdy_stream.cc b/net/spdy/bidirectional_spdy_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..077081534cb807b72f1b54f5f0222bf504efd183
--- /dev/null
+++ b/net/spdy/bidirectional_spdy_stream.cc
@@ -0,0 +1,194 @@
+// 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_spdy_stream.h"
+
+#include "base/memory/scoped_ptr.h"
+#include "base/time/time.h"
+#include "net/base/request_priority.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 base::TimeDelta kBufferTime = base::TimeDelta::FromMilliseconds(1);
+
+BidirectionalSpdyStream::BidirectionalSpdyStream(
+    const base::WeakPtr<SpdySession>& spdy_session)
+    : spdy_session_(spdy_session),
+      stream_closed_(false),
+      closed_stream_status_(ERR_FAILED),
+      buffered_read_callback_pending_(false),
+      more_read_data_pending_(false),
+      weak_factory_(this) {}
+
+BidirectionalSpdyStream::~BidirectionalSpdyStream() {
+  if (stream_.get()) {
+    stream_->DetachDelegate();
+    DCHECK(!stream_.get());
+  }
+}
+
+void BidirectionalSpdyStream::Start(const HttpRequestInfo* request_info,
+                                    RequestPriority priority,
+                                    const BoundNetLog& net_log,
+                                    BidirectionalStream::Delegate* delegate) {
+  delegate_ = delegate;
+  DCHECK(!stream_);
+  if (!spdy_session_)
+    delegate_->OnFailed(ERR_CONNECTION_CLOSED);
+
+  request_info_ = request_info;
+
+  int rv = stream_request_.StartRequest(
+      SPDY_REQUEST_RESPONSE_STREAM, spdy_session_, request_info_->url, priority,
+      net_log, base::Bind(&BidirectionalSpdyStream::OnStreamInitialized,
+                          weak_factory_.GetWeakPtr()));
+  if (rv != ERR_IO_PENDING)
+    OnStreamInitialized(rv);
+}
+
+int BidirectionalSpdyStream::ReadData(IOBuffer* buf, int buf_len) {
+  if (stream_.get())
+    CHECK(!stream_->IsIdle());
+
+  CHECK(buf);
+  CHECK(buf_len);
+  if (!stream_closed_)
+    CHECK(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_;
+  }
+  user_buffer_ = buf;
+  user_buffer_len_ = buf_len;
+  return ERR_IO_PENDING;
+}
+
+void BidirectionalSpdyStream::SendData(IOBuffer* data,
+                                       int length,
+                                       bool end_stream) {
+  stream_->SendData(data, length,
+                    end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
+}
+
+void BidirectionalSpdyStream::OnRequestHeadersSent() {
+  delegate_->OnRequestHeadersSent();
+}
+
+SpdyResponseHeadersStatus BidirectionalSpdyStream::OnResponseHeadersUpdated(
+    const SpdyHeaderBlock& response_headers) {
+  delegate_->OnHeaders(response_headers);
+  return RESPONSE_HEADERS_ARE_COMPLETE;
+}
+
+void BidirectionalSpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
+  DCHECK(stream_);
+  data_queue_.Enqueue(buffer.Pass());
+  if (user_buffer_.get()) {
+    // Handing small chunks of data to the caller creates measurable overhead.
+    // So buffer data in short time-spans and send a single read notification.
+    ScheduleBufferedReadCallback();
+  }
+}
+
+void BidirectionalSpdyStream::OnDataSent() {
+  delegate_->OnDataSent();
+}
+
+void BidirectionalSpdyStream::OnTrailers(const SpdyHeaderBlock& trailers) {
+  delegate_->OnTrailers(trailers);
+}
+
+void BidirectionalSpdyStream::OnClose(int status) {
+  if (stream_.get()) {
+    stream_closed_ = true;
+    closed_stream_status_ = status;
+  }
+
+  stream_.reset();
+  // Complete remaining buffered read.

[mef, 2015/10/07 23:44:56]

what if there is no pending read?

[xunjieli, 2015/10/19 21:07:46]

Done. Good catch! I added null check to handle this case.

+  if (status == OK) {
+    DoBufferedReadCallback();
+    return;
+  }
+
+  delegate_->OnClose(status);
+}
+
+void BidirectionalSpdyStream::SendRequestHeaders() {
+  stream_->SetDelegate(this);
+  scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
+  CreateSpdyHeadersFromHttpRequest(*request_info_, request_info_->extra_headers,
+                                   stream_->GetProtocolVersion(), true,
+                                   headers.get());
+  bool end_stream = (request_info_->method == "GET");
+  stream_->SendRequestHeaders(
+      headers.Pass(), end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
+}
+
+void BidirectionalSpdyStream::OnStreamInitialized(int rv) {
+  DCHECK_NE(ERR_IO_PENDING, rv);
+  if (rv == OK) {
+    stream_ = stream_request_.ReleaseStream();
+    SendRequestHeaders();
+    return;
+  }
+  delegate_->OnFailed(rv);
+}
+
+void BidirectionalSpdyStream::ScheduleBufferedReadCallback() {
+  // If there is already a scheduled DoBufferedReadCallback, don't issue
+  // another one. Mark that we have received more data and return.
+  if (buffered_read_callback_pending_) {
+    more_read_data_pending_ = true;
+    return;
+  }
+
+  more_read_data_pending_ = false;
+  buffered_read_callback_pending_ = true;
+  base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
+      FROM_HERE, base::Bind(&BidirectionalSpdyStream::DoBufferedReadCallback,
+                            weak_factory_.GetWeakPtr()),
+      kBufferTime);
+}
+
+void BidirectionalSpdyStream::DoBufferedReadCallback() {
+  buffered_read_callback_pending_ = false;
+  // If the stream errored out, do not complete the read.
+  if (!stream_ && !stream_closed_)
+    return;
+  if (stream_closed_ && closed_stream_status_ != OK)
+    return;
+
+  // 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()) {
+    ScheduleBufferedReadCallback();
+    return;
+  }
+
+  if (!user_buffer_.get())
+    return;
+
+  int rv = ReadData(user_buffer_.get(), user_buffer_len_);
+  DCHECK_NE(ERR_IO_PENDING, rv);
+  delegate_->OnReadCompleted(rv);
+  if (data_queue_.IsEmpty() && stream_closed_)
+    delegate_->OnClose(closed_stream_status_);
+}
+
+bool BidirectionalSpdyStream::ShouldWaitForMoreBufferedData() const {
+  if (stream_closed_)
+    return false;
+  DCHECK_GT(user_buffer_len_, 0);
+  return data_queue_.GetTotalSize() < static_cast<size_t>(user_buffer_len_);
+}
+
+}  // namespace net