AsyncProgressWorker: add template types for .send

This allows for arbitrary types to be sent from .send. To achieve this, the
abstract base class can be implemented with a template itself. Either at
implementation or instantiation a type MUST be declared though.

Makefile

@@ -35,6 +35,7 @@ LINT_SOURCES = \
 	test/cpp/accessors2.cpp \
 	test/cpp/asyncworker.cpp \
 	test/cpp/asyncprogressworker.cpp \
+	test/cpp/asyncprogressworkerstream.cpp \
 	test/cpp/asyncprogressworkersignal.cpp \
 	test/cpp/asyncworkererror.cpp \
 	test/cpp/buffer.cpp \

README.md

@@ -244,7 +244,7 @@ NAN's `node::Buffer` helpers exist as the API has changed across supported Node
 `Nan::AsyncWorker` and `Nan::AsyncProgressWorker` are helper classes that make working with asynchronous code easier.
 
  - <a href="doc/asyncworker.md#api_nan_async_worker"><b><code>Nan::AsyncWorker</code></b></a>
- - <a href="doc/asyncworker.md#api_nan_async_progress_worker"><b><code>Nan::AsyncProgressWorker</code></b></a>
+ - <a href="doc/asyncworker.md#api_nan_async_progress_worker"><b><code>Nan::AsyncProgressWorkerBase & Nan::AsyncProgressWorker</code></b></a>
  - <a href="doc/asyncworker.md#api_nan_async_queue_worker"><b><code>Nan::AsyncQueueWorker</code></b></a>
 
 ### Strings & Bytes

doc/asyncworker.md

