Remove data_pipe_utils from //flutter/tonic

flutter/tonic/BUILD.gn

@@ -24,7 +24,6 @@ source_set("tonic") {
 
   deps = [
     "//dart/runtime/bin:embedded_dart_io",
-    "//mojo/data_pipe_utils",
     "//mojo/public/cpp/system",
   ]
 

flutter/tonic/dart_isolate_reloader.cc

@@ -10,16 +10,15 @@
 #include "flutter/tonic/dart_library_loader.h"
 #include "flutter/tonic/dart_library_provider.h"
 #include "flutter/tonic/dart_state.h"
+#include "glue/drain_data_pipe_job.h"
 #include "glue/thread.h"
 #include "lib/ftl/logging.h"
 #include "lib/ftl/synchronization/monitor.h"
 #include "lib/tonic/converter/dart_converter.h"
 #include "lib/tonic/logging/dart_error.h"
 #include "lib/tonic/scopes/dart_api_scope.h"
 #include "lib/tonic/scopes/dart_isolate_scope.h"
-#include "mojo/data_pipe_utils/data_pipe_drainer.h"
 
-using mojo::common::DataPipeDrainer;
 using tonic::ToDart;
 using tonic::StdStringToDart;
 using tonic::StdStringFromDart;
@@ -35,7 +34,7 @@ class DartIsolateReloader::LoadResult {
              const std::string& url,
              const std::string& library_url,
              const std::string& resolved_url,
-             std::vector<uint8_t> payload)
+             std::vector<char> payload)
       : success_(true),
         tag_(tag),
         url_(url),
