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

third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp

 // 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 "modules/fetch/FetchDataLoader.h"
 
+#include <memory>
 #include "core/html/parser/TextResourceDecoder.h"
 #include "modules/fetch/BytesConsumer.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "platform/instrumentation/tracing/TraceEvent.h"
+#include "wtf/Functional.h"
 #include "wtf/PtrUtil.h"
 #include "wtf/text/StringBuilder.h"
 #include "wtf/text/WTFString.h"
 #include "wtf/typed_arrays/ArrayBufferBuilder.h"
-#include <memory>
 
 namespace blink {
 
 namespace {
 
 class FetchDataLoaderAsBlobHandle final : public FetchDataLoader,
                                           public BytesConsumer::Client {
   USING_GARBAGE_COLLECTED_MIXIN(FetchDataLoaderAsBlobHandle);
 
  public:
@@ skipping 200 matching lines @@
   }
 
  private:
   Member<BytesConsumer> consumer_;
   Member<FetchDataLoader::Client> client_;
 
   std::unique_ptr<TextResourceDecoder> decoder_;
   StringBuilder builder_;
 };
 
-class FetchDataLoaderAsStream final : public FetchDataLoader,
-                                      public BytesConsumer::Client {
-  USING_GARBAGE_COLLECTED_MIXIN(FetchDataLoaderAsStream);
+class FetchDataLoaderAsDataPipe final : public FetchDataLoader,
+                                        public BytesConsumer::Client {
+  USING_GARBAGE_COLLECTED_MIXIN(FetchDataLoaderAsDataPipe);
 
  public:
-  explicit FetchDataLoaderAsStream(Stream* out_stream)
-      : out_stream_(out_stream) {}
-
+  explicit FetchDataLoaderAsDataPipe(
+      mojo::ScopedDataPipeProducerHandle out_datapipe)
+      : out_datapipe_(std::move(out_datapipe)),
+        datapipe_watcher_(FROM_HERE,
+                          mojo::SimpleWatcher::ArmingPolicy::AUTOMATIC) {
+    TRACE_EVENT0("ServiceWorker",
+                 "FetchDataLoaderAsDataPipe::FetchDataLoaderAsDataPipe");
+  }
+  ~FetchDataLoaderAsDataPipe() override {}
   void Start(BytesConsumer* consumer,
              FetchDataLoader::Client* client) override {
+    TRACE_EVENT_NESTABLE_ASYNC_BEGIN0("ServiceWorker",
+                                      "FetchDataLoaderAsDataPipe", this);
     DCHECK(!client_);
     DCHECK(!consumer_);
+    datapipe_watcher_.Watch(
+        out_datapipe_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+        ConvertToBaseCallback(WTF::Bind(&FetchDataLoaderAsDataPipe::OnWritable,
+                                        WrapWeakPersistent(this))));
     client_ = client;
     consumer_ = consumer;
     consumer_->SetClient(this);
+  }
+
+  void OnWritable(MojoResult result) {
+    if (result != MOJO_RESULT_OK)
+      Cancel();
     OnStateChange();
   }
 
+  // Implements BytesConsumer::Client.
   void OnStateChange() override {
-    bool need_to_flush = false;
-    while (true) {
+    bool should_wait = false;
+    while (!should_wait) {
       const char* buffer;
       size_t available;
       auto result = consumer_->BeginRead(&buffer, &available);
-      if (result == BytesConsumer::Result::kShouldWait) {
-        if (need_to_flush)
-          out_stream_->Flush();
+      if (result == BytesConsumer::Result::kShouldWait)
         return;
-      }
       if (result == BytesConsumer::Result::kOk) {
-        out_stream_->AddData(buffer, available);
-        need_to_flush = true;
-        result = consumer_->EndRead(available);
+        DCHECK_GT(available, 0UL);
+        uint32_t num_bytes = available;
+        MojoResult mojo_result = mojo::WriteDataRaw(
+            out_datapipe_.get(), buffer, &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+        if (mojo_result == MOJO_RESULT_OK) {
+          TRACE_EVENT_NESTABLE_ASYNC_INSTANT2(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "MOJO_RESULT_OK", "bytes", num_bytes);
+          result = consumer_->EndRead(num_bytes);
+        } else if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+          TRACE_EVENT_NESTABLE_ASYNC_INSTANT1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "MOJO_RESULT_SHOULD_WAIT");
+          result = consumer_->EndRead(0);
+          should_wait = true;
+        } else {
+          TRACE_EVENT_NESTABLE_ASYNC_END1("ServiceWorker",
+                                          "FetchDataLoaderAsDataPipe", this,
+                                          "MojoResult", mojo_result);
+          result = consumer_->EndRead(0);
+          Cancel();
+          client_->DidFetchDataLoadFailed();
+          return;
+        }
       }
       switch (result) {
         case BytesConsumer::Result::kOk:
           break;
         case BytesConsumer::Result::kShouldWait:
           NOTREACHED();
           return;
         case BytesConsumer::Result::kDone:
-          if (need_to_flush)
-            out_stream_->Flush();
-          out_stream_->Finalize();
-          client_->DidFetchDataLoadedStream();
+          TRACE_EVENT_NESTABLE_ASYNC_END1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "BytesConsumer::Result::Done");
+          Cancel();
+          client_->DidFetchDataLoadedDataPipe();
           return;
         case BytesConsumer::Result::kError:
-          // If the stream is aborted soon after the stream is registered
-          // to the StreamRegistry, ServiceWorkerURLRequestJob may not
-          // notice the error and continue waiting forever.
-          // TODO(yhirano): Add new message to report the error to the
-          // browser process.
-          out_stream_->Abort();
+          TRACE_EVENT_NESTABLE_ASYNC_END1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "BytesConsumer::Result::Error");
+          Cancel();
           client_->DidFetchDataLoadFailed();
           return;
       }
     }
   }
 
-  void Cancel() override { consumer_->Cancel(); }
+  void Cancel() override {
+    consumer_->Cancel();
+    datapipe_watcher_.Cancel();
+    out_datapipe_.reset();
+  }
 
   DEFINE_INLINE_TRACE() {
     visitor->Trace(consumer_);
     visitor->Trace(client_);
-    visitor->Trace(out_stream_);
     FetchDataLoader::Trace(visitor);
     BytesConsumer::Client::Trace(visitor);
   }
 
+ private:
   Member<BytesConsumer> consumer_;
   Member<FetchDataLoader::Client> client_;
-  Member<Stream> out_stream_;
+
+  mojo::ScopedDataPipeProducerHandle out_datapipe_;
+  mojo::SimpleWatcher datapipe_watcher_;
 };
 
 }  // namespace
 
 FetchDataLoader* FetchDataLoader::CreateLoaderAsBlobHandle(
     const String& mime_type) {
   return new FetchDataLoaderAsBlobHandle(mime_type);
 }
 
 FetchDataLoader* FetchDataLoader::CreateLoaderAsArrayBuffer() {
   return new FetchDataLoaderAsArrayBuffer();
 }
 
 FetchDataLoader* FetchDataLoader::CreateLoaderAsString() {
   return new FetchDataLoaderAsString();
 }
 
-FetchDataLoader* FetchDataLoader::CreateLoaderAsStream(Stream* out_stream) {
-  return new FetchDataLoaderAsStream(out_stream);
+FetchDataLoader* FetchDataLoader::CreateLoaderAsDataPipe(
+    mojo::ScopedDataPipeProducerHandle out_datapipe) {
+  return new FetchDataLoaderAsDataPipe(std::move(out_datapipe));
 }
 
 }  // namespace blink