ServiceWorker: Use mojo's data pipe for respondWith(stream)

content/browser/service_worker/service_worker_data_pipe_reader.cc

+// Copyright 2017 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 "content/browser/service_worker/service_worker_data_pipe_reader.h"
+
+#include "base/trace_event/trace_event.h"
+#include "content/browser/service_worker/service_worker_url_request_job.h"
+#include "content/browser/service_worker/service_worker_version.h"
+#include "net/base/io_buffer.h"
+
+namespace content {
+
+ServiceWorkerDataPipeReader::ServiceWorkerDataPipeReader(
+    ServiceWorkerURLRequestJob* owner,
+    scoped_refptr<ServiceWorkerVersion> streaming_version,
+    blink::mojom::ServiceWorkerStreamHandlePtr stream_handle)
+    : owner_(owner),
+      streaming_version_(streaming_version),
+      handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::AUTOMATIC),
+      stream_(std::move(stream_handle->stream)),
+      binding_(this, std::move(stream_handle->callback_request)),
+      producer_state_(State::STREAMING) {
+  TRACE_EVENT_ASYNC_BEGIN1("ServiceWorker", "ServiceWorkerDataPipeReader", this,
+                           "Url", owner->request()->url().spec());
+  streaming_version_->AddStreamingURLRequestJob(owner_);
+  binding_.set_connection_error_handler(base::Bind(
+      &ServiceWorkerDataPipeReader::OnAborted, base::Unretained(this)));
+}
+
+ServiceWorkerDataPipeReader::~ServiceWorkerDataPipeReader() {
+  if (streaming_version_) {
+    streaming_version_->RemoveStreamingURLRequestJob(owner_);
+    streaming_version_ = nullptr;
+  }
+  TRACE_EVENT_ASYNC_END0("ServiceWorker", "ServiceWorkerDataPipeReader", this);
+}
+
+void ServiceWorkerDataPipeReader::Start() {
+  TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                               this, "Start");
+  handle_watcher_.Watch(
+      stream_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      base::Bind(&ServiceWorkerDataPipeReader::OnHandleGotReadable,
+                 base::Unretained(this)));
+  owner_->OnResponseStarted();
+}
+
+void ServiceWorkerDataPipeReader::OnHandleGotReadable(MojoResult) {

[horo, 2017/04/06 05:22:08]

DCHECK_EQ(State::STREAMING, producer_state_)?

[shimazu, 2017/04/07 08:52:29]

Done.

+  TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                               this, "OnHandleGotReadable");
+  // Do nothing if stream_pending_buffer_ is empty, i.e. there's no ReadRawData
+  // operation waiting for IO completion.
+  if (!stream_pending_buffer_)
+    return;
+
+  // |stream_pending_buffer_| is set to the IOBuffer instance provided to
+  // ReadRawData() by URLRequestJob.
+  uint32_t size_to_pass = stream_pending_buffer_size_;
+  MojoResult mojo_result =
+      mojo::ReadDataRaw(stream_.get(), stream_pending_buffer_->data(),
+                        &size_to_pass, MOJO_READ_DATA_FLAG_NONE);
+
+  switch (mojo_result) {
+    case MOJO_RESULT_OK:
+      stream_pending_buffer_ = nullptr;
+      stream_pending_buffer_size_ = 0;
+      owner_->OnReadRawDataComplete(size_to_pass);
+      return;
+    case MOJO_RESULT_FAILED_PRECONDITION:
+      stream_.reset();
+      handle_watcher_.Cancel();
+      MaybeComplete();
+      return;
+    case MOJO_RESULT_SHOULD_WAIT:
+      return;
+    case MOJO_RESULT_BUSY:
+    case MOJO_RESULT_RESOURCE_EXHAUSTED:

[horo, 2017/04/06 05:22:08]

mojo::ReadDataRaw doesn't return MOJO_RESULT_RESOURCE_EXHAUSTED.

https://chromium.googlesource.com/chromium/src/+/dd3c9a967d856c4dc8b41fc1738c...

[shimazu, 2017/04/07 08:52:29]

Done.

+      stream_pending_buffer_ = nullptr;
+      stream_pending_buffer_size_ = 0;
+      stream_.reset();
+      handle_watcher_.Cancel();
+      owner_->OnReadRawDataComplete(net::ERR_FAILED);
+      return;
+  }

