Factor DrainDataPipeJob out of MojoLoader

We want to re-use the code for draining Mojo data pipes in more places that
just the MojoLoader.

R=eseidel@chromium.org

Review URL: https://codereview.chromium.org/650903006

engine/core/loader/MojoLoader.cpp

@@ -21,7 +21,6 @@ using namespace mojo;
 
 MojoLoader::MojoLoader(LocalFrame& frame)
     : m_frame(frame)
-    , m_weakFactory(this)
 {
 }
 
@@ -44,39 +43,18 @@ void MojoLoader::load(const KURL& url, ScopedDataPipeConsumerHandle responseStre
     // response headers and set them on Document::contentLanguage.
 
     document->startParsing();
-    m_responseStream = responseStream.Pass();
-    readMore();
-}
 
-void MojoLoader::readMore()
-{
-    const void* buf = nullptr;
-    uint32_t buf_size = 0;
-    MojoResult rv = BeginReadDataRaw(m_responseStream.get(),
-        &buf, &buf_size, MOJO_READ_DATA_FLAG_NONE);
-    if (rv == MOJO_RESULT_OK) {
-        m_frame.document()->parser()->appendBytes(static_cast<const char*>(buf), buf_size);
-        EndReadDataRaw(m_responseStream.get(), buf_size);
-        waitToReadMore();
-    } else if (rv == MOJO_RESULT_SHOULD_WAIT) {
-        waitToReadMore();
-    } else if (rv == MOJO_RESULT_FAILED_PRECONDITION) {
-        m_frame.document()->parser()->finish();
-    } else {
-        ASSERT_NOT_REACHED();
-    }
+    m_drainJob = adoptPtr(new DataPipeDrainer(this, responseStream.Pass()));
 }
 
-void MojoLoader::waitToReadMore()
+void MojoLoader::OnDataAvailable(const void* data, size_t numberOfBytes)
 {
-    m_handleWatcher.Start(m_responseStream.get(),
-        MOJO_HANDLE_SIGNAL_READABLE, MOJO_DEADLINE_INDEFINITE,
-        base::Bind(&MojoLoader::moreDataReady,m_weakFactory.GetWeakPtr()));
+    m_frame.document()->parser()->appendBytes(static_cast<const char*>(data), numberOfBytes);
 }
 
-void MojoLoader::moreDataReady(MojoResult result)
+void MojoLoader::OnDataComplete()
 {
-    readMore();
+    m_frame.document()->parser()->finish();
 }
 
 }

engine/core/loader/MojoLoader.h

@@ -8,13 +8,14 @@
 #include "base/memory/weak_ptr.h"
 #include "mojo/common/handle_watcher.h"
 #include "mojo/public/cpp/system/data_pipe.h"
