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/memory/scoped_ptr.h"
+#include "base/time/time.h"
+#include "base/timer/timer.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);
+
+BidirectionalStreamSpdyJob::BidirectionalStreamSpdyJob(
+    const base::WeakPtr<SpdySession>& spdy_session,
+    scoped_ptr<base::Timer> timer)
+    : spdy_session_(spdy_session),
+      timer_(timer.release()),
+      stream_closed_(false),
+      closed_stream_status_(ERR_FAILED),
+      more_read_data_pending_(false),
+      weak_factory_(this) {}
+
+BidirectionalStreamSpdyJob::BidirectionalStreamSpdyJob(
+    const base::WeakPtr<SpdySession>& spdy_session)
+    : BidirectionalStreamSpdyJob(
+          spdy_session,
+          make_scoped_ptr(new base::Timer(false, false))) {}
+
+BidirectionalStreamSpdyJob::~BidirectionalStreamSpdyJob() {
+  if (stream_.get()) {
+    stream_->DetachDelegate();
+    DCHECK(!stream_.get());
+  }
+}
+
+void BidirectionalStreamSpdyJob::Start(
+    const HttpRequestInfo& request_info,
+    RequestPriority priority,
+    const BoundNetLog& net_log,
+    BidirectionalStreamJob::Delegate* delegate) {
+  delegate_ = delegate;
+  DCHECK(!stream_);
+  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_.get())
+    CHECK(!stream_->IsIdle());
+
+  CHECK(buf);
+  CHECK(buf_len);
+  CHECK(!timer_->IsRunning()) << "There should be only one ReadData in flight";
+
+  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_;

[mef, 2015/12/07 21:18:07]

Shouldn't it return 0? 
The comment says that it returns number of bytes read.
This would explain why I was getting -137 when server was sending 2 header
frames in the row.

[xunjieli, 2015/12/08 16:08:46]

It should return a non-negative number that indicates the number of bytes read
or a negative number (e.g. ERR_IO_PENDING) if any error occurred. I have changed
the comment accordingly.

+  }
+  // Read will complete asynchronously and Delegate::OnReadCompleted will be
+  // called upon completion.
+  user_buffer_ = buf;
+  user_buffer_len_ = buf_len;
+  return ERR_IO_PENDING;
+}
+
+void BidirectionalStreamSpdyJob::SendData(IOBuffer* data,
+                                          int length,
+                                          bool end_stream) {
+  CHECK(!stream_closed_);
+  CHECK(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_);
+}
+
+void BidirectionalStreamSpdyJob::OnRequestHeadersSent() {
+  delegate_->OnRequestHeadersSent();
+}
+
+SpdyResponseHeadersStatus BidirectionalStreamSpdyJob::OnResponseHeadersUpdated(
+    const SpdyHeaderBlock& response_headers) {
+  delegate_->OnHeaders(response_headers);
+  return RESPONSE_HEADERS_ARE_COMPLETE;
+}
+
+void BidirectionalStreamSpdyJob::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
+  DCHECK(stream_);
+  DCHECK(!stream_closed_);
+
+  if (buffer) {
+    data_queue_.Enqueue(buffer.Pass());
+    if (user_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();
+    }
+  }
+  // If |buffer| is null, BidirectionalStreamSpdyJob::OnClose will be invoked by
+  // SpdyStream to indicate the end of stream.
+}
+
+void BidirectionalStreamSpdyJob::OnDataSent() {
+  DCHECK(stream_);
+  DCHECK(!stream_closed_);
+
+  delegate_->OnDataSent();
+}
+
+void BidirectionalStreamSpdyJob::OnTrailers(const SpdyHeaderBlock& trailers) {
+  DCHECK(stream_);
+  DCHECK(!stream_closed_);
+
+  delegate_->OnTrailers(trailers);
+}
+
+void BidirectionalStreamSpdyJob::OnClose(int status) {
+  DCHECK(stream_);
+
+  stream_closed_ = true;
+  closed_stream_status_ = status;
+  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 |user_buffer_| is nullptr), this will
+  // do nothing.
+  DCHECK_EQ(OK, status);
+  timer_->Stop();
+  DoBufferedRead();
+}
+
+void BidirectionalStreamSpdyJob::SendRequestHeaders() {
+  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");

[mef, 2015/12/07 21:18:07]

I suggest that we add at least 'HEAD', but I would also consider adding an
explicit flag whether or not there is data to write as we have to have the same
logic on Cronet Java side to figure out whether we should wait for writes.

[xunjieli, 2015/12/08 16:08:46]

Done.

[mef, 2015/12/11 20:06:12]

Per our chat I've meant that we need to pass an explicit 'end_stream' flag.

[xunjieli, 2015/12/11 23:48:39]

Done.

+  stream_->SendRequestHeaders(
+      headers.Pass(), end_stream ? 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, kBufferTime,
+                base::Bind(&BidirectionalStreamSpdyJob::DoBufferedRead,
+                           weak_factory_.GetWeakPtr()));
+}
+
+void BidirectionalStreamSpdyJob::DoBufferedRead() {
+  DCHECK(!timer_->IsRunning());
+  // 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()) {
+    ScheduleBufferedRead();
+    return;
+  }
+
+  int rv = 0;
+  if (user_buffer_) {
+    rv = ReadData(user_buffer_.get(), user_buffer_len_);
+    DCHECK_NE(ERR_IO_PENDING, rv);
+    user_buffer_ = nullptr;
+    user_buffer_len_ = 0;
+    delegate_->OnReadCompleted(rv);
+  }
+}
+
+bool BidirectionalStreamSpdyJob::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