@@ -58,14 +58,17 @@ class AsyncWorker {
 ```
 
 <a name="api_nan_async_progress_worker"></a>
-### Nan::AsyncProgressWorker
+### Nan::AsyncProgressWorkerBase & Nan::AsyncProgressWorker
 
-`Nan::AsyncProgressWorker` is an _abstract_ class that extends `Nan::AsyncWorker` and adds additional progress reporting callbacks that can be used during the asynchronous work execution to provide progress data back to JavaScript.
+`Nan::AsyncProgressWorkerBase` is an _abstract_ class template that extends `Nan::AsyncWorker` and adds additional progress reporting callbacks that can be used during the asynchronous work execution to provide progress data back to JavaScript.
+
+Previously the definiton of `Nan::AsyncProgressWorker` only allowed sending `const char` data. Now extending `Nan::AsyncProgressWorker` will yield an instance of the implicit `Nan::AsyncProgressWorkerBase` template with type `<char>` for compatibility.
 
 Definition:
 
 ```c++
-class AsyncProgressWorker : public AsyncWorker {
+template<class T>
+class AsyncProgressWorkerBase<T> : public AsyncWorker {
  public:
   explicit AsyncProgressWorker(Callback *callback_);
 
@@ -76,14 +79,16 @@ class AsyncProgressWorker : public AsyncWorker {
   class ExecutionProgress {
    public:
     void Signal() const;
-    void Send(const char* data, size_t size) const;
+    void Send(const T* data, size_t size) const;
   };
 
   virtual void Execute(const ExecutionProgress& progress) = 0;
 
-  virtual void HandleProgressCallback(const char *data, size_t size) = 0;
+  virtual void HandleProgressCallback(const T *data, size_t size) = 0;
 
   virtual void Destroy();
+
+typedef AsyncProgressWorkerBase<T> AsyncProgressWorker;
 ```
 
 <a name="api_nan_async_queue_worker"></a>

nan.h

@@ -1578,9 +1578,11 @@ class Callback {
   char *errmsg_;
 };
 
-/* abstract */ class AsyncProgressWorker : public AsyncWorker {
+
+template<class T>
+/* abstract */ class AsyncProgressWorkerBase : public AsyncWorker {
  public:
-  explicit AsyncProgressWorker(Callback *callback_)
+  explicit AsyncProgressWorkerBase(Callback *callback_)
       : AsyncWorker(callback_), asyncdata_(NULL), asyncsize_(0) {
     async = new uv_async_t;
     uv_async_init(
@@ -1593,45 +1595,45 @@ class Callback {
     uv_mutex_init(&async_lock);
   }
 
-  virtual ~AsyncProgressWorker() {
+  virtual ~AsyncProgressWorkerBase() {
     uv_mutex_destroy(&async_lock);
 
     delete[] asyncdata_;
   }
 
   void WorkProgress() {
     uv_mutex_lock(&async_lock);
-    char *data = asyncdata_;
+    T *data = asyncdata_;
     size_t size = asyncsize_;
     asyncdata_ = NULL;
     uv_mutex_unlock(&async_lock);
 
-    // Dont send progress events after we've already completed.
+    // Don't send progress events after we've already completed.
     if (callback) {
         HandleProgressCallback(data, size);
     }
     delete[] data;
   }
 
   class ExecutionProgress {
-    friend class AsyncProgressWorker;
+    friend class AsyncProgressWorkerBase;
    public:
     void Signal() const {
         uv_async_send(that_->async);
     }
-    // You could do fancy generics with templates here.
-    void Send(const char* data, size_t size) const {
+
+    void Send(const T* data, size_t size) const {
         that_->SendProgress_(data, size);
     }
 
    private:
-    explicit ExecutionProgress(AsyncProgressWorker* that) : that_(that) {}
+    explicit ExecutionProgress(AsyncProgressWorkerBase *that) : that_(that) {}
     NAN_DISALLOW_ASSIGN_COPY_MOVE(ExecutionProgress)
-    AsyncProgressWorker* const that_;
+    AsyncProgressWorkerBase* const that_;
   };
 
   virtual void Execute(const ExecutionProgress& progress) = 0;
-  virtual void HandleProgressCallback(const char *data, size_t size) = 0;
+  virtual void HandleProgressCallback(const T *data, size_t size) = 0;
 
   virtual void Destroy() {
       uv_close(reinterpret_cast<uv_handle_t*>(async), AsyncClose_);
@@ -1643,12 +1645,15 @@ class Callback {
       Execute(progress);
   }
 
-  void SendProgress_(const char *data, size_t size) {
-    char *new_data = new char[size];
-    memcpy(new_data, data, size);
+  void SendProgress_(const T *data, size_t size) {
+    T *new_data = new T[size];
+    {
+      T *it = new_data;
+      std::copy(data, data + size, it);
+    }
 
     uv_mutex_lock(&async_lock);
-    char *old_data = asyncdata_;
+    T *old_data = asyncdata_;
     asyncdata_ = new_data;
     asyncsize_ = size;
     uv_mutex_unlock(&async_lock);
@@ -1658,24 +1663,28 @@ class Callback {
   }
 
   inline static NAUV_WORK_CB(AsyncProgress_) {
-    AsyncProgressWorker *worker =
-            static_cast<AsyncProgressWorker*>(async->data);
+    AsyncProgressWorkerBase *worker =
+            static_cast<AsyncProgressWorkerBase*>(async->data);
     worker->WorkProgress();
   }
 
   inline static void AsyncClose_(uv_handle_t* handle) {
-    AsyncProgressWorker *worker =
-            static_cast<AsyncProgressWorker*>(handle->data);
+    AsyncProgressWorkerBase *worker =
+            static_cast<AsyncProgressWorkerBase*>(handle->data);
     delete reinterpret_cast<uv_async_t*>(handle);
     delete worker;
   }
 
   uv_async_t *async;
   uv_mutex_t async_lock;
-  char *asyncdata_;
+  T *asyncdata_;
   size_t asyncsize_;
 };
 
+// This ensures compatibility to the previous un-templated AsyncProgressWorker
+// class definition.
+typedef AsyncProgressWorkerBase<char> AsyncProgressWorker;
+
 inline void AsyncExecute (uv_work_t* req) {
   AsyncWorker *worker = static_cast<AsyncWorker*>(req->data);
   worker->Execute();

package.json

@@ -26,6 +26,7 @@
     "commander": "^2.8.1",
     "glob": "^5.0.14",
     "node-gyp": "~3.0.1",
+    "readable-stream": "^2.1.4",
     "tap": "~0.7.1",
     "xtend": "~4.0.0"
   },

test/binding.gyp

@@ -97,6 +97,10 @@
         "target_name" : "asyncprogressworker"
       , "sources"     : [ "cpp/asyncprogressworker.cpp" ]
     }
+    , {
+        "target_name" : "asyncprogressworkerstream"
+      , "sources"     : [ "cpp/asyncprogressworkerstream.cpp" ]
+    }
     , {
         "target_name" : "asyncprogressworkersignal"
       , "sources"     : ["cpp/asyncprogressworkersignal.cpp"]

test/cpp/asyncprogressworkerstream.cpp

@@ -0,0 +1,86 @@
+/*********************************************************************
+ * NAN - Native Abstractions for Node.js
+ *
+ * Copyright (c) 2016 NAN contributors
+ *
+ * MIT License <https://github.com/nodejs/nan/blob/master/LICENSE.md>
+ ********************************************************************/
+
+#ifndef _WIN32
+#include <unistd.h>
+#define Sleep(x) usleep((x)*1000)
+#endif
+#include <nan.h>
+
+using namespace Nan;  // NOLINT(build/namespaces)
+
+// Custom data type: This serves as an example of how external
+// libraries could be hooked in, populate their objects and send them to JS.
+struct data_t {
+  int index;
+  int data;
+};
+
+// Unlike test/cpp/ayncprogressworker.cpp this test is explicitly templated.
+template<typename T>
+class ProgressWorker : public AsyncProgressWorkerBase<T> {
+ public:
+  ProgressWorker(
+      Callback *callback
+    , Callback *progress
+    , int milliseconds
+    , int iters)
+    : AsyncProgressWorkerBase<T>(callback), progress(progress)
+    , milliseconds(milliseconds), iters(iters) {}
+  ~ProgressWorker() {}
+
+  void Execute (
+    const typename AsyncProgressWorkerBase<T>::ExecutionProgress& progress) {
+    data_t data;
+    for (int i = 0; i < iters; ++i) {
+      data.index = i;
+      data.data = i * 2;
+      progress.Send(&data, sizeof( data ));
+      Sleep(milliseconds);
+    }
+  }
+
+  void HandleProgressCallback(const T *data, size_t size) {
+    HandleScope scope;
+    v8::Local<v8::Object> obj = Nan::New<v8::Object>();
+    Nan::Set(
+      obj,
+      Nan::New("index").ToLocalChecked(),
+      New<v8::Integer>(data->index));
+    Nan::Set(
+      obj,
+      Nan::New("data").ToLocalChecked(),
+      New<v8::Integer>(data->data));
+
+    v8::Local<v8::Value> argv[] = { obj };
+    progress->Call(1, argv);
+  }
+
+ private:
+  Callback *progress;
+  int milliseconds;
+  int iters;
+};
+
+NAN_METHOD(DoProgress) {
+  Callback *progress = new Callback(info[2].As<v8::Function>());
+  Callback *callback = new Callback(info[3].As<v8::Function>());
+  AsyncQueueWorker(new ProgressWorker<data_t>(
+      callback
+    , progress
+    , To<uint32_t>(info[0]).FromJust()
+    , To<uint32_t>(info[1]).FromJust()));
+}
+
+NAN_MODULE_INIT(Init) {
+  Set(target
+    , New<v8::String>("a").ToLocalChecked()
+    , New<v8::FunctionTemplate>(DoProgress)->GetFunction());
+}
+
+NODE_MODULE(asyncprogressworkerstream, Init)

test/js/asyncprogressworkerstream-test.js

@@ -0,0 +1,54 @@
+/*********************************************************************
+ * NAN - Native Abstractions for Node.js
+ *
+ * Copyright (c) 2016 NAN contributors
+ *
+ * MIT License <https://github.com/nodejs/nan/blob/master/LICENSE.md>
+ ********************************************************************/
+
+const test     = require('tap').test
+    , testRoot = require('path').resolve(__dirname, '..')
+    , bindings = require('bindings')({ module_root: testRoot, bindings: 'asyncprogressworkerstream' })
+    , util = require('util');
+
+const nodeVersion = process.versions.node.split('.')
+var Readable
+if (nodeVersion[0] == 0 && nodeVersion[1] <= 8)
+  Readable = require('readable-stream')
+else
+  Readable = require('stream').Readable
+
+function StreamProgressWorker(t) {
+  Readable.call(this, {objectMode: true})
+  var self = this
+  // initialize stream from cpp on next tick
+  process.nextTick(function () {
+    var worker = bindings.a
+    worker(100, 5, function(i) {
+      self.push(i)
+    }, function () {
+      self.push(null)
+    })
+  })
+}
+util.inherits(StreamProgressWorker, Readable)
+
+StreamProgressWorker.prototype._read = function (data) {
+
+};
+
+
+test('asyncprogressworker', function (t) {
+  var stream = new StreamProgressWorker(t)
+  var progressed = 0;
+
+  stream
+    .on('end', function() {
+      t.ok(progressed === 5, 'cpp should have sent 5 objects')
+      t.end()
+    })
+    .on('data', function(data) {
+      progressed++
+      console.log(data);
+    })
+})