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

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

Index: third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp
diff --git a/third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp b/third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp
index e5b6ce557785ad9816bf866ffd45be9bc671477d..ed7e96e510a60cc1ed112394964226878770c123 100644
--- a/third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp
+++ b/third_party/WebKit/Source/modules/fetch/FetchDataLoader.cpp
@@ -4,13 +4,16 @@
 
 #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 {
 
@@ -231,39 +234,75 @@ class FetchDataLoaderAsString final : public FetchDataLoader,
   StringBuilder m_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* outStream)
-      : m_outStream(outStream) {}
-
+  explicit FetchDataLoaderAsDataPipe(
+      mojo::ScopedDataPipeProducerHandle outDataPipe)
+      : m_outDataPipe(std::move(outDataPipe)),
+        m_dataPipeWatcher(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(!m_client);
     DCHECK(!m_consumer);
+    m_dataPipeWatcher.Watch(
+        m_outDataPipe.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+        convertToBaseCallback(WTF::bind(&FetchDataLoaderAsDataPipe::onWritable,
+                                        wrapWeakPersistent(this))));
     m_client = client;
     m_consumer = consumer;
     m_consumer->setClient(this);
+  }
+
+  void onWritable(MojoResult result) {
+    if (result != MOJO_RESULT_OK)
+      cancel();
     onStateChange();
   }
 
+  // Implements BytesConsumer::Client.
   void onStateChange() override {
-    bool needToFlush = false;
-    while (true) {
+    bool shouldWait = false;
+    while (!shouldWait) {
       const char* buffer;
       size_t available;
       auto result = m_consumer->beginRead(&buffer, &available);
-      if (result == BytesConsumer::Result::ShouldWait) {
-        if (needToFlush)
-          m_outStream->flush();
+      if (result == BytesConsumer::Result::ShouldWait)
         return;
-      }
       if (result == BytesConsumer::Result::Ok) {
-        m_outStream->addData(buffer, available);
-        needToFlush = true;
-        result = m_consumer->endRead(available);
+        DCHECK_GT(available, 0UL);
+        uint32_t numBytes = available;
+        MojoResult mojoResult = mojo::WriteDataRaw(
+            m_outDataPipe.get(), buffer, &numBytes, MOJO_WRITE_DATA_FLAG_NONE);
+        if (mojoResult == MOJO_RESULT_OK) {
+          TRACE_EVENT_NESTABLE_ASYNC_INSTANT2(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "MOJO_RESULT_OK", "bytes", numBytes);
+          result = m_consumer->endRead(numBytes);
+        } else if (mojoResult == MOJO_RESULT_SHOULD_WAIT) {
+          TRACE_EVENT_NESTABLE_ASYNC_INSTANT1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "MOJO_RESULT_SHOULD_WAIT");
+          result = m_consumer->endRead(0);
+          shouldWait = true;
+        } else {
+          TRACE_EVENT_NESTABLE_ASYNC_END1("ServiceWorker",
+                                          "FetchDataLoaderAsDataPipe", this,
+                                          "MojoResult", mojoResult);
+          result = m_consumer->endRead(0);
+          cancel();
+          m_client->didFetchDataLoadFailed();
+          return;
+        }
       }
       switch (result) {
         case BytesConsumer::Result::Ok:
@@ -272,37 +311,43 @@ class FetchDataLoaderAsStream final : public FetchDataLoader,
           NOTREACHED();
           return;
         case BytesConsumer::Result::Done:
-          if (needToFlush)
-            m_outStream->flush();
-          m_outStream->finalize();
-          m_client->didFetchDataLoadedStream();
+          TRACE_EVENT_NESTABLE_ASYNC_END1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "BytesConsumer::Result::Done");
+          cancel();
+          m_client->didFetchDataLoadedDataPipe();
           return;
         case BytesConsumer::Result::Error:
-          // 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.
-          m_outStream->abort();
+          TRACE_EVENT_NESTABLE_ASYNC_END1(
+              "ServiceWorker", "FetchDataLoaderAsDataPipe", this, "state",
+              "BytesConsumer::Result::Error");
+          // TODO(shimazu): Some mechanism to notify error state is needed.
+          cancel();
           m_client->didFetchDataLoadFailed();
           return;
       }
     }
   }
 
-  void cancel() override { m_consumer->cancel(); }
+  void cancel() override {
+    m_consumer->cancel();
+    m_dataPipeWatcher.Cancel();
+    m_outDataPipe.reset();
+  }
 
   DEFINE_INLINE_TRACE() {
     visitor->trace(m_consumer);
     visitor->trace(m_client);
-    visitor->trace(m_outStream);
     FetchDataLoader::trace(visitor);
     BytesConsumer::Client::trace(visitor);
   }
 
+ private:
   Member<BytesConsumer> m_consumer;
   Member<FetchDataLoader::Client> m_client;
-  Member<Stream> m_outStream;
+
+  mojo::ScopedDataPipeProducerHandle m_outDataPipe;
+  mojo::SimpleWatcher m_dataPipeWatcher;
 };
 
 }  // namespace
@@ -320,8 +365,9 @@ FetchDataLoader* FetchDataLoader::createLoaderAsString() {
   return new FetchDataLoaderAsString();
 }
 
-FetchDataLoader* FetchDataLoader::createLoaderAsStream(Stream* outStream) {
-  return new FetchDataLoaderAsStream(outStream);
+FetchDataLoader* FetchDataLoader::createLoaderAsDataPipe(
+    mojo::ScopedDataPipeProducerHandle outDataPipe) {
+  return new FetchDataLoaderAsDataPipe(std::move(outDataPipe));
 }
 
 }  // namespace blink