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

content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc

Index: content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc
diff --git a/content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc b/content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..2fd521f6d3e4044adaae6beeeb05e47d197f4635
--- /dev/null
+++ b/content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc
@@ -0,0 +1,411 @@
+// 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/run_loop.h"
+#include "content/browser/service_worker/embedded_worker_test_helper.h"
+#include "content/browser/service_worker/service_worker_context_core.h"
+#include "content/browser/service_worker/service_worker_context_wrapper.h"
+#include "content/browser/service_worker/service_worker_registration.h"
+#include "content/browser/service_worker/service_worker_url_request_job.h"
+#include "content/browser/service_worker/service_worker_version.h"
+#include "content/public/test/test_browser_thread_bundle.h"
+#include "net/base/io_buffer.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+const char kTestData[] = "Here is sample text for the blob.";
+
+}  // namespace
+
+class MockServiceWorkerURLRequestJob : public ServiceWorkerURLRequestJob {
+ public:
+  MockServiceWorkerURLRequestJob(ServiceWorkerURLRequestJob::Delegate* delegate)
+      : ServiceWorkerURLRequestJob(nullptr,
+                                   nullptr,
+                                   "",
+                                   nullptr,
+                                   nullptr,
+                                   FETCH_REQUEST_MODE_NO_CORS,
+                                   FETCH_CREDENTIALS_MODE_OMIT,
+                                   FetchRedirectMode::FOLLOW_MODE,
+                                   RESOURCE_TYPE_MAIN_FRAME,
+                                   REQUEST_CONTEXT_TYPE_HYPERLINK,
+                                   REQUEST_CONTEXT_FRAME_TYPE_TOP_LEVEL,
+                                   scoped_refptr<ResourceRequestBodyImpl>(),
+                                   ServiceWorkerFetchType::FETCH,
+                                   base::Optional<base::TimeDelta>(),
+                                   delegate),
+        is_response_started_(false) {}
+
+  void OnResponseStarted() override { is_response_started_ = true; }
+
+  void OnReadRawDataComplete(int bytes_read) override {
+    async_read_bytes_.push_back(bytes_read);
+  }
+
+  void RecordResult(ServiceWorkerMetrics::URLRequestJobResult result) override {
+    results_.push_back(result);
+  }
+
+  bool is_response_started() { return is_response_started_; }
+  const std::vector<int>& async_read_bytes() { return async_read_bytes_; }
+  const std::vector<ServiceWorkerMetrics::URLRequestJobResult>& results() {
+    return results_;
+  }
+
+ private:
+  bool is_response_started_;
+  std::vector<int> async_read_bytes_;
+  std::vector<ServiceWorkerMetrics::URLRequestJobResult> results_;
+};
+
+class ServiceWorkerDataPipeReaderTest
+    : public testing::Test,
+      public ServiceWorkerURLRequestJob::Delegate {
+ public:
+  ServiceWorkerDataPipeReaderTest()
+      : thread_bundle_(TestBrowserThreadBundle::IO_MAINLOOP) {}
+
+  void SetUp() override {
+    helper_ = base::MakeUnique<EmbeddedWorkerTestHelper>(base::FilePath());
+    mock_url_request_job_ =
+        base::MakeUnique<MockServiceWorkerURLRequestJob>(this);
+    registration_ = new ServiceWorkerRegistration(
+        GURL("https://example.com/"), 1L, helper_->context()->AsWeakPtr());
+    version_ = new ServiceWorkerVersion(
+        registration_.get(), GURL("https://example.com/service_worker.js"), 1L,
+        helper_->context()->AsWeakPtr());
+    std::vector<ServiceWorkerDatabase::ResourceRecord> records;
+    records.push_back(
+        ServiceWorkerDatabase::ResourceRecord(10, version_->script_url(), 100));
+    version_->script_cache_map()->SetResources(records);
+    version_->set_fetch_handler_existence(
+        ServiceWorkerVersion::FetchHandlerExistence::EXISTS);
+  }
+
+  std::unique_ptr<ServiceWorkerDataPipeReader> CreateTargetDataPipeReader(
+      blink::mojom::ServiceWorkerStreamCallbackPtr* stream_callback,
+      mojo::DataPipe* data_pipe) {
+    blink::mojom::ServiceWorkerStreamHandlePtr stream_handle =
+        blink::mojom::ServiceWorkerStreamHandle::New();
+    stream_handle->stream = std::move(data_pipe->consumer_handle);
+    stream_handle->callback_request = mojo::MakeRequest(stream_callback);
+    return base::MakeUnique<ServiceWorkerDataPipeReader>(
+        mock_url_request_job_.get(), version_, std::move(stream_handle));
+  }
+
+  // Implements ServiceWorkerURLRequestJob::Delegate.
+  void OnPrepareToRestart() override { NOTREACHED(); }
+
+  ServiceWorkerVersion* GetServiceWorkerVersion(
+      ServiceWorkerMetrics::URLRequestJobResult*) override {
+    NOTREACHED();
+    return nullptr;
+  }
+
+  bool RequestStillValid(ServiceWorkerMetrics::URLRequestJobResult*) override {
+    NOTREACHED();
+    return false;
+  }
+
+  void TearDown() override { helper_.reset(); }
+
+  MockServiceWorkerURLRequestJob* mock_url_request_job() {
+    return mock_url_request_job_.get();
+  }
+
+ protected:
+  TestBrowserThreadBundle thread_bundle_;
+
+  std::unique_ptr<EmbeddedWorkerTestHelper> helper_;
+  std::unique_ptr<MockServiceWorkerURLRequestJob> mock_url_request_job_;
+  scoped_refptr<ServiceWorkerRegistration> registration_;
+  scoped_refptr<ServiceWorkerVersion> version_;
+};
+
+class ServiceWorkerDataPipeReaderTestP
+    : public ServiceWorkerDataPipeReaderTest,
+      public testing::WithParamInterface<std::tuple<bool, bool>> {
+ public:
+  ServiceWorkerDataPipeReaderTestP() {}
+  virtual ~ServiceWorkerDataPipeReaderTestP() {}
+
+ protected:
+  bool is_closing_connection_first() const { return std::get<0>(GetParam()); }
+  bool is_body_exist() const { return std::get<1>(GetParam()); }
+};
+
+TEST_P(ServiceWorkerDataPipeReaderTestP, SyncRead) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
+      CreateTargetDataPipeReader(&stream_callback, &data_pipe);
+
+  // Push enough data.
+  if (is_body_exist()) {
+    std::string expected_response;
+    expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+    for (int i = 0; i < 1024; ++i) {
+      expected_response += kTestData;
+      uint32_t written_bytes = sizeof(kTestData) - 1;
+      MojoResult result =
+          mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                             &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+      ASSERT_EQ(MOJO_RESULT_OK, result);
+      EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+    }
+  }
+  data_pipe.producer_handle.reset();
+  stream_callback->OnCompleted();
+  base::RunLoop().RunUntilIdle();
+
+  // Nothing has started.
+  EXPECT_FALSE(mock_url_request_job()->is_response_started());
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(0UL, mock_url_request_job()->results().size());
+
+  // Start to read.
+  data_pipe_reader->Start();
+  EXPECT_TRUE(mock_url_request_job()->is_response_started());
+  const int buffer_size = sizeof(kTestData);
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
+  buffer->data()[buffer_size - 1] = '\0';
+
+  // Read successfully.
+  if (is_body_exist()) {
+    std::string retrieved_response;
+    retrieved_response.reserve(buffer_size * 1024);
+    for (int i = 0; i < 1024; ++i) {
+      EXPECT_EQ(buffer_size - 1,
+                data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
+      EXPECT_STREQ(kTestData, buffer->data());
+      retrieved_response += buffer->data();
+    }
+  }
+
+  // Finish successfully.
+  EXPECT_EQ(net::OK,
+            data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  ASSERT_EQ(1UL, mock_url_request_job()->results().size());
+  EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
+            mock_url_request_job()->results()[0]);
+}
+
+TEST_P(ServiceWorkerDataPipeReaderTestP, SyncAbort) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
+      CreateTargetDataPipeReader(&stream_callback, &data_pipe);
+
+  // Push enough data.
+  if (is_body_exist()) {
+    std::string expected_response;
+    expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+    for (int i = 0; i < 1024; ++i) {
+      expected_response += kTestData;
+      uint32_t written_bytes = sizeof(kTestData) - 1;
+      MojoResult result =
+          mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                             &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+      ASSERT_EQ(MOJO_RESULT_OK, result);
+      EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+    }
+  }
+  data_pipe.producer_handle.reset();
+  stream_callback->OnAborted();
+  base::RunLoop().RunUntilIdle();
+
+  // Nothing has started.
+  EXPECT_FALSE(mock_url_request_job()->is_response_started());
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(0UL, mock_url_request_job()->results().size());
+
+  // Start to read.
+  data_pipe_reader->Start();
+  EXPECT_TRUE(mock_url_request_job()->is_response_started());
+  const int buffer_size = sizeof(kTestData);
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
+  buffer->data()[buffer_size - 1] = '\0';
+
+  // Read successfully.
+  if (is_body_exist()) {
+    std::string retrieved_response;
+    retrieved_response.reserve(buffer_size * 1024);
+    for (int i = 0; i < 1024; ++i) {
+      EXPECT_EQ(buffer_size - 1,
+                data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
+      EXPECT_STREQ(kTestData, buffer->data());
+      retrieved_response += buffer->data();
+    }
+  }
+
+  // Abort after all data has been read.
+  EXPECT_EQ(net::ERR_CONNECTION_RESET,
+            data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  ASSERT_EQ(1UL, mock_url_request_job()->results().size());
+  EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
+            mock_url_request_job()->results()[0]);
+}
+
+TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncRead) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
+      CreateTargetDataPipeReader(&stream_callback, &data_pipe);
+
+  // Nothing has started.
+  EXPECT_FALSE(mock_url_request_job()->is_response_started());
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(0UL, mock_url_request_job()->results().size());
+
+  // Start to read.
+  data_pipe_reader->Start();
+  EXPECT_TRUE(mock_url_request_job()->is_response_started());
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
+  buffer->data()[sizeof(kTestData) - 1] = '\0';
+  std::string expected_response;
+  std::string retrieved_response;
+  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+  retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);
+
+  if (is_body_exist()) {
+    for (int i = 0; i < 1024; ++i) {
+      // Data is not coming. It should be pending state.
+      EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
+                                         buffer.get(), sizeof(kTestData) - 1));
+
+      // Push a portion of data.
+      uint32_t written_bytes = sizeof(kTestData) - 1;
+      MojoResult result =
+          mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                             &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+      ASSERT_EQ(MOJO_RESULT_OK, result);
+      EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+      expected_response += kTestData;
+      base::RunLoop().RunUntilIdle();
+
+      // Read the pushed data correctly.
+      ASSERT_EQ(static_cast<size_t>(i + 1),
+                mock_url_request_job()->async_read_bytes().size());
+      EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
+                mock_url_request_job()->async_read_bytes()[i]);
+      EXPECT_STREQ(kTestData, buffer->data());
+    }
+  }
+
+  // Data is not coming. It should be pending state.
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));
+
+  // Finish successfully when connection is closed AND OnCompleted is delivered.
+  size_t num_read = mock_url_request_job()->async_read_bytes().size();
+  if (is_closing_connection_first()) {
+    data_pipe.producer_handle.reset();
+  } else {
+    stream_callback->OnCompleted();
+  }
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
+  ASSERT_EQ(0UL, mock_url_request_job()->results().size());
+
+  if (is_closing_connection_first()) {
+    stream_callback->OnCompleted();
+  } else {
+    data_pipe.producer_handle.reset();
+  }
+  base::RunLoop().RunUntilIdle();
+  ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(net::OK, mock_url_request_job()->async_read_bytes().back());
+  ASSERT_EQ(1UL, mock_url_request_job()->results().size());
+  EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
+            mock_url_request_job()->results()[0]);
+}
+
+TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncAbort) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
+      CreateTargetDataPipeReader(&stream_callback, &data_pipe);
+
+  // Nothing has started.
+  EXPECT_FALSE(mock_url_request_job()->is_response_started());
+  EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(0UL, mock_url_request_job()->results().size());
+
+  // Start to read.
+  data_pipe_reader->Start();
+  EXPECT_TRUE(mock_url_request_job()->is_response_started());
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
+  buffer->data()[sizeof(kTestData) - 1] = '\0';
+  std::string expected_response;
+  std::string retrieved_response;
+  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+  retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);
+
+  if (is_body_exist()) {
+    for (int i = 0; i < 1024; ++i) {
+      // Data is not coming. It should be pending state.
+      EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
+                                         buffer.get(), sizeof(kTestData) - 1));
+
+      // Push a portion of data.
+      uint32_t written_bytes = sizeof(kTestData) - 1;
+      MojoResult result =
+          mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                             &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+      ASSERT_EQ(MOJO_RESULT_OK, result);
+      EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+      expected_response += kTestData;
+      base::RunLoop().RunUntilIdle();
+
+      // Read the pushed data correctly.
+      ASSERT_EQ(static_cast<size_t>(i + 1),
+                mock_url_request_job()->async_read_bytes().size());
+      EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
+                mock_url_request_job()->async_read_bytes()[i]);
+      EXPECT_STREQ(kTestData, buffer->data());
+    }
+  }
+
+  // Data is not coming. It should be pending state.
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));
+
+  // Abort when connection is closed AND OnAborted is delivered.
+  size_t num_read = mock_url_request_job()->async_read_bytes().size();
+  if (is_closing_connection_first()) {
+    data_pipe.producer_handle.reset();
+  } else {
+    stream_callback->OnAborted();
+  }
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
+  ASSERT_EQ(0UL, mock_url_request_job()->results().size());
+
+  if (is_closing_connection_first()) {
+    stream_callback->OnAborted();
+  } else {
+    data_pipe.producer_handle.reset();
+  }
+  base::RunLoop().RunUntilIdle();
+  ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
+  EXPECT_EQ(net::ERR_CONNECTION_RESET,
+            mock_url_request_job()->async_read_bytes().back());
+  ASSERT_EQ(1UL, mock_url_request_job()->results().size());
+  EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
+            mock_url_request_job()->results()[0]);
+}
+
+INSTANTIATE_TEST_CASE_P(ServiceWorkerDataPipeReaderTest,
+                        ServiceWorkerDataPipeReaderTestP,
+                        testing::Combine(testing::Bool(), testing::Bool()));
+
+}  // namespace content