Stream's api methods must never throw - handling errors in DecodeStream, EncodeStream

lib/decode-stream.js

@@ -26,7 +26,11 @@ function DecodeStream(options) {
 }
 
 DecodeStream.prototype._transform = function(chunk, encoding, callback) {
-  this.decoder.write(chunk);
-  this.decoder.flush();
-  if (callback) callback();
+  try {
+    this.decoder.write(chunk);
+    this.decoder.flush();
+  } catch(e) {
+    return callback(e);
+  }
+  callback();
 };

lib/encode-stream.js

@@ -27,11 +27,19 @@ function EncodeStream(options) {
 }
 
 EncodeStream.prototype._transform = function(chunk, encoding, callback) {
-  this.encoder.write(chunk);
-  if (callback) callback();
+  try {
+    this.encoder.write(chunk);
+  } catch(e) {
+    return callback(e);
+  }
+  callback();
 };
 
 EncodeStream.prototype._flush = function(callback) {
-  this.encoder.flush();
-  if (callback) callback();
+  try {
+    this.encoder.flush();
+  } catch(e) {
+    return callback(e);
+  }
+  callback();
 };

test/30.stream.js

@@ -22,6 +22,9 @@ var encoded = [
 
 var encodeall = Buffer.concat(encoded);
 
+// invalid msgpack type
+var invalencoded = Buffer([0xc1]);
+
 describe(TITLE, function() {
 
   it("msgpack.createEncodeStream()", function(done) {
@@ -116,4 +119,38 @@ describe(TITLE, function() {
       if (++count === 3) done();
     }
   });
+
+  it("msgpack.createDecodeStream().on('error',fn)", function(done) {
+    var decoder = msgpack.createDecodeStream();
+
+    decoder.on("error", function(e) {
+      assert.ok(e instanceof Error, "should be an error");
+      setTimeout(done, 1);
+    });
+
+    decoder.on("data", function(data) {
+      assert.fail("should not emit data");
+    });
+
+    decoder.end(invalencoded);
+  });
+
+  it("msgpack.createEncodeStream().on('error',fn)", function(done) {
+    var circular = [];
+    var encoder = msgpack.createEncodeStream();
+
+    circular.push(circular);
+
+    encoder.on("error", function(e) {
+      assert.ok(e instanceof Error, "should be an error");
+      setTimeout(done, 1);
+    });
+
+    encoder.on("data", function(data) {
+      assert.fail("should not emit data");
+    });
+
+    encoder.end(circular);
+  });
+
 });