[horo, 2017/04/06 05:22:08]

ASSERT_NOT_REACHED();

[shimazu, 2017/04/07 08:52:29]

Done.

+}
+
+int ServiceWorkerDataPipeReader::ReadRawData(net::IOBuffer* buf, int buf_size) {
+  TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                               this, "ReadRawData");
+  if (producer_state_ == State::ABORTED) {
+    // ABORT has been already notified.
+    // Handle this as connection reset.
+    stream_.reset();
+    handle_watcher_.Cancel();
+    owner_->RecordResult(
+        ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED);
+    return net::ERR_CONNECTION_RESET;
+  }
+
+  uint32_t size_to_pass = buf_size;
+  MojoResult mojo_result = mojo::ReadDataRaw(
+      stream_.get(), buf->data(), &size_to_pass, MOJO_READ_DATA_FLAG_NONE);
+  switch (mojo_result) {
+    case MOJO_RESULT_OK:
+      return size_to_pass;
+    case MOJO_RESULT_FAILED_PRECONDITION:
+      stream_.reset();
+      handle_watcher_.Cancel();
+      if (state() == State::COMPLETED)
+        return 0;
+      // Complete/Abort asynchronously.
+      stream_pending_buffer_ = buf;
+      stream_pending_buffer_size_ = buf_size;
+      return net::ERR_IO_PENDING;
+    case MOJO_RESULT_BUSY:
+      return net::ERR_FAILED;
+    case MOJO_RESULT_SHOULD_WAIT:
+      stream_pending_buffer_ = buf;
+      stream_pending_buffer_size_ = buf_size;
+      return net::ERR_IO_PENDING;
+    case MOJO_RESULT_RESOURCE_EXHAUSTED:

[horo, 2017/04/06 05:22:08]

mojo::ReadDataRaw doesn't return MOJO_RESULT_RESOURCE_EXHAUSTED.

[shimazu, 2017/04/07 08:52:29]

Done.

+      return net::ERR_FAILED;
+  }
+  return net::ERR_FAILED;

[horo, 2017/04/06 05:22:08]

ASSERT_NOT_REACHED();?

[shimazu, 2017/04/07 08:52:29]

Done.

+}
+
+void ServiceWorkerDataPipeReader::OnCompleted() {
+  producer_state_ = State::COMPLETED;
+  MaybeComplete();
+}
+
+void ServiceWorkerDataPipeReader::OnAborted() {
+  producer_state_ = State::ABORTED;
+  MaybeComplete();
+}
+
+void ServiceWorkerDataPipeReader::MaybeComplete() {
+  // This works only after ReadRawData returns net::ERR_IO_PENDING.
+  if (!stream_pending_buffer_)
+    return;
+
+  switch (state()) {
+    case State::STREAMING:
+      return;
+    case State::COMPLETED:
+      stream_pending_buffer_ = nullptr;
+      stream_pending_buffer_size_ = 0;
+      handle_watcher_.Cancel();
+      owner_->RecordResult(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE);
+      owner_->OnReadRawDataComplete(0);
+      return;
+    case State::ABORTED:
+      stream_pending_buffer_ = nullptr;
+      stream_pending_buffer_size_ = 0;
+      handle_watcher_.Cancel();
+      owner_->RecordResult(
+          ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED);
+      owner_->OnReadRawDataComplete(net::ERR_CONNECTION_RESET);
+      return;
+  }
+}
+
+ServiceWorkerDataPipeReader::State ServiceWorkerDataPipeReader::state() {
+  if (!stream_.is_valid())
+    return producer_state_;
+  return State::STREAMING;
+}
+
+}  // namespace content