[core] [node] pass thread pool impl to Map constructor

Updates mbgl::Map constructor usage everywhere

Adds NodeThreadPool implementation using AsyncQueue to call
Nan::AsyncQueueWorker from main thread

7 files changed, 86 insertions, 8 deletions

diff --git a/platform/node/src/node_log.cpp b/platform/node/src/node_log.cpp
index 5ea0bcc06a..0a97ebce36 100644
--- a/platform/node/src/node_log.cpp
+++ b/platform/node/src/node_log.cpp
@@ -1,5 +1,4 @@
 #include "node_log.hpp"
-#include "util/async_queue.hpp"
 
 #include <mbgl/util/enum.hpp>
 
@@ -19,7 +18,7 @@ struct NodeLogObserver::LogMessage {
 };
 
 NodeLogObserver::NodeLogObserver(v8::Local<v8::Object> target)
-    : queue(new Queue(uv_default_loop(), [this](LogMessage &message) {
+    : queue(new util::AsyncQueue<LogMessage>(uv_default_loop(), [this](LogMessage &message) {
           Nan::HandleScope scope;
 
           auto msg = Nan::New<v8::Object>();
diff --git a/platform/node/src/node_log.hpp b/platform/node/src/node_log.hpp
index d29e4e28e0..a19c61284b 100644
--- a/platform/node/src/node_log.hpp
+++ b/platform/node/src/node_log.hpp
@@ -1,5 +1,7 @@
 #pragma once
 
+#include "util/async_queue.hpp"
+
 #include <mbgl/platform/log.hpp>
 
 #pragma GCC diagnostic push
@@ -24,8 +26,7 @@ private:
     Nan::Persistent<v8::Object> module;
 
     struct LogMessage;
-    using Queue = util::AsyncQueue<LogMessage>;
-    Queue *queue = nullptr;
+    util::AsyncQueue<LogMessage>* queue;
 };
 
 }
diff --git a/platform/node/src/node_map.cpp b/platform/node/src/node_map.cpp
index 4d4be5be66..a13adbc417 100644
--- a/platform/node/src/node_map.cpp
+++ b/platform/node/src/node_map.cpp
@@ -777,7 +777,8 @@ NodeMap::NodeMap(v8::Local<v8::Object> options) :
         Nan::HandleScope scope;
         return Nan::Has(options, Nan::New("ratio").ToLocalChecked()).FromJust() ? Nan::Get(options, Nan::New("ratio").ToLocalChecked()).ToLocalChecked()->NumberValue() : 1.0;
     }()),
-    map(std::make_unique<mbgl::Map>(view, *this, mbgl::MapMode::Still)),
+    threadpool(),
+    map(std::make_unique<mbgl::Map>(view, *this, threadpool, mbgl::MapMode::Still)),
     async(new uv_async_t) {
 
     view.setMapChangeCallback([&](mbgl::MapChange change) {
diff --git a/platform/node/src/node_map.hpp b/platform/node/src/node_map.hpp
index c0b025c369..cdca1587ea 100644
--- a/platform/node/src/node_map.hpp
+++ b/platform/node/src/node_map.hpp
@@ -1,5 +1,7 @@
 #pragma once
 
+#include "node_thread_pool.hpp"
+
 #include <mbgl/map/map.hpp>
 #include <mbgl/storage/file_source.hpp>
 #include <mbgl/platform/default/headless_view.hpp>
@@ -52,6 +54,7 @@ public:
     std::unique_ptr<mbgl::AsyncRequest> request(const mbgl::Resource&, mbgl::FileSource::Callback);
 
     mbgl::HeadlessView view;
+    NodeThreadPool threadpool;
     std::unique_ptr<mbgl::Map> map;
 
     std::exception_ptr error;
diff --git a/platform/node/src/node_thread_pool.cpp b/platform/node/src/node_thread_pool.cpp
new file mode 100644
index 0000000000..a9faef6f09
--- /dev/null
+++ b/platform/node/src/node_thread_pool.cpp
@@ -0,0 +1,39 @@
+#include "node_thread_pool.hpp"
+#include "util/async_queue.hpp"
+
+#include <mbgl/actor/mailbox.hpp>
+
+namespace node_mbgl {
+
+NodeThreadPool::NodeThreadPool()
+    : queue(new util::AsyncQueue<std::weak_ptr<mbgl::Mailbox>>(uv_default_loop(), [this](std::weak_ptr<mbgl::Mailbox> mailbox) {
+        Worker* worker = new Worker(mailbox);
+        Nan::AsyncQueueWorker(worker);
+    })) {
+    // Don't keep the event loop alive.
+    queue->unref();
+}
+
+NodeThreadPool::~NodeThreadPool() {
+    queue->stop();
+}
+
+void NodeThreadPool::schedule(std::weak_ptr<mbgl::Mailbox> mailbox) {
+    queue->send(std::move(mailbox));
+}
+
+NodeThreadPool::Worker::Worker(std::weak_ptr<mbgl::Mailbox> mailbox_)
+    : AsyncWorker(nullptr),
+    mailbox(std::move(mailbox_)) {};
+
+void NodeThreadPool::Worker::Execute() {
+    if (auto locked = mailbox.lock()) {
+        locked->receive();
+    }
+}
+
+void NodeThreadPool::Worker::WorkComplete() {
+    // no-op to avoid calling nullptr callback
+}
+
+} // namespace node_mbgl
diff --git a/platform/node/src/node_thread_pool.hpp b/platform/node/src/node_thread_pool.hpp
new file mode 100644
index 0000000000..d412e53d3d
--- /dev/null
+++ b/platform/node/src/node_thread_pool.hpp
@@ -0,0 +1,37 @@
+#pragma once
+
+#include <mbgl/actor/scheduler.hpp>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#pragma GCC diagnostic ignored "-Wshadow"
+#include <nan.h>
+#pragma GCC diagnostic pop
+
+namespace node_mbgl {
+
+namespace util { template <typename T> class AsyncQueue; }
+
+class NodeThreadPool : public mbgl::Scheduler {
+public:
+    NodeThreadPool();
+    ~NodeThreadPool();
+
+    void schedule(std::weak_ptr<mbgl::Mailbox>) override;
+
+private:
+    util::AsyncQueue<std::weak_ptr<mbgl::Mailbox>>* queue;
+
+    class Worker : public Nan::AsyncWorker {
+    public:
+        Worker(std::weak_ptr<mbgl::Mailbox>);
+
+        void Execute();
+        void WorkComplete();
+
+    private:
+        std::weak_ptr<mbgl::Mailbox> mailbox;
+    };
+};
+
+} // namespace node_mbgl
diff --git a/platform/node/src/util/async_queue.hpp b/platform/node/src/util/async_queue.hpp
index b9081b3aeb..87737437c3 100644
--- a/platform/node/src/util/async_queue.hpp
+++ b/platform/node/src/util/async_queue.hpp
@@ -28,6 +28,7 @@ public:
             q->process();
         });
     }
+    ~AsyncQueue() {}
 
     void send(T &&data) {
         {
@@ -60,9 +61,6 @@ public:
     }
 
 private:
-    ~AsyncQueue() {
-    }
-
     void process() {
         std::unique_ptr<T> item;
         while (true) {