Add encoder stream event for type errors

mtth · Dec 13, 2019 · e1e1a56 · e1e1a56
1 parent 30fb2d8
commit e1e1a56
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 4 deletions.
diff --git a/lib/containers.js b/lib/containers.js
@@ -327,10 +327,12 @@ function RawEncoder(schema, opts) {
     try {
       this._type._write(tap, val);
     } catch (err) {
-      this.emit('error', err);
+      this.emit('typeError', err, val, this._type);
     }
   };
   this._tap = new Tap(utils.newBuffer(opts.batchSize || 65536));
+
+  this.on('typeError', function (err) { this.emit('error', err); });
 }
 util.inherits(RawEncoder, stream.Transform);
 
@@ -404,7 +406,7 @@ function BlockEncoder(schema, opts) {
     try {
       this._type._write(tap, val);
     } catch (err) {
-      this.emit('error', err);
+      this.emit('typeError', err, val, this._type);
       return false;
     }
     return true;
@@ -457,6 +459,8 @@ function BlockEncoder(schema, opts) {
       this.push(null);
     }
   });
+
+  this.on('typeError', function (err) { this.emit('error', err); });
 }
 util.inherits(BlockEncoder, stream.Duplex);
 
@@ -494,7 +498,6 @@ BlockEncoder.prototype._write = function (val, encoding, cb) {
   var flushing = false;
 
   if (this._writeValue(tap, val)) {
-    // The value was valid
     if (!tap.isValid()) {
       if (pos) {
         this._flushChunk(pos, cb);

diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "avsc",
-  "version": "5.4.17",
+  "version": "5.4.18",
   "description": "Avro for JavaScript",
   "homepage": "https://github.com/mtth/avsc",
   "keywords": [

diff --git a/test/test_containers.js b/test/test_containers.js
@@ -711,6 +711,27 @@ suite('containers', function () {
       encoder.end();
     });
 
+    test('custom type error handler', function (cb) {
+      var okVals = [];
+      var badVals = [];
+      var encoder = new streams.BlockEncoder('int')
+        .removeAllListeners('typeError')
+        .on('typeError', function (err, val) { badVals.push(val); });
+      var decoder = new streams.BlockDecoder()
+        .on('data', function (val) { okVals.push(val); })
+        .on('end', function () {
+          assert.deepEqual(okVals, [1, 2]);
+          assert.deepEqual(badVals, ['foo', 5.4]);
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.write('foo');
+      encoder.write(1);
+      encoder.write(2);
+      encoder.write(5.4);
+      encoder.end();
+    });
+
     test('metadata', function (cb) {
       var t = Type.forSchema('string');
       var buf = t.toBuffer('hello');