Add ServiceWorkerResourceReader::ReadData()

This CL adds a method called ReadData() that returns a data pipe
consumer handle. The producer of the data pipe supplies the content of
a service worker script from storage. See [1] for the detail of the
proposal.

The internal implementation is almost the same as
ServiceWorkerInstalledScriptReader. Once ServiceWorkerStorage is
migrated to the storage service, we could stop using
ServiceWorkerInstalledScriptReader and pass the data pipe directly to
renderers.

This CL also moves ReadDataPipe() which was defined in
service_worker_installed_scripts_sender_unittest.cc to
service_worker_test_utils.cc as the function is useful to test
ServiceWorkerStorageControlImpl.

[1] https://docs.google.com/document/d/1UhdZnaeEP8GywZR0Gik6wF6jKmNY3je6aDhPfJ05YhU/edit#heading=h.z6754cqqpv4k

Bug: 1055677
Change-Id: I05aa5a27fabde65102ae3e34781ea1cf39e8b8f5
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2138677
Commit-Queue: Kenichi Ishibashi <bashi@chromium.org>
Reviewed-by: Makoto Shimazu <shimazu@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Cr-Commit-Position: refs/heads/master@{#757670}

components/services/storage/public/mojom/service_worker_storage_control.mojom

@@ -41,8 +41,9 @@ interface ServiceWorkerResourceReader {
   // Reads the response head of the resource associated with this reader.
   ReadResponseHead() =>
       (int32 status, network.mojom.URLResponseHead response_head);
-  // TODO(crbug.com/1055677): Add ways to read the content and the metadata of
-  // the script.
+  // Reads the content of the resource associated with this reader.
+  ReadData(int64 size) => (handle<data_pipe_consumer> pipe);
+  // TODO(crbug.com/1055677): Add ways to read the metadata of the script.
 };
 
 // An interface that writes a service worker script (resource) to storage.

content/browser/service_worker/service_worker_installed_scripts_sender_unittest.cc

@@ -25,56 +25,6 @@
 
 namespace content {
 
-namespace {
-
-void ReadDataPipeInternal(mojo::DataPipeConsumerHandle handle,
-                          std::string* result,
-                          base::OnceClosure quit_closure) {
-  while (true) {
-    uint32_t num_bytes;
-    const void* buffer = nullptr;
-    MojoResult rv =
-        handle.BeginReadData(&buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
-    switch (rv) {
-      case MOJO_RESULT_BUSY:
-      case MOJO_RESULT_INVALID_ARGUMENT:
-        NOTREACHED();
-        return;
-      case MOJO_RESULT_FAILED_PRECONDITION:
-        std::move(quit_closure).Run();
-        return;
-      case MOJO_RESULT_SHOULD_WAIT:
-        base::ThreadTaskRunnerHandle::Get()->PostTask(
-            FROM_HERE, base::BindOnce(&ReadDataPipeInternal, handle, result,
-                                      std::move(quit_closure)));
-        return;
-      case MOJO_RESULT_OK:
-        EXPECT_NE(nullptr, buffer);
-        EXPECT_GT(num_bytes, 0u);
-        uint32_t before_size = result->size();
-        result->append(static_cast<const char*>(buffer), num_bytes);
-        uint32_t read_size = result->size() - before_size;
-        EXPECT_EQ(num_bytes, read_size);
-        rv = handle.EndReadData(read_size);
-        EXPECT_EQ(MOJO_RESULT_OK, rv);
-        break;
-    }
-  }
-  NOTREACHED();
-  return;
-}
-
-std::string ReadDataPipe(mojo::ScopedDataPipeConsumerHandle handle) {
-  EXPECT_TRUE(handle.is_valid());
-  std::string result;
-  base::RunLoop loop;
-  ReadDataPipeInternal(handle.get(), &result, loop.QuitClosure());
-  loop.Run();
-  return result;
-}
-
-}  // namespace
-
 class ExpectedScriptInfo {
  public:
   ExpectedScriptInfo(

content/browser/service_worker/service_worker_resource_ops.cc

@@ -5,6 +5,9 @@
 #include "content/browser/service_worker/service_worker_resource_ops.h"
 
 #include "content/browser/service_worker/service_worker_loader_helpers.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "services/network/public/cpp/net_adapters.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 
 namespace content {
 
@@ -75,6 +78,103 @@ void DidReadInfo(
 
 }  // namespace
 
+class ServiceWorkerResourceReaderImpl::DataReader {
+ public:
+  DataReader(base::WeakPtr<ServiceWorkerResourceReaderImpl> owner,
+             size_t total_bytes_to_read,
+             mojo::ScopedDataPipeProducerHandle producer_handle)
+      : owner_(std::move(owner)),
+        total_bytes_to_read_(total_bytes_to_read),
+        producer_handle_(std::move(producer_handle)),
+        watcher_(FROM_HERE,
+                 mojo::SimpleWatcher::ArmingPolicy::MANUAL,
+                 base::SequencedTaskRunnerHandle::Get()) {
+    watcher_.Watch(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+                   base::BindRepeating(&DataReader::OnWritable,
+                                       weak_factory_.GetWeakPtr()));
+    watcher_.ArmOrNotify();
+  }
+  ~DataReader() = default;
+
+  DataReader(const DataReader&) = delete;
+  DataReader operator=(const DataReader&) = delete;
+
+ private:
+  void OnWritable(MojoResult) {
+    DCHECK(producer_handle_.is_valid());
+    DCHECK(!pending_buffer_);
+
+    if (!owner_) {
+      Complete(net::ERR_ABORTED);
+      return;
+    }
+
+    uint32_t num_bytes = 0;
+    MojoResult rv = network::NetToMojoPendingBuffer::BeginWrite(
+        &producer_handle_, &pending_buffer_, &num_bytes);
+    switch (rv) {
+      case MOJO_RESULT_INVALID_ARGUMENT:
+      case MOJO_RESULT_BUSY:
+        NOTREACHED();
+        return;
+      case MOJO_RESULT_FAILED_PRECONDITION:
+        Complete(net::ERR_ABORTED);
+        return;
+      case MOJO_RESULT_SHOULD_WAIT:
+        watcher_.ArmOrNotify();
+        return;
+      case MOJO_RESULT_OK:
+        // |producer__handle_| must have been taken by |pending_buffer_|.
+        DCHECK(pending_buffer_);
+        DCHECK(!producer_handle_.is_valid());
+        break;
+    }
+
+    num_bytes = std::min(num_bytes, blink::BlobUtils::GetDataPipeChunkSize());
+    auto buffer =
+        base::MakeRefCounted<network::NetToMojoIOBuffer>(pending_buffer_.get());
+    owner_->reader_->ReadData(
+        buffer.get(), num_bytes,
+        base::BindOnce(&DataReader::DidReadData, weak_factory_.GetWeakPtr()));
+  }
+
+  void DidReadData(int read_bytes) {
+    if (read_bytes < 0) {
+      Complete(read_bytes);
+      return;
+    }
+
+    producer_handle_ = pending_buffer_->Complete(read_bytes);
+    DCHECK(producer_handle_.is_valid());
+    pending_buffer_.reset();
+    current_bytes_read_ += read_bytes;
+
+    if (read_bytes == 0 || current_bytes_read_ == total_bytes_to_read_) {
+      // All data has been read.
+      Complete(current_bytes_read_);
+      return;
+    }
+    watcher_.ArmOrNotify();
+  }
+
+  void Complete(int status) {
+    watcher_.Cancel();
+    producer_handle_.reset();
+    if (owner_) {
+      owner_->DidReadDataComplete();
+    }
+  }
+
+  base::WeakPtr<ServiceWorkerResourceReaderImpl> owner_;
+  const size_t total_bytes_to_read_;
+  size_t current_bytes_read_ = 0;
+  mojo::ScopedDataPipeProducerHandle producer_handle_;
+  mojo::SimpleWatcher watcher_;
+  scoped_refptr<network::NetToMojoPendingBuffer> pending_buffer_;
+
+  base::WeakPtrFactory<DataReader> weak_factory_{this};
+};
+
 ServiceWorkerResourceReaderImpl::ServiceWorkerResourceReaderImpl(
     std::unique_ptr<ServiceWorkerResponseReader> reader)
     : reader_(std::move(reader)) {
@@ -91,6 +191,33 @@ void ServiceWorkerResourceReaderImpl::ReadResponseHead(
                                                std::move(callback)));
 }
 
+void ServiceWorkerResourceReaderImpl::ReadData(int64_t size,
+                                               ReadDataCallback callback) {
+  MojoCreateDataPipeOptions options;
+  options.struct_size = sizeof(MojoCreateDataPipeOptions);
+  options.flags = MOJO_CREATE_DATA_PIPE_FLAG_NONE;
+  options.element_num_bytes = 1;
+  options.capacity_num_bytes = blink::BlobUtils::GetDataPipeCapacity(size);
+
+  mojo::ScopedDataPipeConsumerHandle consumer_handle;
+  mojo::ScopedDataPipeProducerHandle producer_handle;
+  MojoResult rv =
+      mojo::CreateDataPipe(&options, &producer_handle, &consumer_handle);
+  if (rv != MOJO_RESULT_OK) {
+    std::move(callback).Run(mojo::ScopedDataPipeConsumerHandle());
+    return;
+  }
+
+  data_reader_ = std::make_unique<DataReader>(weak_factory_.GetWeakPtr(), size,
+                                              std::move(producer_handle));
+  std::move(callback).Run(std::move(consumer_handle));
+}
+
+void ServiceWorkerResourceReaderImpl::DidReadDataComplete() {
+  DCHECK(data_reader_);
+  data_reader_.reset();
+}
+
 ServiceWorkerResourceWriterImpl::ServiceWorkerResourceWriterImpl(
     std::unique_ptr<ServiceWorkerResponseWriter> writer)
     : writer_(std::move(writer)) {

content/browser/service_worker/service_worker_resource_ops.h

@@ -5,6 +5,7 @@
 #ifndef CONTENT_BROWSER_SERVICE_WORKER_SERVICE_WORKER_RESOURCE_OPS_H_
 #define CONTENT_BROWSER_SERVICE_WORKER_SERVICE_WORKER_RESOURCE_OPS_H_
 
+#include "base/memory/weak_ptr.h"
 #include "components/services/storage/public/mojom/service_worker_storage_control.mojom.h"
 #include "content/browser/service_worker/service_worker_disk_cache.h"
 
@@ -31,8 +32,16 @@ class ServiceWorkerResourceReaderImpl
  private:
   // storage::mojom::ServiceWorkerResourceReader implementations:
   void ReadResponseHead(ReadResponseHeadCallback callback) override;
+  void ReadData(int64_t size, ReadDataCallback callback) override;
+
+  void DidReadDataComplete();
 
   const std::unique_ptr<ServiceWorkerResponseReader> reader_;
+
+  class DataReader;
+  std::unique_ptr<DataReader> data_reader_;
+
+  base::WeakPtrFactory<ServiceWorkerResourceReaderImpl> weak_factory_{this};
 };
 
 // The implementation of storage::mojom::ServiceWorkerResourceWriter.

content/browser/service_worker/service_worker_storage_control_impl_unittest.cc

@@ -9,8 +9,10 @@
 #include "base/test/bind_test_util.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "content/browser/service_worker/service_worker_storage.h"
+#include "content/browser/service_worker/service_worker_test_utils.h"
 #include "content/public/test/browser_task_environment.h"
 #include "content/public/test/test_utils.h"
+#include "mojo/public/cpp/system/data_pipe_utils.h"
 #include "net/disk_cache/disk_cache.h"
 #include "net/http/http_util.h"
 #include "net/test/cert_test_util.h"
@@ -40,6 +42,21 @@ int ReadResponseHead(storage::mojom::ServiceWorkerResourceReader* reader,
   return return_value;
 }
 
+std::string ReadResponseData(
+    storage::mojom::ServiceWorkerResourceReader* reader,
+    int data_size) {
+  mojo::ScopedDataPipeConsumerHandle data_consumer;
+  base::RunLoop loop;
+  reader->ReadData(data_size, base::BindLambdaForTesting(
+                                  [&](mojo::ScopedDataPipeConsumerHandle pipe) {
+                                    data_consumer = std::move(pipe);
+                                    loop.Quit();
+                                  }));
+  loop.Run();
+
+  return ReadDataPipe(std::move(data_consumer));
+}
+
 int WriteResponseHead(storage::mojom::ServiceWorkerResourceWriter* writer,
                       network::mojom::URLResponseHeadPtr response_head) {
   int return_value;
@@ -446,6 +463,7 @@ TEST_F(ServiceWorkerStorageControlImplTest, WriteAndReadResource) {
   // Write content.
   {
     mojo_base::BigBuffer data(base::as_bytes(base::make_span(kData)));
+    int data_size = data.size();
 
     int result = WriteResponseData(writer.get(), std::move(data));
     ASSERT_EQ(data_size, result);
@@ -454,7 +472,7 @@ TEST_F(ServiceWorkerStorageControlImplTest, WriteAndReadResource) {
   mojo::Remote<storage::mojom::ServiceWorkerResourceReader> reader =
       CreateResourceReader(resource_id);
 
-  // Read the response head.
+  // Read the response head and the content.
   {
     network::mojom::URLResponseHeadPtr response_head;
     int result = ReadResponseHead(reader.get(), response_head);
@@ -465,9 +483,12 @@ TEST_F(ServiceWorkerStorageControlImplTest, WriteAndReadResource) {
     EXPECT_TRUE(response_head->ssl_info->is_valid());
     EXPECT_EQ(response_head->ssl_info->cert->serial_number(),
               ssl_info.cert->serial_number());
+
+    std::string data = ReadResponseData(reader.get(), data_size);
+    EXPECT_EQ(data, kData);
   }
 
-  // TODO(crbug.com/1055677): Read and check the content of the resource.
+  // TODO(crbug.com/1055677): Write metadata, read it then check the metadata.
 }
 
 }  // namespace content

content/browser/service_worker/service_worker_test_utils.cc

@@ -872,4 +872,50 @@ bool ServiceWorkerUpdateCheckTestUtils::VerifyStoredResponse(
   return true;
 }
 
+void ReadDataPipeInternal(mojo::DataPipeConsumerHandle handle,
+                          std::string* result,
+                          base::OnceClosure quit_closure) {
+  while (true) {
+    uint32_t num_bytes;
+    const void* buffer = nullptr;
+    MojoResult rv =
+        handle.BeginReadData(&buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+    switch (rv) {
+      case MOJO_RESULT_BUSY:
+      case MOJO_RESULT_INVALID_ARGUMENT:
+        NOTREACHED();
+        return;
+      case MOJO_RESULT_FAILED_PRECONDITION:
+        std::move(quit_closure).Run();
+        return;
+      case MOJO_RESULT_SHOULD_WAIT:
+        base::ThreadTaskRunnerHandle::Get()->PostTask(
+            FROM_HERE, base::BindOnce(&ReadDataPipeInternal, handle, result,
+                                      std::move(quit_closure)));
+        return;
+      case MOJO_RESULT_OK:
+        EXPECT_NE(nullptr, buffer);
+        EXPECT_GT(num_bytes, 0u);
+        uint32_t before_size = result->size();
+        result->append(static_cast<const char*>(buffer), num_bytes);
+        uint32_t read_size = result->size() - before_size;
+        EXPECT_EQ(num_bytes, read_size);
+        rv = handle.EndReadData(read_size);
+        EXPECT_EQ(MOJO_RESULT_OK, rv);
+        break;
+    }
+  }
+  NOTREACHED();
+  return;
+}
+
+std::string ReadDataPipe(mojo::ScopedDataPipeConsumerHandle handle) {
+  EXPECT_TRUE(handle.is_valid());
+  std::string result;
+  base::RunLoop loop;
+  ReadDataPipeInternal(handle.get(), &result, loop.QuitClosure());
+  loop.Run();
+  return result;
+}
+
 }  // namespace content

content/browser/service_worker/service_worker_test_utils.h

@@ -417,6 +417,12 @@ class ServiceWorkerUpdateCheckTestUtils {
                                    const std::string& expected_body);
 };
 
+// Reads all data from the given |handle| and returns data as a string.
+// This is similar to mojo::BlockingCopyToString() but a bit different. This
+// doesn't wait synchronously but keep posting a task when |handle| returns
+// MOJO_RESULT_SHOULD_WAIT.
+std::string ReadDataPipe(mojo::ScopedDataPipeConsumerHandle handle);
+
 }  // namespace content
 
 #endif  // CONTENT_BROWSER_SERVICE_WORKER_SERVICE_WORKER_TEST_UTILS_H_