Implement Readable instead of Stream.

Automattic · Mar 21, 2016 · dbee175 · dbee175
1 parent a3492e3
commit dbee175
Show file tree

Hide file tree

Showing 5 changed files with 74 additions and 62 deletions.
diff --git a/lib/canvas.js b/lib/canvas.js
@@ -269,15 +269,15 @@ Canvas.prototype.toDataURL = function(a1, a2, a3){
  var stream = this.jpegStream(opts);
  // note that jpegStream is synchronous
  var buffers = [];
- stream.on('data', function (chunk) {
- buffers.push(chunk);
- });
  stream.on('error', function (err) {
  fn(err);
  });
  stream.on('end', function() {
  var result = 'data:image/jpeg;base64,' + Buffer.concat(buffers).toString('base64');
  fn(null, result);
  });
+ stream.on('data', function (chunk) {
+ buffers.push(chunk);
+ });
  }
 };
diff --git a/lib/jpegstream.js b/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,37 @@ var Stream = require('stream').Stream;
  */
 
 var JPEGStream = module.exports = function JPEGStream(canvas, options, sync) {
- var self = this
- , method = sync
+ 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;
+ 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);
diff --git a/lib/pngstream.js b/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,36 @@ var Stream = require('stream').Stream;
  */
 
 var PNGStream = module.exports = function PNGStream(canvas, sync) {
- var self = this
- , method = 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';
- 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;
+ canvas[method](function(err, chunk, len){
+ console.log("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);
diff --git a/src/Canvas.cc b/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;
 }

diff --git a/test/canvas.test.js b/test/canvas.test.js
@@ -749,26 +749,35 @@ describe('Canvas', function () {
  it('Canvas#createSyncPNGStream()', function (done) {
  var canvas = new Canvas(20, 20);
  var stream = canvas.createSyncPNGStream();
+ assert(stream instanceof require('stream').Readable);
  var firstChunk = true;
- stream.on('data', function(chunk){
- if (firstChunk) {
- firstChunk = false;
- assert.equal('PNG', chunk.slice(1,4).toString());
- }
- });
  stream.on('end', function(){
  done();
  });
  stream.on('error', function(err) {
  done(err);
  });
+ stream.on('data', function(chunk){
+ if (firstChunk) {
+ firstChunk = false;
+ assert.equal('PNG', chunk.slice(1,4).toString());
+ }
+ });
  });
 
  it('Canvas#jpegStream()', function (done) {
  var canvas = new Canvas(640, 480);
  var stream = canvas.jpegStream();
+ assert(stream instanceof require('stream').Readable);
  var firstChunk = true;
  var bytes = 0;
+ stream.on('end', function(){
+ assert.equal(bytes, 8192);
+ done();
+ });
+ stream.on('error', function(err) {
+ done(err);
+ });
  stream.on('data', function(chunk){
  if (firstChunk) {
  firstChunk = false;
@@ -778,12 +787,5 @@ describe('Canvas', function () {
  }
  bytes += chunk.length;
  });
- stream.on('end', function(){
- assert.equal(bytes, 8192);
- done();
- });
- stream.on('error', function(err) {
- done(err);
- });
  });
 });