Ugly async hack

lib/pdfstream.js

@@ -35,8 +35,6 @@ var PDFStream = module.exports = function PDFStream(canvas, sync) {
   this.sync = sync;
   this.canvas = canvas;
   this.readable = true;
-  // TODO: implement async
-  if ('streamPDF' == method) method = 'streamPDFSync';
   process.nextTick(function(){
     canvas[method](function(err, chunk, len){
       if (err) {

src/Canvas.cc

@@ -40,6 +40,7 @@ Canvas::Initialize(Nan::ADDON_REGISTER_FUNCTION_ARGS_TYPE target) {
   Local<ObjectTemplate> proto = ctor->PrototypeTemplate();
   Nan::SetPrototypeMethod(ctor, "toBuffer", ToBuffer);
   Nan::SetPrototypeMethod(ctor, "streamPNGSync", StreamPNGSync);
+  Nan::SetPrototypeMethod(ctor, "streamPDF", StreamPDF);
   Nan::SetPrototypeMethod(ctor, "streamPDFSync", StreamPDFSync);
 #ifdef HAVE_JPEG
   Nan::SetPrototypeMethod(ctor, "streamJPEGSync", StreamJPEGSync);
@@ -445,8 +446,7 @@ void stream_pdf_free(char *, void *) {}
  * Canvas::StreamPDF callback.
  */
 
-static cairo_status_t
-streamPDF(void *c, const uint8_t *data, unsigned len) {
+cairo_status_t streamPDF(void *c, const uint8_t *data, unsigned len) {
   Nan::HandleScope scope;
   closure_t *closure = static_cast<closure_t *>(c);
   Local<Object> buf = Nan::NewBuffer(const_cast<char *>(reinterpret_cast<const char *>(data)), len, stream_pdf_free, 0).ToLocalChecked();
@@ -459,10 +459,10 @@ streamPDF(void *c, const uint8_t *data, unsigned len) {
 }
 
 
-cairo_status_t canvas_write_to_pdf_stream(cairo_surface_t *surface, cairo_write_func_t write_func, void *closure) {
+cairo_status_t canvas_write_to_pdf_stream(cairo_write_func_t write_func, void *closure) {
   closure_t *pdf_closure = static_cast<closure_t *>(closure);
   size_t whole_chunks = pdf_closure->len / PAGE_SIZE;
-  size_t remainder = pdf_closure->len - whole_chunks * PAGE_SIZE;
+  size_t remainder = pdf_closure->len % PAGE_SIZE;
 
   for (size_t i = 0; i < whole_chunks; ++i) {
     write_func(pdf_closure, &pdf_closure->data[i * PAGE_SIZE], PAGE_SIZE);
@@ -475,6 +475,71 @@ cairo_status_t canvas_write_to_pdf_stream(cairo_surface_t *surface, cairo_write_
   return CAIRO_STATUS_SUCCESS;
 }
 
+class StreamPDFWorker : public Nan::AsyncProgressWorker {
+ public:
+  StreamPDFWorker(Nan::Callback *callback, closure_t *closure) : Nan::AsyncProgressWorker(callback), closure_(closure) {
+    uv_mutex_init(&mutex_);
+  }
+
+  ~StreamPDFWorker() {
+    uv_mutex_destroy(&mutex_);
+  }
+
+  void Execute(const Nan::AsyncProgressWorker::ExecutionProgress& progress) {
+    size_t whole_chunks = closure_->len / PAGE_SIZE;
+    size_t remainder = closure_->len % PAGE_SIZE;
+
+    for (size_t i = 0; i < whole_chunks; ++i) {
+      uv_mutex_lock(&mutex_);
+      progress.Send(reinterpret_cast<char *>(&closure_->data[i * PAGE_SIZE]), PAGE_SIZE);
+    }
+
+    if (remainder) {
+       uv_mutex_lock(&mutex_);
+       progress.Send(reinterpret_cast<char *>(&closure_->data[whole_chunks * PAGE_SIZE]), remainder);
+    }
+
+    uv_mutex_lock(&mutex_);
+    uv_mutex_unlock(&mutex_);
+  }
+
+  void HandleProgressCallback(const char *data, size_t size) {
+    Nan::HandleScope scope;
+
+    v8::Local<v8::Value> argv[] = {
+        Nan::Null(),
+        Nan::NewBuffer(const_cast<char *>(data), size, stream_pdf_free, 0).ToLocalChecked(),
+        Nan::New<Number>(size)
+    };
+
+    callback->Call(sizeof (argv) / sizeof (argv[0]), argv);
+
+    uv_mutex_unlock(&mutex_);
+  }
+
+ private:
+  closure_t *closure_;
+  uv_mutex_t mutex_;
+};
+
+/*
+ * Stream PDF data asynchronously.
+ */
+
+NAN_METHOD(Canvas::StreamPDF) {
+  if (!info[0]->IsFunction())
+    return Nan::ThrowTypeError("callback function required");
+
+  Canvas *canvas = Nan::ObjectWrap::Unwrap<Canvas>(info.Holder());
+
+  if (!canvas->isPDF())
+    return Nan::ThrowTypeError("wrong canvas type");
+
+  cairo_surface_finish(canvas->surface());
+
+  Nan::AsyncQueueWorker(new StreamPDFWorker(new Nan::Callback(info[0].As<Function>()), static_cast<closure_t *>(canvas->closure())));
+}
+
 /*
  * Stream PDF data synchronously.
  */
@@ -497,7 +562,7 @@ NAN_METHOD(Canvas::StreamPDFSync) {
 
   Nan::TryCatch try_catch;
 
-  cairo_status_t status = canvas_write_to_pdf_stream(canvas->surface(), streamPDF, &closure);
+  cairo_status_t status = canvas_write_to_pdf_stream(streamPDF, &closure);
 
   if (try_catch.HasCaught()) {
     try_catch.ReThrow();

src/Canvas.h

@@ -62,6 +62,7 @@ class Canvas: public Nan::ObjectWrap {
     static NAN_SETTER(SetWidth);
     static NAN_SETTER(SetHeight);
     static NAN_METHOD(StreamPNGSync);
+    static NAN_METHOD(StreamPDF);
     static NAN_METHOD(StreamPDFSync);
     static NAN_METHOD(StreamJPEGSync);
     static Local<Value> Error(cairo_status_t status);

test/canvas.test.js

@@ -802,6 +802,24 @@ describe('Canvas', function () {
     });
   });
 
+  it('Canvas#createPDFStream()', function (done) {
+    var canvas = new Canvas(20, 20, 'pdf');
+    var stream = canvas.createPDFStream();
+    var firstChunk = true;
+    stream.on('data', function (chunk) {
+      if (firstChunk) {
+        firstChunk = false;
+        assert.equal('PDF', chunk.slice(1, 4).toString());
+      }
+    });
+    stream.on('end', function () {
+      done();
+    });
+    stream.on('error', function () {
+      done(err);
+    });
+  });
+
   it('Canvas#createSyncPDFStream()', function (done) {
     var canvas = new Canvas(20, 20, 'pdf');
     var stream = canvas.createSyncPDFStream();