@@ -71,8 +70,8 @@ class DartIsolateReloader::LoadResult {
     if (!library_url_.empty()) {
       library = Dart_LookupLibrary(StdStringToDart(library_url_));
     }
-    Dart_Handle source =
-        Dart_NewStringFromUTF8(payload_.data(), payload_.size());
+    Dart_Handle source = Dart_NewStringFromUTF8(
+        reinterpret_cast<const uint8_t*>(payload_.data()), payload_.size());
     Dart_Handle result = Dart_Null();
     switch (tag_) {
       case Dart_kImportTag:
@@ -99,7 +98,7 @@ class DartIsolateReloader::LoadResult {
   std::string library_url_;
   std::string resolved_url_;
   std::string error_;
-  std::vector<uint8_t> payload_;
+  std::vector<char> payload_;
 };
 
 DartIsolateReloader::DartIsolateReloader(DartLibraryProvider* library_provider)
@@ -140,7 +139,7 @@ void DartIsolateReloader::PostResult(std::unique_ptr<LoadResult> load_result) {
 
 // As each source file is requested, a LoadRequest is queued to be processed on
 // worker thread.
-class DartIsolateReloader::LoadRequest : public DataPipeDrainer::Client {
+class DartIsolateReloader::LoadRequest {
  public:
   LoadRequest(DartLibraryProvider* library_provider,
               DartIsolateReloader* isolate_reloader,
@@ -163,30 +162,22 @@ class DartIsolateReloader::LoadRequest : public DataPipeDrainer::Client {
           new DartIsolateReloader::LoadResult(
               tag_, url_, library_url_,
               "File " + url_ + " could not be read."));
-      LOG(ERROR) << "Load failed for " << url_;
+      FTL_LOG(ERROR) << "Load failed for " << url_;
       isolate_reloader_->PostResult(std::move(result));
       // We are finished with this request.
       delete this;
       return;
     }
     resolved_url_ = std::move(resolved_url);
-    drainer_.reset(new DataPipeDrainer(this, std::move(handle)));
-  }
-
-  // DataPipeDrainer::Client
-  void OnDataAvailable(const void* data, size_t num_bytes) override {
-    const uint8_t* bytes = static_cast<const uint8_t*>(data);
-    buffer_.insert(buffer_.end(), bytes, bytes + num_bytes);
-  }
-
-  // DataPipeDrainer::Client
-  void OnDataComplete() override {
-    std::unique_ptr<DartIsolateReloader::LoadResult> result(
-        new DartIsolateReloader::LoadResult(tag_, url_, library_url_,
-                                            resolved_url_, std::move(buffer_)));
-    isolate_reloader_->PostResult(std::move(result));
-    // We are finished with this request.
-    delete this;
+    drainer_.reset(new glue::DrainDataPipeJob(
+        std::move(handle), [this](std::vector<char> buffer) {
+          std::unique_ptr<DartIsolateReloader::LoadResult> result(
+              new DartIsolateReloader::LoadResult(
+                  tag_, url_, library_url_, resolved_url_, std::move(buffer)));
+          isolate_reloader_->PostResult(std::move(result));
+          // We are finished with this request.
+          delete this;
+        }));
   }
 
  private:
@@ -196,7 +187,7 @@ class DartIsolateReloader::LoadRequest : public DataPipeDrainer::Client {
   std::string library_url_;
   std::string resolved_url_;
   std::vector<uint8_t> buffer_;
-  std::unique_ptr<DataPipeDrainer> drainer_;
+  std::unique_ptr<glue::DrainDataPipeJob> drainer_;
 };
 
 void DartIsolateReloader::RequestTask(DartLibraryProvider* library_provider,

flutter/tonic/dart_library_loader.cc

@@ -10,17 +10,16 @@
 #include "flutter/tonic/dart_isolate_reloader.h"
 #include "flutter/tonic/dart_library_provider.h"
 #include "flutter/tonic/dart_state.h"
+#include "glue/drain_data_pipe_job.h"
 #include "glue/trace_event.h"
 #include "lib/tonic/converter/dart_converter.h"
 #include "lib/tonic/dart_persistent_value.h"
 #include "lib/tonic/logging/dart_error.h"
 #include "lib/tonic/scopes/dart_api_scope.h"
 #include "lib/tonic/scopes/dart_isolate_scope.h"
-#include "mojo/data_pipe_utils/data_pipe_drainer.h"
 
-using mojo::common::DataPipeDrainer;
 using tonic::StdStringFromDart;
-using tonic::StdStringToDart;
+using tonic::ToDart;
 
 namespace blink {
 
@@ -33,13 +32,13 @@ void EraseUniquePtr(C& container, T* item) {
   container.erase(key);
   key.release();
 }
-}
+
+}  // namespace
 
 // A DartLibraryLoader::Job represents a network load. It fetches data from the
 // network and buffers the data in std::vector. To cancel the job, delete this
 // object.
-class DartLibraryLoader::Job : public DartDependency,
-                               public DataPipeDrainer::Client {
+class DartLibraryLoader::Job : public DartDependency {
  public:
   Job(DartLibraryLoader* loader, const std::string& name)
       : loader_(loader), name_(name) {
@@ -48,13 +47,12 @@ class DartLibraryLoader::Job : public DartDependency,
   }
 
   const std::string& name() const { return name_; }
-
   const std::string& resolved_url() const { return resolved_url_; }
 
  protected:
   DartLibraryLoader* loader_;
-  // TODO(abarth): Should we be using SharedBuffer to buffer the data?
-  std::vector<uint8_t> buffer_;
+
+  virtual void OnDataAvailable(std::vector<char> buffer) = 0;
 
  private:
   void OnStreamAvailable(mojo::ScopedDataPipeConsumerHandle pipe,
@@ -64,20 +62,15 @@ class DartLibraryLoader::Job : public DartDependency,
       return;
     }
     resolved_url_ = resolved_url;
-    drainer_ = std::unique_ptr<DataPipeDrainer>(
-        new DataPipeDrainer(this, pipe.Pass()));
-  }
-
-  // DataPipeDrainer::Client
-  void OnDataAvailable(const void* data, size_t num_bytes) override {
-    const uint8_t* bytes = static_cast<const uint8_t*>(data);
-    buffer_.insert(buffer_.end(), bytes, bytes + num_bytes);
+    drainer_.reset(new glue::DrainDataPipeJob(
+        std::move(pipe), [this](std::vector<char> buffer) {
+          OnDataAvailable(std::move(buffer));
+        }));
   }
-  // Subclasses must implement OnDataComplete.
 
   std::string name_;
   std::string resolved_url_;
-  std::unique_ptr<DataPipeDrainer> drainer_;
+  std::unique_ptr<glue::DrainDataPipeJob> drainer_;
 };
 
 class DartLibraryLoader::ImportJob : public Job {
@@ -93,10 +86,9 @@ class DartLibraryLoader::ImportJob : public Job {
   bool should_load_as_script() const { return should_load_as_script_; }
 
  private:
-  // DataPipeDrainer::Client
-  void OnDataComplete() override {
+  void OnDataAvailable(std::vector<char> buffer) override {
     TRACE_EVENT_ASYNC_END0("flutter", "DartLibraryLoader::ImportJob", this);
-    loader_->DidCompleteImportJob(this, buffer_);
+    loader_->DidCompleteImportJob(this, std::move(buffer));
   }
 
   bool should_load_as_script_;
@@ -115,10 +107,9 @@ class DartLibraryLoader::SourceJob : public Job {
   Dart_PersistentHandle library() const { return library_.value(); }
 
  private:
-  // DataPipeDrainer::Client
-  void OnDataComplete() override {
+  void OnDataAvailable(std::vector<char> buffer) override {
     TRACE_EVENT_ASYNC_END0("flutter", "DartLibraryLoader::SourceJob", this);
-    loader_->DidCompleteSourceJob(this, buffer_);
+    loader_->DidCompleteSourceJob(this, std::move(buffer));
   }
 
   tonic::DartPersistentValue library_;
@@ -281,50 +272,49 @@ Dart_Handle DartLibraryLoader::CanonicalizeURL(Dart_Handle library,
   return library_provider_->CanonicalizeURL(library, url);
 }
 
-void DartLibraryLoader::DidCompleteImportJob(
-    ImportJob* job,
-    const std::vector<uint8_t>& buffer) {
+void DartLibraryLoader::DidCompleteImportJob(ImportJob* job,
+                                             std::vector<char> buffer) {
   tonic::DartIsolateScope scope(dart_state_->isolate());
   tonic::DartApiScope api_scope;
 
   WatcherSignaler watcher_signaler(*this, job);
 
   Dart_Handle result;
+  Dart_Handle source = Dart_NewStringFromUTF8(
+      reinterpret_cast<const uint8_t*>(buffer.data()), buffer.size());
 
   if (job->should_load_as_script()) {
-    result = Dart_LoadScript(
-        StdStringToDart(job->name()), StdStringToDart(job->resolved_url()),
-        Dart_NewStringFromUTF8(buffer.data(), buffer.size()), 0, 0);
+    result = Dart_LoadScript(ToDart(job->name()), ToDart(job->resolved_url()),
+                             source, 0, 0);
   } else {
-    result = Dart_LoadLibrary(
-        StdStringToDart(job->name()), StdStringToDart(job->resolved_url()),
-        Dart_NewStringFromUTF8(buffer.data(), buffer.size()), 0, 0);
+    result = Dart_LoadLibrary(ToDart(job->name()), ToDart(job->resolved_url()),
+                              source, 0, 0);
   }
   if (Dart_IsError(result)) {
-    LOG(ERROR) << "Error Loading " << job->name() << " "
-               << Dart_GetError(result);
+    FTL_LOG(ERROR) << "Error Loading " << job->name() << " "
+                   << Dart_GetError(result);
   }
 
   pending_libraries_.erase(job->name());
   EraseUniquePtr<Job>(jobs_, job);
 }
 
-void DartLibraryLoader::DidCompleteSourceJob(
-    SourceJob* job,
-    const std::vector<uint8_t>& buffer) {
+void DartLibraryLoader::DidCompleteSourceJob(SourceJob* job,
+                                             std::vector<char> buffer) {
   tonic::DartIsolateScope scope(dart_state_->isolate());
   tonic::DartApiScope api_scope;
 
   WatcherSignaler watcher_signaler(*this, job);
 
+  Dart_Handle source = Dart_NewStringFromUTF8(
+      reinterpret_cast<const uint8_t*>(buffer.data()), buffer.size());
   Dart_Handle result = Dart_LoadSource(
-      Dart_HandleFromPersistent(job->library()), StdStringToDart(job->name()),
-      StdStringToDart(job->resolved_url()),
-      Dart_NewStringFromUTF8(buffer.data(), buffer.size()), 0, 0);
+      Dart_HandleFromPersistent(job->library()), ToDart(job->name()),
+      ToDart(job->resolved_url()), source, 0, 0);
 
   if (Dart_IsError(result)) {
-    LOG(ERROR) << "Error Loading " << job->name() << " "
-               << Dart_GetError(result);
+    FTL_LOG(ERROR) << "Error Loading " << job->name() << " "
+                   << Dart_GetError(result);
   }
 
   EraseUniquePtr<Job>(jobs_, job);
@@ -336,7 +326,7 @@ void DartLibraryLoader::DidFailJob(Job* job) {
 
   WatcherSignaler watcher_signaler(*this, job);
 
-  LOG(ERROR) << "Library Load failed: " << job->name();
+  FTL_LOG(ERROR) << "Library Load failed: " << job->name();
   // TODO(eseidel): Call Dart_LibraryHandleError in the SourceJob case?
 
   EraseUniquePtr<Job>(jobs_, job);

flutter/tonic/dart_library_loader.h

@@ -67,8 +67,8 @@ class DartLibraryLoader {
   Dart_Handle Import(Dart_Handle library, Dart_Handle url);
   Dart_Handle Source(Dart_Handle library, Dart_Handle url);
   Dart_Handle CanonicalizeURL(Dart_Handle library, Dart_Handle url);
-  void DidCompleteImportJob(ImportJob* job, const std::vector<uint8_t>& buffer);
-  void DidCompleteSourceJob(SourceJob* job, const std::vector<uint8_t>& buffer);
+  void DidCompleteImportJob(ImportJob* job, std::vector<char> buffer);
+  void DidCompleteSourceJob(SourceJob* job, std::vector<char> buffer);
   void DidFailJob(Job* job);
 
   DartState* dart_state_;

flutter/tonic/dart_snapshot_loader.cc

@@ -26,7 +26,7 @@ void DartSnapshotLoader::LoadSnapshot(mojo::ScopedDataPipeConsumerHandle pipe,
                                       const ftl::Closure& callback) {
   TRACE_EVENT_ASYNC_BEGIN0("flutter", "DartSnapshotLoader::LoadSnapshot", this);
 
-  drain_job_.reset(new glue::DrainDataPipeJob(
+  drainer_.reset(new glue::DrainDataPipeJob(
       std::move(pipe), [this, callback](std::vector<char> buffer) {
         TRACE_EVENT_ASYNC_END0("flutter", "DartSnapshotLoader::LoadSnapshot",
                                this);

flutter/tonic/dart_snapshot_loader.h

@@ -25,7 +25,7 @@ class DartSnapshotLoader {
 
  private:
   ftl::WeakPtr<tonic::DartState> dart_state_;
-  std::unique_ptr<glue::DrainDataPipeJob> drain_job_;
+  std::unique_ptr<glue::DrainDataPipeJob> drainer_;
 
   FTL_DISALLOW_COPY_AND_ASSIGN(DartSnapshotLoader);
 };

sky/engine/core/script/dart_controller.cc

@@ -19,7 +19,6 @@
 #include "lib/tonic/logging/dart_invoke.h"
 #include "lib/tonic/scopes/dart_api_scope.h"
 #include "lib/tonic/scopes/dart_isolate_scope.h"
-#include "mojo/data_pipe_utils/data_pipe_utils.h"
 #include "mojo/public/cpp/system/data_pipe.h"
 #include "sky/engine/bindings/dart_mojo_internal.h"
 #include "sky/engine/bindings/dart_runtime_hooks.h"

sky/engine/platform/BUILD.gn

@@ -297,7 +297,6 @@ source_set("platform") {
   deps = [
     "//base:base",
     "//mojo/application",
-    "//mojo/data_pipe_utils",
     "//mojo/environment:chromium",
     "//mojo/public/c/system",
     "//mojo/public/cpp/bindings",