SignedExchangeHandler uses SourceStream

Bug: 803774
Change-Id: I358d7311bf3aad9a9f96e56cadbd6b3b0dd31485
Reviewed-on: https://chromium-review.googlesource.com/890156
Commit-Queue: Kinuko Yasuda <kinuko@chromium.org>
Reviewed-by: Tsuyoshi Horo <horo@chromium.org>
Cr-Commit-Position: refs/heads/master@{#533613}

content/browser/BUILD.gn

@@ -966,6 +966,8 @@ jumbo_source_set("browser") {
     "leveldb_wrapper_impl.h",
     "loader/cross_site_document_resource_handler.cc",
     "loader/cross_site_document_resource_handler.h",
+    "loader/data_pipe_to_source_stream.cc",
+    "loader/data_pipe_to_source_stream.h",
     "loader/detachable_resource_handler.cc",
     "loader/detachable_resource_handler.h",
     "loader/downloaded_temp_file_impl.cc",
@@ -1021,6 +1023,8 @@ jumbo_source_set("browser") {
     "loader/resource_scheduler_filter.h",
     "loader/signed_exchange_handler.cc",
     "loader/signed_exchange_handler.h",
+    "loader/source_stream_to_data_pipe.cc",
+    "loader/source_stream_to_data_pipe.h",
     "loader/stream_resource_handler.cc",
     "loader/stream_resource_handler.h",
     "loader/stream_writer.cc",

content/browser/loader/data_pipe_to_source_stream.cc

+// Copyright 2018 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/loader/data_pipe_to_source_stream.h"
+
+#include "base/auto_reset.h"
+#include "net/base/io_buffer.h"
+
+namespace content {
+
+DataPipeToSourceStream::DataPipeToSourceStream(
+    mojo::ScopedDataPipeConsumerHandle body)
+    : net::SourceStream(net::SourceStream::TYPE_NONE),
+      body_(std::move(body)),
+      handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL) {
+  handle_watcher_.Watch(
+      body_.get(), MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      base::BindRepeating(&DataPipeToSourceStream::OnReadable,
+                          base::Unretained(this)));
+}
+
+DataPipeToSourceStream::~DataPipeToSourceStream() = default;
+
+std::string DataPipeToSourceStream::Description() const {
+  return "DataPipe";
+}
+
+int DataPipeToSourceStream::Read(net::IOBuffer* buf,
+                                 int buf_size,
+                                 const net::CompletionCallback& callback) {
+  base::AutoReset<bool>(&inside_read_, true);
+
+  if (!body_.get()) {
+    // We have finished reading the pipe.
+    return 0;
+  }
+
+  const void* buffer = nullptr;
+  uint32_t available = 0;
+  MojoResult result =
+      body_->BeginReadData(&buffer, &available, MOJO_READ_DATA_FLAG_NONE);
+  switch (result) {
+    case MOJO_RESULT_OK: {
+      uint32_t consume =
+          std::min(base::checked_cast<uint32_t>(buf_size), available);
+      memcpy(buf->data(), buffer, consume);
+      body_->EndReadData(consume);
+      return base::checked_cast<int>(consume);
+    }
+    case MOJO_RESULT_FAILED_PRECONDITION:
+      // Finished reading.
+      FinishReading();
+      return 0;
+    case MOJO_RESULT_SHOULD_WAIT:
+      // Data is not available yet.
+      pending_callback_ = callback;
+      output_buf_ = buf;
+      output_buf_size_ = buf_size;
+      handle_watcher_.ArmOrNotify();
+      return net::ERR_IO_PENDING;
+  }
+  NOTREACHED() << static_cast<int>(result);
+  return net::ERR_UNEXPECTED;
+}
+
+void DataPipeToSourceStream::OnReadable(MojoResult unused) {
+  // It's not expected that we call this synchronously inside Read.
+  DCHECK(!inside_read_);
+  DCHECK(pending_callback_);
+  DCHECK(output_buf_);
+  const void* buffer = nullptr;
+  uint32_t available = 0;
+  MojoResult result =
+      body_->BeginReadData(&buffer, &available, MOJO_READ_DATA_FLAG_NONE);
+  switch (result) {
+    case MOJO_RESULT_OK: {
+      uint32_t consume =
+          std::min(base::checked_cast<uint32_t>(output_buf_size_), available);
+      memcpy(output_buf_->data(), buffer, consume);
+      body_->EndReadData(consume);
+      std::move(pending_callback_).Run(consume);
+      return;
+    }
+    case MOJO_RESULT_FAILED_PRECONDITION:
+      FinishReading();
+      std::move(pending_callback_).Run(0);
+      return;
+    case MOJO_RESULT_SHOULD_WAIT:
+      handle_watcher_.ArmOrNotify();
+      return;
+  }
+  NOTREACHED() << static_cast<int>(result);
+}
+
+void DataPipeToSourceStream::FinishReading() {
+  handle_watcher_.Cancel();
+  body_.reset();
+}
+
+}  // namespace content

content/browser/loader/data_pipe_to_source_stream.h

+// Copyright 2018 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.
+
+#ifndef CONTENT_BROWSER_LOADER_DATA_PIPE_TO_SOURCE_STREAM_H_
+#define CONTENT_BROWSER_LOADER_DATA_PIPE_TO_SOURCE_STREAM_H_
+
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/filter/source_stream.h"
+
+namespace content {
+
+class DataPipeToSourceStream final : public net::SourceStream {
+ public:
+  explicit DataPipeToSourceStream(mojo::ScopedDataPipeConsumerHandle body);
+  ~DataPipeToSourceStream() override;
+
+  int Read(net::IOBuffer* buf,
+           int buf_size,
+           const net::CompletionCallback& callback) override;
+  std::string Description() const override;
+
+ private:
+  void OnReadable(MojoResult result);
+  void FinishReading();
+
+  mojo::ScopedDataPipeConsumerHandle body_;
+  mojo::SimpleWatcher handle_watcher_;
+
+  bool inside_read_ = false;
+
+  scoped_refptr<net::IOBuffer> output_buf_;
+  int output_buf_size_ = 0;
+  net::CompletionCallback pending_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeToSourceStream);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_LOADER_DATA_PIPE_TO_SOURCE_STREAM_H_

content/browser/loader/signed_exchange_handler.cc

@@ -7,87 +7,124 @@
 #include "base/feature_list.h"
 #include "content/public/common/content_features.h"
 #include "mojo/public/cpp/system/string_data_pipe_producer.h"
+#include "net/base/io_buffer.h"
 #include "net/http/http_response_headers.h"
 #include "services/network/public/cpp/resource_response.h"
 #include "services/network/public/cpp/url_loader_completion_status.h"
 
 namespace content {
-namespace {
-
-constexpr size_t kPipeSizeForSignedResponseBody = 65536;
-
-}  // namespace
 
 SignedExchangeHandler::SignedExchangeHandler(
-    mojo::ScopedDataPipeConsumerHandle body)
-    : body_(std::move(body)) {
+    std::unique_ptr<net::SourceStream> upstream,
+    ExchangeHeadersCallback headers_callback)
+    : net::FilterSourceStream(net::SourceStream::TYPE_NONE,
+                              std::move(upstream)),
+      headers_callback_(std::move(headers_callback)),
+      weak_factory_(this) {
   DCHECK(base::FeatureList::IsEnabled(features::kSignedHTTPExchange));
+
+  // Triggering the first read (asynchronously) for header parsing.
+  header_out_buf_ = base::MakeRefCounted<net::IOBufferWithSize>(1);
+  base::SequencedTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE, base::BindOnce(&SignedExchangeHandler::DoHeaderLoop,
+                                weak_factory_.GetWeakPtr()));
 }
 
 SignedExchangeHandler::~SignedExchangeHandler() = default;
 
-void SignedExchangeHandler::GetHTTPExchange(
-    ExchangeFoundCallback found_callback,
-    ExchangeFinishedCallback finish_callback) {
-  DCHECK(!found_callback_);
-  DCHECK(!finish_callback_);
-  found_callback_ = std::move(found_callback);
-  finish_callback_ = std::move(finish_callback);
+int SignedExchangeHandler::FilterData(net::IOBuffer* output_buffer,
+                                      int output_buffer_size,
+                                      net::IOBuffer* input_buffer,
+                                      int input_buffer_size,
+                                      int* consumed_bytes,
+                                      bool upstream_eof_reached) {
+  *consumed_bytes = 0;
+
+  original_body_string_.append(input_buffer->data(), input_buffer_size);
+  *consumed_bytes += input_buffer_size;
+
+  // We shouldn't write any data into the out buffer while we're
+  // parsing the header.
+  if (headers_callback_)
+    return 0;
+
+  if (upstream_eof_reached) {
+    // Run the parser code if this part is run for the first time.
+    // Now original_body_string_ has the entire body.
+    // (Note that we may come here multiple times if output_buffer_size
+    // is not big enough.
+    // TODO(https://crbug.com/803774): Do the streaming instead.
+    size_t size_to_copy =
+        std::min(original_body_string_.size() - body_string_offset_,
+                 base::checked_cast<size_t>(output_buffer_size));
+    memcpy(output_buffer->data(),
+           original_body_string_.data() + body_string_offset_, size_to_copy);
+    body_string_offset_ += size_to_copy;
+    return base::checked_cast<int>(size_to_copy);
+  }
+
+  return 0;
+}
 
-  drainer_.reset(new mojo::common::DataPipeDrainer(this, std::move(body_)));
+std::string SignedExchangeHandler::GetTypeAsString() const {
+  return "HTXG";  // Tentative.
 }
 
-void SignedExchangeHandler::OnDataAvailable(const void* data,
-                                            size_t num_bytes) {
-  original_body_string_.append(static_cast<const char*>(data), num_bytes);
+void SignedExchangeHandler::DoHeaderLoop() {
+  // Run the internal read loop by ourselves until we finish
+  // parsing the headers. (After that the caller should pumb
+  // the Read() calls).
+  DCHECK(headers_callback_);
+  DCHECK(header_out_buf_);
+  int rv = Read(header_out_buf_.get(), header_out_buf_->size(),
+                base::BindRepeating(&SignedExchangeHandler::DidReadForHeaders,
+                                    base::Unretained(this), false /* sync */));
+  if (rv != net::ERR_IO_PENDING)
+    DidReadForHeaders(true /* sync */, rv);
 }
 
-void SignedExchangeHandler::OnDataComplete() {
-  if (!found_callback_)
+void SignedExchangeHandler::DidReadForHeaders(bool completed_syncly,
+                                              int result) {
+  if (MaybeRunHeadersCallback() || result < 0)
     return;
+  DCHECK_EQ(0, result);
+  if (completed_syncly) {
+    base::SequencedTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::BindOnce(&SignedExchangeHandler::DoHeaderLoop,
+                                  weak_factory_.GetWeakPtr()));
+  } else {
+    DoHeaderLoop();
+  }
+}
+
+bool SignedExchangeHandler::MaybeRunHeadersCallback() {
+  if (!headers_callback_)
+    return false;
+
+  // If this was the first read, fire the headers callback now.
+  // TODO(https://crbug.com/803774): This is just for testing, we should
+  // implement the CBOR parsing here.
+  FillMockExchangeHeaders();
+  std::move(headers_callback_)
+      .Run(request_url_, request_method_, response_head_, ssl_info_);
 
+  // TODO(https://crbug.com/803774) Consume the bytes size that were
+  // necessary to read out the headers.
+
+  return true;
+}
+
+void SignedExchangeHandler::FillMockExchangeHeaders() {
   // TODO(https://crbug.com/803774): Get the request url by parsing CBOR format.
-  GURL request_url = GURL("https://example.com/test.html");
+  request_url_ = GURL("https://example.com/test.html");
   // TODO(https://crbug.com/803774): Get the request method by parsing CBOR
   // format.
-  std::string request_method = "GET";
-  // TODO(https://crbug.com/803774): Get the payload by parsing CBOR format.
-  std::string payload = original_body_string_;
-  original_body_string_.clear();
-
+  request_method_ = "GET";
   // TODO(https://crbug.com/803774): Get more headers by parsing CBOR.
   scoped_refptr<net::HttpResponseHeaders> headers(
       new net::HttpResponseHeaders("HTTP/1.1 200 OK"));
-  network::ResourceResponseHead resource_response;
-  resource_response.headers = headers;
-  resource_response.mime_type = "text/html";
-
-  // TODO(https://crbug.com/803774): Get the certificate by parsing CBOR and
-  // verify.
-  base::Optional<net::SSLInfo> ssl_info;
-
-  mojo::DataPipe pipe(kPipeSizeForSignedResponseBody);
-  // TODO(https://crbug.com/803774): Write the error handling code. This could
-  // fail.
-  DCHECK(pipe.consumer_handle.is_valid());
-  mojo::ScopedDataPipeConsumerHandle response_body =
-      std::move(pipe.consumer_handle);
-  std::move(found_callback_)
-      .Run(request_url, request_method, resource_response, std::move(ssl_info),
-           std::move(response_body));
-
-  data_producer_ = std::make_unique<mojo::StringDataPipeProducer>(
-      std::move(pipe.producer_handle));
-  data_producer_->Write(payload,
-                        base::BindOnce(&SignedExchangeHandler::OnDataWritten,
-                                       base::Unretained(this)));
-}
-
-void SignedExchangeHandler::OnDataWritten(MojoResult result) {
-  data_producer_.reset();
-  std::move(finish_callback_)
-      .Run(network::URLLoaderCompletionStatus(
-          result == MOJO_RESULT_OK ? net::OK : net::ERR_FAILED));
+  response_head_.headers = headers;
+  response_head_.mime_type = "text/html";
 }
 
 }  // namespace content

content/browser/loader/signed_exchange_handler.h

@@ -9,20 +9,13 @@
 
 #include "base/callback.h"
 #include "base/optional.h"
-#include "mojo/common/data_pipe_drainer.h"
 #include "mojo/public/cpp/system/data_pipe.h"
+#include "net/base/completion_callback.h"
+#include "net/filter/filter_source_stream.h"
 #include "net/ssl/ssl_info.h"
+#include "services/network/public/cpp/resource_response.h"
 #include "url/gurl.h"
 
-namespace mojo {
-class StringDataPipeProducer;
-}  // namespace mojo
-
-namespace network {
-struct ResourceResponseHead;
-struct URLLoaderCompletionStatus;
-}  // namespace network
-
 namespace content {
 
 // IMPORTANT: Currenly SignedExchangeHandler doesn't implement any CBOR parsing
@@ -31,47 +24,58 @@ namespace content {
 // a response with a payload which is equal to the original body.
 // TODO(https://crbug.com/803774): Implement CBOR parsing logic and verifying
 // logic.
-class SignedExchangeHandler final
-    : public mojo::common::DataPipeDrainer::Client {
+class SignedExchangeHandler final : public net::FilterSourceStream {
  public:
-  using ExchangeFoundCallback =
+  // TODO(https://crbug.com/803774): Add verification status here.
+  using ExchangeHeadersCallback =
       base::OnceCallback<void(const GURL& request_url,
                               const std::string& request_method,
                               const network::ResourceResponseHead&,
-                              base::Optional<net::SSLInfo>,
-                              mojo::ScopedDataPipeConsumerHandle)>;
-  using ExchangeFinishedCallback =
-      base::OnceCallback<void(const network::URLLoaderCompletionStatus&)>;
-
-  // The passed |body| will be read to parse the signed HTTP exchange.
-  // TODO(https://crbug.com/803774): Consider making SignedExchangeHandler
-  // independent from DataPipe to make it easy to port it in //net.
-  explicit SignedExchangeHandler(mojo::ScopedDataPipeConsumerHandle body);
+                              base::Optional<net::SSLInfo>)>;
 
+  // Once constructed |this| starts reading the |body| and parses the response
+  // as a signed HTTP exchange. The response body of the exchange can be read
+  // from |this| as a net::SourceStream after |headers_callback| is called.
+  SignedExchangeHandler(std::unique_ptr<net::SourceStream> body,
+                        ExchangeHeadersCallback headers_callback);
   ~SignedExchangeHandler() override;
 
-  // TODO(https://crbug.com/803774): Currently SignedExchangeHandler always
-  // calls found_callback and then calls finish_callback after reading the all
-  // buffer. Need to redesign this callback model when we will introduce
-  // SignedExchangeHandler::Reader class to read the body and introduce the
-  // cert verification.
-  void GetHTTPExchange(ExchangeFoundCallback found_callback,
-                       ExchangeFinishedCallback finish_callback);
+  // net::FilterSourceStream:
+  int FilterData(net::IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 net::IOBuffer* input_buffer,
+                 int input_buffer_size,
+                 int* consumed_bytes,
+                 bool upstream_eof_reached) override;
+  std::string GetTypeAsString() const override;
 
  private:
-  // mojo::Common::DataPipeDrainer::Client
-  void OnDataAvailable(const void* data, size_t num_bytes) override;
-  void OnDataComplete() override;
+  void DoHeaderLoop();
+  void DidReadForHeaders(bool completed_syncly, int result);
+  bool MaybeRunHeadersCallback();
+
+  // TODO(https://crbug.com/803774): Remove this.
+  void FillMockExchangeHeaders();
 
-  // Called from |data_producer_|.
-  void OnDataWritten(MojoResult result);
+  // Signed exchange contents.
+  GURL request_url_;
+  std::string request_method_;
+  network::ResourceResponseHead response_head_;
+  base::Optional<net::SSLInfo> ssl_info_;
 
-  mojo::ScopedDataPipeConsumerHandle body_;
-  std::unique_ptr<mojo::common::DataPipeDrainer> drainer_;
-  ExchangeFoundCallback found_callback_;
-  ExchangeFinishedCallback finish_callback_;
+  ExchangeHeadersCallback headers_callback_;
+
+  // Internal IOBuffer used during reading the header. Note that during parsing
+  // the header we don't really need the output buffer, but we still need to
+  // give some > 0 buffer.
+  scoped_refptr<net::IOBufferWithSize> header_out_buf_;
+
+  // TODO(https://crbug.cxom/803774): Just for now. Implement the streaming
+  // parser.
   std::string original_body_string_;
-  std::unique_ptr<mojo::StringDataPipeProducer> data_producer_;
+  size_t body_string_offset_ = 0;
+
+  base::WeakPtrFactory<SignedExchangeHandler> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(SignedExchangeHandler);
 };

content/browser/loader/source_stream_to_data_pipe.cc

+// Copyright 2018 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/loader/source_stream_to_data_pipe.h"
+
+#include "base/bind.h"
+#include "net/filter/source_stream.h"
+
+namespace content {
+
+SourceStreamToDataPipe::SourceStreamToDataPipe(
+    net::SourceStream* source,
+    mojo::ScopedDataPipeProducerHandle dest,
+    base::OnceCallback<void(int)> completion_callback)
+    : source_(source),
+      dest_(std::move(dest)),
+      completion_callback_(std::move(completion_callback)),
+      writable_handle_watcher_(FROM_HERE,
+                               mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+      peer_closed_handle_watcher_(FROM_HERE,
+                                  mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+      weak_factory_(this) {
+  peer_closed_handle_watcher_.Watch(
+      dest_.get(), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      base::BindRepeating(&SourceStreamToDataPipe::OnDataPipeClosed,
+                          base::Unretained(this)));
+  peer_closed_handle_watcher_.ArmOrNotify();
+  writable_handle_watcher_.Watch(
+      dest_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+      base::BindRepeating(&SourceStreamToDataPipe::OnDataPipeWritable,
+                          base::Unretained(this)));
+}
+
+SourceStreamToDataPipe::~SourceStreamToDataPipe() = default;
+
+void SourceStreamToDataPipe::Start() {
+  ReadMore();
+}
+
+void SourceStreamToDataPipe::ReadMore() {
+  DCHECK(!pending_write_.get());
+
+  uint32_t num_bytes;
+  MojoResult mojo_result = network::NetToMojoPendingBuffer::BeginWrite(
+      &dest_, &pending_write_, &num_bytes);
+  if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+    // The pipe is full.  We need to wait for it to have more space.
+    writable_handle_watcher_.ArmOrNotify();
+    return;
+  } else if (mojo_result != MOJO_RESULT_OK) {
+    // The body stream is in a bad state. Bail out.
+    OnComplete(net::ERR_UNEXPECTED);
+    return;
+  }
+
+  scoped_refptr<net::IOBuffer> buffer(
+      new network::NetToMojoIOBuffer(pending_write_.get()));
+  int result =
+      source_->Read(buffer.get(), base::checked_cast<int>(num_bytes),
+                    base::BindRepeating(&SourceStreamToDataPipe::DidRead,
+                                        base::Unretained(this)));
+
+  if (result != net::ERR_IO_PENDING)
+    DidRead(result);
+}
+
+void SourceStreamToDataPipe::DidRead(int result) {
+  DCHECK(pending_write_);
+  if (result < 0) {
+    // An error case.
+    OnComplete(result);
+    return;
+  }
+  if (result == 0) {
+    pending_write_->Complete(0);
+    OnComplete(net::OK);
+    return;
+  }
+  dest_ = pending_write_->Complete(result);
+  pending_write_ = nullptr;
+
+  base::SequencedTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE, base::BindOnce(&SourceStreamToDataPipe::ReadMore,
+                                weak_factory_.GetWeakPtr()));
+}
+
+void SourceStreamToDataPipe::OnDataPipeWritable(MojoResult result) {
+  if (result == MOJO_RESULT_FAILED_PRECONDITION) {
+    OnComplete(net::ERR_ABORTED);
+    return;
+  }
+  DCHECK_EQ(result, MOJO_RESULT_OK) << result;
+
+  ReadMore();
+}
+
+void SourceStreamToDataPipe::OnDataPipeClosed(MojoResult result) {
+  OnComplete(net::ERR_ABORTED);
+}
+
+void SourceStreamToDataPipe::OnComplete(int result) {
+  // Resets the watchers, pipes and the exchange handler, so that
+  // we will never be called back.
+  writable_handle_watcher_.Cancel();
+  peer_closed_handle_watcher_.Cancel();
+  pending_write_ = nullptr;  // Closes the data pipe if this was holding it.
+  dest_.reset();
+
+  std::move(completion_callback_).Run(result);
+}
+
+}  // namespace content

content/browser/loader/source_stream_to_data_pipe.h

+// Copyright 2018 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.
+
+#ifndef CONTENT_BROWSER_LOADER_SOURCE_STREAM_TO_DATA_PIPE_H_
+#define CONTENT_BROWSER_LOADER_SOURCE_STREAM_TO_DATA_PIPE_H_
+
+#include "base/callback_forward.h"
+#include "base/memory/weak_ptr.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "services/network/public/cpp/net_adapters.h"
+
+namespace net {
+class SourceStream;
+}
+
+namespace content {
+
+// A convenient adapter class to read out data from net::SourceStream
+// and write them into a data pipe.
+class SourceStreamToDataPipe {
+ public:
+  // Reads out the data from |source| and write into |dest|.
+  // Note that this does not take the ownership of |source|, the caller
+  // needs to take care that it is kept alive.
+  SourceStreamToDataPipe(net::SourceStream* source,
+                         mojo::ScopedDataPipeProducerHandle dest,
+                         base::OnceCallback<void(int)> completion_callback);
+  ~SourceStreamToDataPipe();
+
+  // Start reading the source.
+  void Start();
+
+ private:
+  void ReadMore();
+  void DidRead(int result);
+
+  void OnDataPipeWritable(MojoResult result);
+  void OnDataPipeClosed(MojoResult result);
+  void OnComplete(int result);
+
+  net::SourceStream* source_;  // Not owned.
+  mojo::ScopedDataPipeProducerHandle dest_;
+  base::OnceCallback<void(int)> completion_callback_;
+
+  scoped_refptr<network::NetToMojoPendingBuffer> pending_write_;
+  mojo::SimpleWatcher writable_handle_watcher_;
+  mojo::SimpleWatcher peer_closed_handle_watcher_;
+
+  base::WeakPtrFactory<SourceStreamToDataPipe> weak_factory_;
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_LOADER_SOURCE_STREAM_TO_DATA_PIPE_H_

content/browser/loader/web_package_loader.cc

@@ -6,10 +6,13 @@
 
 #include "base/feature_list.h"
 #include "base/strings/stringprintf.h"
+#include "content/browser/loader/data_pipe_to_source_stream.h"
 #include "content/browser/loader/signed_exchange_handler.h"
+#include "content/browser/loader/source_stream_to_data_pipe.h"
 #include "content/public/common/content_features.h"
 #include "net/http/http_util.h"
 #include "services/network/public/cpp/features.h"
+#include "services/network/public/cpp/url_loader_completion_status.h"
 
 namespace content {
 
@@ -24,6 +27,8 @@ net::RedirectInfo CreateRedirectInfo(const GURL& new_url) {
   return redirect_info;
 }
 
+constexpr static int kDefaultBufferSize = 64 * 1024;
+
 }  // namespace
 
 class WebPackageLoader::ResponseTimingInfo {
@@ -135,12 +140,9 @@ void WebPackageLoader::OnTransferSizeUpdated(int32_t transfer_size_diff) {
 
 void WebPackageLoader::OnStartLoadingResponseBody(
     mojo::ScopedDataPipeConsumerHandle body) {
-  signed_exchange_handler_ =
-      base::MakeUnique<SignedExchangeHandler>(std::move(body));
-  signed_exchange_handler_->GetHTTPExchange(
+  signed_exchange_handler_ = std::make_unique<SignedExchangeHandler>(
+      base::MakeUnique<DataPipeToSourceStream>(std::move(body)),
       base::BindOnce(&WebPackageLoader::OnHTTPExchangeFound,
-                     weak_factory_.GetWeakPtr()),
-      base::BindOnce(&WebPackageLoader::OnHTTPExchangeFinished,
                      weak_factory_.GetWeakPtr()));
 }
 
@@ -158,12 +160,11 @@ void WebPackageLoader::ProceedWithResponse() {
   // TODO(https://crbug.com/791049): Remove this when NetworkService is
   // enabled by default.
   DCHECK(!base::FeatureList::IsEnabled(network::features::kNetworkService));
-  DCHECK(pending_body_.is_valid());
-  DCHECK(client_);
-  client_->OnStartLoadingResponseBody(std::move(pending_body_));
-  if (pending_completion_status_) {
-    client_->OnComplete(*pending_completion_status_);
-  }
+  DCHECK(body_data_pipe_adapter_);
+  DCHECK(pending_body_consumer_.is_valid());
+
+  body_data_pipe_adapter_->Start();
+  client_->OnStartLoadingResponseBody(std::move(pending_body_consumer_));
 }
 
 void WebPackageLoader::SetPriority(net::RequestPriority priority,
@@ -190,8 +191,7 @@ void WebPackageLoader::OnHTTPExchangeFound(
     const GURL& request_url,
     const std::string& request_method,
     const network::ResourceResponseHead& resource_response,
-    base::Optional<net::SSLInfo> ssl_info,
-    mojo::ScopedDataPipeConsumerHandle body) {
+    base::Optional<net::SSLInfo> ssl_info) {
   // TODO(https://crbug.com/803774): Handle no-GET request_method as a error.
   DCHECK(original_response_timing_info_);
   forwarding_client_->OnReceiveRedirect(
@@ -203,24 +203,31 @@ void WebPackageLoader::OnHTTPExchangeFound(
   client_->OnReceiveResponse(resource_response, std::move(ssl_info),
                              nullptr /* downloaded_file */);
 
+  // Currently we always assume that we have body.
+  // TODO(https://crbug.com/80374): Add error handling and bail out
+  // earlier if there's an error.
+
+  mojo::DataPipe data_pipe(kDefaultBufferSize);
+  pending_body_consumer_ = std::move(data_pipe.consumer_handle);
+
+  body_data_pipe_adapter_ = std::make_unique<SourceStreamToDataPipe>(
+      signed_exchange_handler_.get(), std::move(data_pipe.producer_handle),
+      base::BindOnce(&WebPackageLoader::FinishReadingBody,
+                     base::Unretained(this)));
+
   if (!base::FeatureList::IsEnabled(network::features::kNetworkService)) {
     // Need to wait until ProceedWithResponse() is called.
-    pending_body_ = std::move(body);
-  } else {
-    client_->OnStartLoadingResponseBody(std::move(body));
+    return;
   }
+
+  // Start reading.
+  body_data_pipe_adapter_->Start();
+  client_->OnStartLoadingResponseBody(std::move(pending_body_consumer_));
 }
 
-void WebPackageLoader::OnHTTPExchangeFinished(
-    const network::URLLoaderCompletionStatus& status) {
-  if (pending_body_.is_valid()) {
-    DCHECK(!base::FeatureList::IsEnabled(network::features::kNetworkService));
-    // If ProceedWithResponse() was not called yet, need to call OnComplete()
-    // after ProceedWithResponse() is called.
-    pending_completion_status_ = status;
-  } else {
-    client_->OnComplete(status);
-  }
+void WebPackageLoader::FinishReadingBody(int result) {
+  // This will eventually delete |this|.
+  client_->OnComplete(network::URLLoaderCompletionStatus(result));
 }
 
 }  // namespace content

content/browser/loader/web_package_loader.h

@@ -7,13 +7,16 @@
 
 #include "base/optional.h"
 #include "mojo/public/cpp/bindings/binding.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
 #include "net/ssl/ssl_info.h"
 #include "net/url_request/redirect_info.h"
+#include "services/network/public/cpp/net_adapters.h"
 #include "services/network/public/interfaces/url_loader.mojom.h"
 
 namespace content {
 
 class SignedExchangeHandler;
+class SourceStreamToDataPipe;
 
 // WebPackageLoader handles an origin-signed HTTP exchange response. It is
 // created when a WebPackageRequestHandler recieves an origin-signed HTTP
@@ -66,12 +69,9 @@ class WebPackageLoader final : public network::mojom::URLLoaderClient,
       const GURL& request_url,
       const std::string& request_method,
       const network::ResourceResponseHead& resource_response,
-      base::Optional<net::SSLInfo> ssl_info,
-      mojo::ScopedDataPipeConsumerHandle body);
+      base::Optional<net::SSLInfo> ssl_info);
 
-  // Called from |signed_exchange_handler_| when it finished sending the
-  // payload of the origin-signed HTTP response.
-  void OnHTTPExchangeFinished(const network::URLLoaderCompletionStatus& status);
+  void FinishReadingBody(int result);
 
   // This timing info is used to create a dummy redirect response.
   std::unique_ptr<const ResponseTimingInfo> original_response_timing_info_;
@@ -92,11 +92,10 @@ class WebPackageLoader final : public network::mojom::URLLoaderClient,
   network::mojom::URLLoaderClientRequest pending_client_request_;
 
   std::unique_ptr<SignedExchangeHandler> signed_exchange_handler_;
+  std::unique_ptr<SourceStreamToDataPipe> body_data_pipe_adapter_;
 
-  // This is used to keep the DataPipe until ProceedWithResponse() is called.
-  mojo::ScopedDataPipeConsumerHandle pending_body_;
-  // This is used to keep the status until ProceedWithResponse() is called.
-  base::Optional<network::URLLoaderCompletionStatus> pending_completion_status_;
+  // Kept around until ProceedWithResponse is called.
+  mojo::ScopedDataPipeConsumerHandle pending_body_consumer_;
 
   base::WeakPtrFactory<WebPackageLoader> weak_factory_;