Implement Readable instead of Stream.

Author: zbjornson

lib/jpegstream.js

@@ -10,7 +10,8 @@
  * Module dependencies.
  */
 
-var Stream = require('stream').Stream;
+var Readable = require('stream').Readable;
+var util = require('util');
 
 /**
  * Initialize a `JPEGStream` with the given `canvas`.
@@ -30,33 +31,43 @@ var Stream = require('stream').Stream;
  */
 
 var JPEGStream = module.exports = function JPEGStream(canvas, options, sync) {
-  var self = this
-    , method = sync
+  if (!(this instanceof JPEGStream)) {
+    throw new TypeError("Class constructors cannot be invoked without 'new'");
+  }
+
+  Readable.call(this);
+
+  var self = this;
+  var method = sync
       ? 'streamJPEGSync'
       : 'streamJPEG';
   this.options = options;
   this.sync = sync;
   this.canvas = canvas;
-  this.readable = true;
+
   // TODO: implement async
   if ('streamJPEG' == method) method = 'streamJPEGSync';
-  process.nextTick(function(){
-    canvas[method](options.bufsize, options.quality, options.progressive, function(err, chunk){
-      if (err) {
-        self.emit('error', err);
-        self.readable = false;
-      } else if (chunk) {
-        self.emit('data', chunk);
-      } else {
-        self.emit('end');
-        self.readable = false;
-      }
-    });
-  });
-};
 
-/**
- * Inherit from `EventEmitter`.
- */
+  // For now we're not controlling the c++ code's data emission, so we only
+  // call canvas.streamPNGSync once and let it emit data at will.
+  var hasStarted = false;
+
+  self._read = function () {
+    if (!hasStarted) {
+      hasStarted = true;
+      process.nextTick(function(){
+        canvas[method](options.bufsize, options.quality, options.progressive, function(err, chunk){
+          if (err) {
+            self.emit('error', err);
+          } else if (chunk) {
+            self.push(chunk);
+          } else {
+            self.push(null);
+          }
+        });
+      });
+    }
+  };
+};
 
-JPEGStream.prototype.__proto__ = Stream.prototype;
+util.inherits(JPEGStream, Readable);

lib/pngstream.js

@@ -10,7 +10,8 @@
  * Module dependencies.
  */
 
-var Stream = require('stream').Stream;
+var Readable = require('stream').Readable;
+var util = require('util');
 
 /**
  * Initialize a `PNGStream` with the given `canvas`.
@@ -30,32 +31,41 @@ var Stream = require('stream').Stream;
  */
 
 var PNGStream = module.exports = function PNGStream(canvas, sync) {
-  var self = this
-    , method = sync
+  if (!(this instanceof PNGStream)) {
+    throw new TypeError("Class constructors cannot be invoked without 'new'");
+  }
+
+  Readable.call(this);
+
+  var self = this;
+  var method = sync
       ? 'streamPNGSync'
       : 'streamPNG';
   this.sync = sync;
   this.canvas = canvas;
-  this.readable = true;
+
   // TODO: implement async
-  if ('streamPNG' == method) method = 'streamPNGSync';
-  process.nextTick(function(){
-    canvas[method](function(err, chunk, len){
-      if (err) {
-        self.emit('error', err);
-        self.readable = false;
-      } else if (len) {
-        self.emit('data', chunk, len);
-      } else {
-        self.emit('end');
-        self.readable = false;
-      }
-    });
-  });
-};
+  if ('streamPNG' === method) method = 'streamPNGSync';
 
-/**
- * Inherit from `EventEmitter`.
- */
+  // For now we're not controlling the c++ code's data emission, so we only
+  // call canvas.streamPNGSync once and let it emit data at will.
+  var hasStarted = false;
+  self._read = function () {
+    if (!hasStarted) {
+      hasStarted = true;
+      process.nextTick(function(){
+        canvas[method](function(err, chunk, len){
+          if (err) {
+            self.emit('error', err);
+          } else if (len) {
+            self.push(chunk);
+          } else {
+            self.push(null);
+          }
+        });
+      });
+    }
+  };
+};
 
-PNGStream.prototype.__proto__ = Stream.prototype;
+util.inherits(PNGStream, Readable);

src/Canvas.cc

@@ -429,7 +429,7 @@ NAN_METHOD(Canvas::StreamPNGSync) {
         Nan::Null()
       , Nan::Null()
       , Nan::New<Uint32>(0) };
-    Nan::MakeCallback(Nan::GetCurrentContext()->Global(), (v8::Local<v8::Function>)closure.fn, 1, argv);
+    Nan::MakeCallback(Nan::GetCurrentContext()->Global(), (v8::Local<v8::Function>)closure.fn, 3, argv);
   }
   return;
 }

test/canvas.test.js

@@ -5,7 +5,8 @@
 var Canvas = require('../')
   , assert = require('assert')
   , parseFont = Canvas.Context2d.parseFont
-  , fs = require('fs');
+  , fs = require('fs')
+  , Readable = require('stream').Readable;
 
 console.log();
 console.log('   canvas: %s', Canvas.version);
@@ -749,6 +750,7 @@ describe('Canvas', function () {
   it('Canvas#createSyncPNGStream()', function (done) {
     var canvas = new Canvas(20, 20);
     var stream = canvas.createSyncPNGStream();
+    assert(stream instanceof Readable);
     var firstChunk = true;
     stream.on('data', function(chunk){
       if (firstChunk) {
@@ -767,6 +769,7 @@ describe('Canvas', function () {
   it('Canvas#jpegStream()', function (done) {
     var canvas = new Canvas(640, 480);
     var stream = canvas.jpegStream();
+    assert(stream instanceof Readable);
     var firstChunk = true;
     var bytes = 0;
     stream.on('data', function(chunk){