Merge pull request #740 from zbjornson/stream-polishing

Implement Readable instead of Stream

Author: LinusU

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,47 @@ 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';
+  this.method = method;
+};
+
+util.inherits(JPEGStream, Readable);
+
+function noop() {}
+
+JPEGStream.prototype._read = function _read() {
+  // For now we're not controlling the c++ code's data emission, so we only
+  // call canvas.streamJPEGSync once and let it emit data at will.
+  this._read = noop;
+  var self = this;
+  var method = this.method;
+  var bufsize = this.options.bufsize;
+  var quality = this.options.quality;
+  var progressive = this.options.progressive;
   process.nextTick(function(){
-    canvas[method](options.bufsize, options.quality, options.progressive, function(err, chunk){
+    self.canvas[method](bufsize, quality, progressive, function(err, chunk){
       if (err) {
         self.emit('error', err);
-        self.readable = false;
       } else if (chunk) {
-        self.emit('data', chunk);
+        self.push(chunk);
       } else {
-        self.emit('end');
-        self.readable = false;
+        self.push(null);
       }
     });
   });
 };
-
-/**
- * Inherit from `EventEmitter`.
- */
-
-JPEGStream.prototype.__proto__ = Stream.prototype;

lib/pdfstream.js

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

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,59 @@ 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;
+
+  // TODO: implement async
+  if ('streamPNG' === method) method = 'streamPNGSync';
+  this.method = method;
+};
+
+util.inherits(PNGStream, Readable);
+
+var PNGStream = module.exports = function PNGStream(canvas, sync) {
+  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';
+  if ('streamPNG' === method) method = 'streamPNGSync';
+  this.method = method;
+};
+
+util.inherits(PNGStream, Readable);
+
+function noop() {}
+
+PNGStream.prototype._read = function _read() {
+  // 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.
+  this._read = noop;
+  var self = this;
   process.nextTick(function(){
-    canvas[method](function(err, chunk, len){
+    self.canvas[self.method](function(err, chunk, len){
       if (err) {
         self.emit('error', err);
-        self.readable = false;
       } else if (len) {
-        self.emit('data', chunk, len);
+        self.push(chunk);
       } else {
-        self.emit('end');
-        self.readable = false;
+        self.push(null);
       }
     });
   });
 };
-
-/**
- * Inherit from `EventEmitter`.
- */
-
-PNGStream.prototype.__proto__ = Stream.prototype;

src/Canvas.cc

@@ -456,7 +456,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

@@ -6,7 +6,8 @@ var Canvas = require('../')
   , assert = require('assert')
   , parseFont = Canvas.Context2d.parseFont
   , fs = require('fs')
-  , os = require('os');
+  , os = require('os')
+  , Readable = require('stream').Readable;
 
 console.log();
 console.log('   canvas: %s', Canvas.version);
@@ -886,6 +887,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) {
@@ -904,6 +906,7 @@ describe('Canvas', function () {
   it('Canvas#createSyncPDFStream()', function (done) {
     var canvas = new Canvas(20, 20, 'pdf');
     var stream = canvas.createSyncPDFStream();
+    assert(stream instanceof Readable);
     var firstChunk = true;
     stream.on('data', function (chunk) {
       if (firstChunk) {
@@ -922,6 +925,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){