diff --git a/lib/canvas.js b/lib/canvas.js
index afb872ee2..16be48203 100644
--- a/lib/canvas.js
+++ b/lib/canvas.js
@@ -269,9 +269,6 @@ 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);
     });
@@ -279,5 +276,8 @@ Canvas.prototype.toDataURL = function(a1, a2, a3){
       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
index 24ca6f396..6de6610e5 100644
--- 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
index 8a538d03a..19e85f609 100644
--- 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
index 83de51ef1..62895ec78 100644
--- 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
index ad62344a7..7c6716b38 100644
--- 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);
-    });
   });
 });