fix: implement ses.getBlobData() for NetworkService (#20041)

* pass data pipe to JS

* implement reading buffer

* re-enable ses.getBlobData test

* remove AtomBlobReader

filenames.gni

@@ -47,6 +47,8 @@ filenames = {
     "shell/browser/api/atom_api_content_tracing.cc",
     "shell/browser/api/atom_api_cookies.cc",
     "shell/browser/api/atom_api_cookies.h",
+    "shell/browser/api/atom_api_data_pipe_holder.cc",
+    "shell/browser/api/atom_api_data_pipe_holder.h",
     "shell/browser/api/atom_api_debugger.cc",
     "shell/browser/api/atom_api_debugger.h",
     "shell/browser/api/atom_api_dialog.cc",
@@ -129,8 +131,6 @@ filenames = {
     "shell/browser/atom_autofill_driver_factory.h",
     "shell/browser/atom_autofill_driver.cc",
     "shell/browser/atom_autofill_driver.h",
-    "shell/browser/atom_blob_reader.cc",
-    "shell/browser/atom_blob_reader.h",
     "shell/browser/atom_browser_client.cc",
     "shell/browser/atom_browser_client.h",
     "shell/browser/atom_browser_context.cc",

shell/browser/api/atom_api_data_pipe_holder.cc

@@ -0,0 +1,185 @@
+// Copyright (c) 2019 GitHub, Inc.
+// Use of this source code is governed by the MIT license that can be
+// found in the LICENSE file.
+
+#include "shell/browser/api/atom_api_data_pipe_holder.h"
+
+#include <utility>
+#include <vector>
+
+#include "base/memory/weak_ptr.h"
+#include "base/strings/string_number_conversions.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/base/net_errors.h"
+#include "shell/common/key_weak_map.h"
+#include "shell/common/promise_util.h"
+
+#include "shell/common/node_includes.h"
+
+namespace electron {
+
+namespace api {
+
+namespace {
+
+// Incremental ID.
+int g_next_id = 0;
+
+// Map that manages all the DataPipeHolder objects.
+KeyWeakMap<std::string> g_weak_map;
+
+// Utility class to read from data pipe.
+class DataPipeReader {
+ public:
+  DataPipeReader(util::Promise<v8::Local<v8::Value>> promise,
+                 network::mojom::DataPipeGetterPtr data_pipe_getter)
+      : promise_(std::move(promise)),
+        data_pipe_getter_(std::move(data_pipe_getter)),
+        handle_watcher_(FROM_HERE,
+                        mojo::SimpleWatcher::ArmingPolicy::MANUAL,
+                        base::SequencedTaskRunnerHandle::Get()) {
+    // Get a new data pipe and start.
+    mojo::DataPipe data_pipe;
+    data_pipe_getter_->Read(std::move(data_pipe.producer_handle),
+                            base::BindOnce(&DataPipeReader::ReadCallback,
+                                           weak_factory_.GetWeakPtr()));
+    data_pipe_ = std::move(data_pipe.consumer_handle);
+    handle_watcher_.Watch(data_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+                          base::BindRepeating(&DataPipeReader::OnHandleReadable,
+                                              weak_factory_.GetWeakPtr()));
+  }
+
+  ~DataPipeReader() = default;
+
+ private:
+  // Callback invoked by DataPipeGetter::Read.
+  void ReadCallback(int32_t status, uint64_t size) {
+    if (status != net::OK) {
+      OnFailure();
+      return;
+    }
+    buffer_.resize(size);
+    head_ = &buffer_.front();
+    remaining_size_ = size;
+    handle_watcher_.ArmOrNotify();
+  }
+
+  // Called by |handle_watcher_| when data is available or the pipe was closed,
+  // and there's a pending Read() call.
+  void OnHandleReadable(MojoResult result) {
+    if (result != MOJO_RESULT_OK) {
+      OnFailure();
+      return;
+    }
+
+    // Read.
+    uint32_t length = remaining_size_;
+    result = data_pipe_->ReadData(head_, &length, MOJO_READ_DATA_FLAG_NONE);
+    if (result == MOJO_RESULT_OK) {  // success
+      remaining_size_ -= length;
+      head_ += length;
+      if (remaining_size_ == 0)
+        OnSuccess();
+    } else if (result == MOJO_RESULT_SHOULD_WAIT) {  // IO pending
+      handle_watcher_.ArmOrNotify();
+    } else {  // error
+      OnFailure();
+    }
+  }
+
+  void OnFailure() {
+    promise_.RejectWithErrorMessage("Could not get blob data");
+    delete this;
+  }
+
+  void OnSuccess() {
+    // Pass the buffer to JS.
+    //
+    // Note that the lifetime of the native buffer belongs to us, and we will
+    // free memory when JS buffer gets garbage collected.
+    v8::Locker locker(promise_.isolate());
+    v8::HandleScope handle_scope(promise_.isolate());
+    v8::Local<v8::Value> buffer =
+        node::Buffer::New(promise_.isolate(), &buffer_.front(), buffer_.size(),
+                          &DataPipeReader::FreeBuffer, this)
+            .ToLocalChecked();
+    promise_.Resolve(buffer);
+
+    // Destroy data pipe.
+    handle_watcher_.Cancel();
+    data_pipe_.reset();
+    data_pipe_getter_ = nullptr;
+  }
+
+  static void FreeBuffer(char* data, void* self) {
+    delete static_cast<DataPipeReader*>(self);
+  }
+
+  util::Promise<v8::Local<v8::Value>> promise_;
+
+  network::mojom::DataPipeGetterPtr data_pipe_getter_;
+  mojo::ScopedDataPipeConsumerHandle data_pipe_;
+  mojo::SimpleWatcher handle_watcher_;
+
+  // Stores read data.
+  std::vector<char> buffer_;
+
+  // The head of buffer.
+  char* head_ = nullptr;
+
+  // Remaining data to read.
+  uint64_t remaining_size_ = 0;
+
+  base::WeakPtrFactory<DataPipeReader> weak_factory_{this};
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeReader);
+};
+
+}  // namespace
+
+gin::WrapperInfo DataPipeHolder::kWrapperInfo = {gin::kEmbedderNativeGin};
+
+DataPipeHolder::DataPipeHolder(const network::DataElement& element)
+    : id_(base::NumberToString(++g_next_id)),
+      data_pipe_(element.CloneDataPipeGetter()) {}
+
+DataPipeHolder::~DataPipeHolder() = default;
+
+v8::Local<v8::Promise> DataPipeHolder::ReadAll(v8::Isolate* isolate) {
+  util::Promise<v8::Local<v8::Value>> promise(isolate);
+  v8::Local<v8::Promise> handle = promise.GetHandle();
+  if (!data_pipe_) {
+    promise.RejectWithErrorMessage("Could not get blob data");
+    return handle;
+  }
+
+  new DataPipeReader(std::move(promise), std::move(data_pipe_));
+  return handle;
+}
+
+// static
+gin::Handle<DataPipeHolder> DataPipeHolder::Create(
+    v8::Isolate* isolate,
+    const network::DataElement& element) {
+  auto handle = gin::CreateHandle(isolate, new DataPipeHolder(element));
+  g_weak_map.Set(isolate, handle->id(),
+                 handle->GetWrapper(isolate).ToLocalChecked());
+  return handle;
+}
+
+// static
+gin::Handle<DataPipeHolder> DataPipeHolder::From(v8::Isolate* isolate,
+                                                 const std::string& id) {
+  v8::MaybeLocal<v8::Object> object = g_weak_map.Get(isolate, id);
+  if (!object.IsEmpty()) {
+    gin::Handle<DataPipeHolder> handle;
+    if (gin::ConvertFromV8(isolate, object.ToLocalChecked(), &handle))
+      return handle;
+  }
+  return gin::Handle<DataPipeHolder>();
+}
+
+}  // namespace api
+
+}  // namespace electron

shell/browser/api/atom_api_data_pipe_holder.h

@@ -0,0 +1,53 @@
+// Copyright (c) 2019 GitHub, Inc.
+// Use of this source code is governed by the MIT license that can be
+// found in the LICENSE file.
+
+#ifndef SHELL_BROWSER_API_ATOM_API_DATA_PIPE_HOLDER_H_
+#define SHELL_BROWSER_API_ATOM_API_DATA_PIPE_HOLDER_H_
+
+#include <string>
+
+#include "gin/handle.h"
+#include "gin/wrappable.h"
+#include "services/network/public/cpp/data_element.h"
+#include "services/network/public/mojom/data_pipe_getter.mojom.h"
+
+namespace electron {
+
+namespace api {
+
+// Retains reference to the data pipe.
+class DataPipeHolder : public gin::Wrappable<DataPipeHolder> {
+ public:
+  static gin::WrapperInfo kWrapperInfo;
+
+  static gin::Handle<DataPipeHolder> Create(
+      v8::Isolate* isolate,
+      const network::DataElement& element);
+  static gin::Handle<DataPipeHolder> From(v8::Isolate* isolate,
+                                          const std::string& id);
+
+  // Read all data at once.
+  //
+  // TODO(zcbenz): This is apparently not suitable for really large data, but
+  // no one has complained about it yet.
+  v8::Local<v8::Promise> ReadAll(v8::Isolate* isolate);
+
+  // The unique ID that can be used to receive the object.
+  const std::string& id() const { return id_; }
+
+ private:
+  explicit DataPipeHolder(const network::DataElement& element);
+  ~DataPipeHolder() override;
+
+  std::string id_;
+  network::mojom::DataPipeGetterPtr data_pipe_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeHolder);
+};
+
+}  // namespace api
+
+}  // namespace electron
+
+#endif  // SHELL_BROWSER_API_ATOM_API_DATA_PIPE_HOLDER_H_

shell/browser/api/atom_api_session.cc

@@ -41,6 +41,7 @@
 #include "services/network/network_service.h"
 #include "services/network/public/cpp/features.h"
 #include "shell/browser/api/atom_api_cookies.h"
+#include "shell/browser/api/atom_api_data_pipe_holder.h"
 #include "shell/browser/api/atom_api_download_item.h"
 #include "shell/browser/api/atom_api_net_log.h"
 #include "shell/browser/api/atom_api_protocol_ns.h"
@@ -502,15 +503,14 @@ std::string Session::GetUserAgent() {
 
 v8::Local<v8::Promise> Session::GetBlobData(v8::Isolate* isolate,
                                             const std::string& uuid) {
-  util::Promise<v8::Local<v8::Value>> promise(isolate);
-  v8::Local<v8::Promise> handle = promise.GetHandle();
+  gin::Handle<DataPipeHolder> holder = DataPipeHolder::From(isolate, uuid);
+  if (holder.IsEmpty()) {
+    util::Promise<v8::Local<v8::Value>> promise(isolate);
+    promise.RejectWithErrorMessage("Could not get blob data handle");
+    return promise.GetHandle();
+  }
 
-  AtomBlobReader* blob_reader = browser_context()->GetBlobReader();
-  base::PostTaskWithTraits(
-      FROM_HERE, {BrowserThread::IO},
-      base::BindOnce(&AtomBlobReader::StartReading,
-                     base::Unretained(blob_reader), uuid, std::move(promise)));
-  return handle;
+  return holder->ReadAll(isolate);
 }
 
 void Session::DownloadURL(const GURL& url) {

shell/browser/api/atom_api_session.h

@@ -13,7 +13,6 @@
 #include "electron/buildflags/buildflags.h"
 #include "native_mate/handle.h"
 #include "shell/browser/api/trackable_object.h"
-#include "shell/browser/atom_blob_reader.h"
 #include "shell/browser/net/resolve_proxy_helper.h"
 #include "shell/common/promise_util.h"
 

shell/browser/atom_blob_reader.cc

@@ -1,126 +0,0 @@
-// Copyright (c) 2016 GitHub, Inc.
-// Use of this source code is governed by the MIT license that can be
-// found in the LICENSE file.
-
-#include "shell/browser/atom_blob_reader.h"
-
-#include <utility>
-
-#include "base/task/post_task.h"
-#include "content/browser/blob_storage/chrome_blob_storage_context.h"  // nogncheck
-#include "content/public/browser/browser_task_traits.h"
-#include "content/public/browser/browser_thread.h"
-#include "net/base/io_buffer.h"
-#include "net/base/net_errors.h"
-#include "shell/common/node_includes.h"
-#include "storage/browser/blob/blob_data_handle.h"
-#include "storage/browser/blob/blob_reader.h"
-#include "storage/browser/blob/blob_storage_context.h"
-
-using content::BrowserThread;
-
-namespace electron {
-
-namespace {
-
-void FreeNodeBufferData(char* data, void* hint) {
-  delete[] data;
-}
-
-void RunPromiseInUI(util::Promise<v8::Local<v8::Value>> promise,
-                    char* blob_data,
-                    int size) {
-  DCHECK_CURRENTLY_ON(BrowserThread::UI);
-  v8::Isolate* isolate = promise.isolate();
-
-  v8::Locker locker(isolate);
-  v8::HandleScope handle_scope(isolate);
-  if (blob_data) {
-    v8::Local<v8::Value> buffer =
-        node::Buffer::New(isolate, blob_data, static_cast<size_t>(size),
-                          &FreeNodeBufferData, nullptr)
-            .ToLocalChecked();
-    promise.Resolve(buffer);
-  } else {
-    promise.RejectWithErrorMessage("Could not get blob data");
-  }
-}
-
-}  // namespace
-
-AtomBlobReader::AtomBlobReader(content::ChromeBlobStorageContext* blob_context)
-    : blob_context_(blob_context) {}
-
-AtomBlobReader::~AtomBlobReader() {}
-
-void AtomBlobReader::StartReading(const std::string& uuid,
-                                  util::Promise<v8::Local<v8::Value>> promise) {
-  DCHECK_CURRENTLY_ON(BrowserThread::IO);
-
-  auto blob_data_handle = blob_context_->context()->GetBlobDataFromUUID(uuid);
-  if (!blob_data_handle) {
-    util::Promise<v8::Local<v8::Value>>::RejectPromise(
-        std::move(promise), "Could not get blob data handle");
-    return;
-  }
-
-  auto blob_reader = blob_data_handle->CreateReader();
-  BlobReadHelper* blob_read_helper =
-      new BlobReadHelper(std::move(blob_reader),
-                         base::BindOnce(&RunPromiseInUI, std::move(promise)));
-  blob_read_helper->Read();
-}
-
-AtomBlobReader::BlobReadHelper::BlobReadHelper(
-    std::unique_ptr<storage::BlobReader> blob_reader,
-    BlobReadHelper::CompletionCallback callback)
-    : blob_reader_(std::move(blob_reader)),
-      completion_callback_(std::move(callback)) {}
-
-AtomBlobReader::BlobReadHelper::~BlobReadHelper() {}
-
-void AtomBlobReader::BlobReadHelper::Read() {
-  DCHECK_CURRENTLY_ON(BrowserThread::IO);
-
-  storage::BlobReader::Status size_status = blob_reader_->CalculateSize(
-      base::BindOnce(&AtomBlobReader::BlobReadHelper::DidCalculateSize,
-                     base::Unretained(this)));
-  if (size_status != storage::BlobReader::Status::IO_PENDING)
-    DidCalculateSize(net::OK);
-}
-
-void AtomBlobReader::BlobReadHelper::DidCalculateSize(int result) {
-  DCHECK_CURRENTLY_ON(BrowserThread::IO);
-
-  if (result != net::OK) {
-    DidReadBlobData(nullptr, 0);
-    return;
-  }
-
-  uint64_t total_size = blob_reader_->total_size();
-  int bytes_read = 0;
-  scoped_refptr<net::IOBuffer> blob_data =
-      new net::IOBuffer(static_cast<size_t>(total_size));
-  auto callback =
-      base::BindRepeating(&AtomBlobReader::BlobReadHelper::DidReadBlobData,
-                          base::Unretained(this), base::RetainedRef(blob_data));
-  storage::BlobReader::Status read_status =
-      blob_reader_->Read(blob_data.get(), total_size, &bytes_read, callback);
-  if (read_status != storage::BlobReader::Status::IO_PENDING)
-    callback.Run(bytes_read);
-}
-
-void AtomBlobReader::BlobReadHelper::DidReadBlobData(
-    const scoped_refptr<net::IOBuffer>& blob_data,
-    int size) {
-  DCHECK_CURRENTLY_ON(BrowserThread::IO);
-
-  char* data = new char[size];
-  memcpy(data, blob_data->data(), size);
-  base::PostTaskWithTraits(
-      FROM_HERE, {BrowserThread::UI},
-      base::BindOnce(std::move(completion_callback_), data, size));
-  delete this;
-}
-
-}  // namespace electron

shell/browser/atom_blob_reader.h

@@ -1,76 +0,0 @@
-// Copyright (c) 2016 GitHub, Inc.
-// Use of this source code is governed by the MIT license that can be
-// found in the LICENSE file.
-
-#ifndef SHELL_BROWSER_ATOM_BLOB_READER_H_
-#define SHELL_BROWSER_ATOM_BLOB_READER_H_
-
-#include <memory>
-#include <string>
-
-#include "base/callback.h"
-#include "shell/common/promise_util.h"
-
-namespace content {
-class ChromeBlobStorageContext;
-}
-
-namespace net {
-class IOBuffer;
-}
-
-namespace storage {
-class BlobDataHandle;
-class BlobReader;
-}  // namespace storage
-
-namespace v8 {
-template <class T>
-class Local;
-class Value;
-}  // namespace v8
-
-namespace electron {
-
-// A class to keep track of the blob context. All methods,
-// except Ctor are expected to be called on IO thread.
-class AtomBlobReader {
- public:
-  explicit AtomBlobReader(content::ChromeBlobStorageContext* blob_context);
-  ~AtomBlobReader();
-
-  void StartReading(const std::string& uuid,
-                    electron::util::Promise<v8::Local<v8::Value>> promise);
-
- private:
-  // A self-destroyed helper class to read the blob data.
-  // Must be accessed on IO thread.
-  class BlobReadHelper {
-   public:
-    using CompletionCallback = base::OnceCallback<void(char*, int)>;
-
-    BlobReadHelper(std::unique_ptr<storage::BlobReader> blob_reader,
-                   BlobReadHelper::CompletionCallback callback);
-    ~BlobReadHelper();
-
-    void Read();
-
-   private:
-    void DidCalculateSize(int result);
-    void DidReadBlobData(const scoped_refptr<net::IOBuffer>& blob_data,
-                         int bytes_read);
-
-    std::unique_ptr<storage::BlobReader> blob_reader_;
-    BlobReadHelper::CompletionCallback completion_callback_;
-
-    DISALLOW_COPY_AND_ASSIGN(BlobReadHelper);
-  };
-
-  scoped_refptr<content::ChromeBlobStorageContext> blob_context_;
-
-  DISALLOW_COPY_AND_ASSIGN(AtomBlobReader);
-};
-
-}  // namespace electron
-
-#endif  // SHELL_BROWSER_ATOM_BLOB_READER_H_

shell/browser/atom_browser_context.cc

@@ -29,7 +29,6 @@
 #include "net/base/escape.h"
 #include "services/network/public/cpp/features.h"
 #include "services/network/public/cpp/wrapper_shared_url_loader_factory.h"
-#include "shell/browser/atom_blob_reader.h"
 #include "shell/browser/atom_browser_client.h"
 #include "shell/browser/atom_browser_main_parts.h"
 #include "shell/browser/atom_download_manager_delegate.h"
@@ -256,15 +255,6 @@ std::string AtomBrowserContext::GetUserAgent() const {
   return user_agent_;
 }
 
-AtomBlobReader* AtomBrowserContext::GetBlobReader() {
-  if (!blob_reader_.get()) {
-    content::ChromeBlobStorageContext* blob_context =
-        content::ChromeBlobStorageContext::GetFor(this);
-    blob_reader_.reset(new AtomBlobReader(blob_context));
-  }
-  return blob_reader_.get();
-}
-
 predictors::PreconnectManager* AtomBrowserContext::GetPreconnectManager() {
   if (!preconnect_manager_.get()) {
     preconnect_manager_.reset(new predictors::PreconnectManager(nullptr, this));

shell/browser/atom_browser_context.h

@@ -40,7 +40,6 @@ class AtomExtensionSystem;
 
 namespace electron {
 
-class AtomBlobReader;
 class AtomBrowserContext;
 class AtomDownloadManagerDelegate;
 class AtomPermissionManager;
@@ -90,7 +89,6 @@ class AtomBrowserContext
   std::string GetUserAgent() const;
   bool CanUseHttpCache() const;
   int GetMaxCacheSize() const;
-  AtomBlobReader* GetBlobReader();
   ResolveProxyHelper* GetResolveProxyHelper();
   predictors::PreconnectManager* GetPreconnectManager();
   scoped_refptr<network::SharedURLLoaderFactory> GetURLLoaderFactory();
@@ -163,7 +161,6 @@ class AtomBrowserContext
   std::unique_ptr<AtomDownloadManagerDelegate> download_manager_delegate_;
   std::unique_ptr<WebViewManager> guest_manager_;
   std::unique_ptr<AtomPermissionManager> permission_manager_;
-  std::unique_ptr<AtomBlobReader> blob_reader_;
   std::unique_ptr<MediaDeviceIDSalt> media_device_id_salt_;
   scoped_refptr<ResolveProxyHelper> resolve_proxy_helper_;
   scoped_refptr<storage::SpecialStoragePolicy> storage_policy_;

shell/common/gin_converters/net_converter.cc

@@ -22,6 +22,7 @@
 #include "net/cert/x509_util.h"
 #include "net/http/http_response_headers.h"
 #include "services/network/public/cpp/resource_request.h"
+#include "shell/browser/api/atom_api_data_pipe_holder.h"
 #include "shell/common/gin_converters/gurl_converter.h"
 #include "shell/common/gin_converters/std_converter.h"
 #include "shell/common/gin_converters/value_converter_gin_adapter.h"
@@ -264,9 +265,16 @@ v8::Local<v8::Value> Converter<network::ResourceRequestBody>::ToV8(
                                                     element.length())
                                      .ToLocalChecked());
         break;
-      case network::mojom::DataElementType::kBlob:
-        upload_data.Set("blobUUID", element.blob_uuid());
+      case network::mojom::DataElementType::kDataPipe: {
+        // TODO(zcbenz): After the NetworkService refactor, the old blobUUID API
+        // becomes unecessarily complex, we should deprecate the getBlobData API
+        // and return the DataPipeHolder wrapper directly.
+        auto holder = electron::api::DataPipeHolder::Create(isolate, element);
+        upload_data.Set("blobUUID", holder->id());
+        // The lifetime of data pipe is bound to the uploadData object.
+        upload_data.Set("dataPipe", holder);
         break;
+      }
       default:
         NOTREACHED() << "Found unsupported data element";
     }

spec-main/api-session-spec.ts

@@ -368,7 +368,7 @@ describe('session module', () => {
     })
   })
 
-  describe.skip('ses.getBlobData()', () => {
+  describe('ses.getBlobData()', () => {
     const scheme = 'cors-blob'
     const protocol = session.defaultSession.protocol
     const url = `${scheme}://host`