+#include "platform/fetcher/DataPipeDrainer.h"
 #include "platform/weborigin/KURL.h"
 
 namespace blink {
 
 class LocalFrame;
 
-class MojoLoader {
+class MojoLoader : public DataPipeDrainer::Client {
 public:
     explicit MojoLoader(LocalFrame&);
 
@@ -23,14 +24,11 @@ public:
 private:
     LocalFrame& m_frame;
 
-    // FIXME: These belong on a helper object for async reading from mojo pipes.
-    void readMore();
-    void waitToReadMore();
-    void moreDataReady(MojoResult);
+    // From DataPipeDrainer::Client
+    void OnDataAvailable(const void* data, size_t numberOfBytes) override;
+    void OnDataComplete() override;
 
-    mojo::common::HandleWatcher m_handleWatcher;
-    mojo::ScopedDataPipeConsumerHandle m_responseStream;
-    base::WeakPtrFactory<MojoLoader> m_weakFactory;
+    OwnPtr<DataPipeDrainer> m_drainJob;
 };
 
 }

engine/platform/BUILD.gn

@@ -242,6 +242,8 @@ component("platform") {
     "exported/WrappedResourceResponse.h",
     "exported/linux/WebFontInfo.cpp",
     "exported/linux/WebFontRenderStyle.cpp",
+    "fetcher/DataPipeDrainer.cpp",
+    "fetcher/DataPipeDrainer.h",
     "fonts/AlternateFontFamily.h",
     "fonts/Character.cpp",
     "fonts/Character.h",
@@ -683,6 +685,10 @@ component("platform") {
     "//third_party/libpng",
     "//third_party/ots",
     "//third_party/qcms",
+    "//mojo/application",
+    "//mojo/public/cpp/bindings",
+    "//mojo/public/cpp/utility",
+    "//mojo/public/c/system:for_shared_library",
     "//sky/engine/wtf",
     "//sky/engine/platform/heap",
     "//url",

engine/platform/DEPS

@@ -3,6 +3,8 @@ include_rules = [
     "+platform",
     "+public/platform",
     "+skia/ext",
+    "+mojo/public",
+    "+mojo/common",
     "+third_party/khronos",
     "+third_party/skia",
     "+url",

engine/platform/fetcher/DataPipeDrainer.cpp

+// Copyright 2014 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 "config.h"
+#include "platform/fetcher/DataPipeDrainer.h"
+
+#include "base/bind.h"
+
+namespace blink {
+
+DataPipeDrainer::DataPipeDrainer(Client* client,
+                                 mojo::ScopedDataPipeConsumerHandle source)
+    : client_(client),
+      source_(source.Pass()),
+      weak_factory_(this) {
+  DCHECK(client_);
+  ReadData();
+}
+
+DataPipeDrainer::~DataPipeDrainer() {
+}
+
+void DataPipeDrainer::ReadData() {
+  const void* buffer = nullptr;
+  uint32_t num_bytes = 0;
+  MojoResult rv = BeginReadDataRaw(source_.get(),
+      &buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+  if (rv == MOJO_RESULT_OK) {
+    client_->OnDataAvailable(buffer, num_bytes);
+    EndReadDataRaw(source_.get(), num_bytes);
+    WaitForData();
+  } else if (rv == MOJO_RESULT_SHOULD_WAIT) {
+    WaitForData();
+  } else if (rv == MOJO_RESULT_FAILED_PRECONDITION) {
+    client_->OnDataComplete();
+  } else {
+    DCHECK(false);
+  }
+}
+
+void DataPipeDrainer::WaitForData() {
+  handle_watcher_.Start(source_.get(),
+      MOJO_HANDLE_SIGNAL_READABLE, MOJO_DEADLINE_INDEFINITE,
+      base::Bind(&DataPipeDrainer::WaitComplete, weak_factory_.GetWeakPtr()));
+}
+
+void DataPipeDrainer::WaitComplete(MojoResult result) {
+  ReadData();
+}
+
+}  // namespace blink

engine/platform/fetcher/DataPipeDrainer.h

+// Copyright 2014 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 SKY_ENGINE_PLATFORM_FETCHER_DATA_PIPE_DRAINER_H_
+#define SKY_ENGINE_PLATFORM_FETCHER_DATA_PIPE_DRAINER_H_
+
+#include "base/memory/weak_ptr.h"
+#include "mojo/common/handle_watcher.h"
+#include "mojo/public/cpp/system/core.h"
+
+namespace blink {
+
+class DataPipeDrainer {
+ public:
+  class Client {
+   public:
+    virtual void OnDataAvailable(const void* data, size_t num_bytes) = 0;
+    virtual void OnDataComplete() = 0;
+
+   protected:
+    virtual ~Client() { }
+  };
+
+  DataPipeDrainer(Client*, mojo::ScopedDataPipeConsumerHandle source);
+  ~DataPipeDrainer();
+
+ private:
+  void ReadData();
+  void WaitForData();
+  void WaitComplete(MojoResult result);
+
+  Client* client_;
+  mojo::ScopedDataPipeConsumerHandle source_;
+  mojo::common::HandleWatcher handle_watcher_;
+
+  base::WeakPtrFactory<DataPipeDrainer> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeDrainer);
+};
+
+}  // namespace blink
+
+#endif  // SKY_ENGINE_PLATFORM_FETCHER_DATA_PIPE_DRAINER_H_