feat: migrate protocol module to NetworkService (Part 4) (#18084)

* Parse stream protocol handler

* Pipe node stream to mojo

* Merge the parser for headers

* Add ToDict helper to simplify code

* Simplify dispatching logic

* Add an experimental API for returning any type of response

* Fix subscribing event

* URL loaders' lifetime is independent of the factory

* HandleError helper is no longer needed

* Rename "SendResponse" => "StartLoading" to follow naming conventions

* Delete when connection error happens

* Fix cpplint warning

atom/browser/net/node_stream_loader.cc

@@ -0,0 +1,119 @@
+// 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 "atom/browser/net/node_stream_loader.h"
+
+#include <utility>
+
+#include "atom/common/api/event_emitter_caller.h"
+#include "atom/common/native_mate_converters/callback.h"
+
+#include "atom/common/node_includes.h"
+
+namespace atom {
+
+NodeStreamLoader::NodeStreamLoader(network::ResourceResponseHead head,
+                                   network::mojom::URLLoaderRequest loader,
+                                   network::mojom::URLLoaderClientPtr client,
+                                   v8::Isolate* isolate,
+                                   v8::Local<v8::Object> emitter)
+    : binding_(this),
+      client_(std::move(client)),
+      isolate_(isolate),
+      emitter_(isolate, emitter),
+      weak_factory_(this) {
+  auto weak = weak_factory_.GetWeakPtr();
+  binding_.Bind(std::move(loader));
+  binding_.set_connection_error_handler(
+      base::BindOnce(&NodeStreamLoader::OnConnectionError, weak));
+
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  MojoResult rv = mojo::CreateDataPipe(nullptr, &producer_, &consumer);
+  if (rv != MOJO_RESULT_OK) {
+    OnError(nullptr);
+    return;
+  }
+
+  client_->OnReceiveResponse(head);
+  client_->OnStartLoadingResponseBody(std::move(consumer));
+
+  On("end", base::BindRepeating(&NodeStreamLoader::OnEnd, weak));
+  On("error", base::BindRepeating(&NodeStreamLoader::OnError, weak));
+  // Since every node::MakeCallback call has a micro scope itself, we have to
+  // subscribe |data| at last otherwise |end|'s listener won't be called when
+  // it is emitted in the same tick.
+  On("data", base::BindRepeating(&NodeStreamLoader::OnData, weak));
+}
+
+NodeStreamLoader::~NodeStreamLoader() {
+  v8::Locker locker(isolate_);
+  v8::Isolate::Scope isolate_scope(isolate_);
+  v8::HandleScope handle_scope(isolate_);
+
+  // Unsubscribe all handlers.
+  for (const auto& it : handlers_) {
+    v8::Local<v8::Value> args[] = {mate::StringToV8(isolate_, it.first),
+                                   it.second.Get(isolate_)};
+    node::MakeCallback(isolate_, emitter_.Get(isolate_), "removeListener",
+                       node::arraysize(args), args, {0, 0});
+  }
+}
+
+void NodeStreamLoader::On(const char* event, EventCallback callback) {
+  v8::Locker locker(isolate_);
+  v8::Isolate::Scope isolate_scope(isolate_);
+  v8::HandleScope handle_scope(isolate_);
+
+  // emitter.on(event, callback)
+  v8::Local<v8::Value> args[] = {
+      mate::StringToV8(isolate_, event),
+      mate::CallbackToV8(isolate_, std::move(callback)),
+  };
+  node::MakeCallback(isolate_, emitter_.Get(isolate_), "on",
+                     node::arraysize(args), args, {0, 0});
+
+  handlers_[event].Reset(isolate_, args[1]);
+}
+
+void NodeStreamLoader::OnData(mate::Arguments* args) {
+  v8::Local<v8::Value> buffer;
+  args->GetNext(&buffer);
+  if (!node::Buffer::HasInstance(buffer)) {
+    args->ThrowError("data must be Buffer");
+    return;
+  }
+
+  size_t ssize = node::Buffer::Length(buffer);
+  uint32_t size = base::saturated_cast<uint32_t>(ssize);
+  MojoResult result = producer_->WriteData(node::Buffer::Data(buffer), &size,
+                                           MOJO_WRITE_DATA_FLAG_NONE);
+  if (result != MOJO_RESULT_OK || size < ssize) {
+    OnError(nullptr);
+    return;
+  }
+}
+
+void NodeStreamLoader::OnEnd(mate::Arguments* args) {
+  client_->OnComplete(network::URLLoaderCompletionStatus(net::OK));
+  client_.reset();
+  MaybeDeleteSelf();
+}
+
+void NodeStreamLoader::OnError(mate::Arguments* args) {
+  client_->OnComplete(network::URLLoaderCompletionStatus(net::ERR_FAILED));
+  client_.reset();
+  MaybeDeleteSelf();
+}
+
+void NodeStreamLoader::OnConnectionError() {
+  binding_.Close();
+  MaybeDeleteSelf();
+}
+
+void NodeStreamLoader::MaybeDeleteSelf() {
+  if (!binding_.is_bound() && !client_.is_bound())
+    delete this;
+}
+
+}  // namespace atom