stream: simplify Transform

Author: ronag

benchmark/streams/transform-manytransforms.js

@@ -0,0 +1,39 @@
+'use strict';
+
+const common = require('../common');
+const Transform = require('stream').Transform;
+
+const bench = common.createBenchmark(main, {
+  n: [2e6],
+  sync: ['yes', 'no'],
+  callback: ['yes', 'no']
+});
+
+function main({ n, sync, callback }) {
+  const b = Buffer.allocUnsafe(1024);
+  const s = new Transform();
+  sync = sync === 'yes';
+
+  const writecb = (cb) => {
+    if (sync)
+      cb();
+    else
+      process.nextTick(cb);
+  };
+
+  s._transform = (chunk, encoding, cb) => writecb(cb);
+
+  const cb = callback === 'yes' ? () => {} : null;
+
+  bench.start();
+
+  let k = 0;
+  function run() {
+    while (k++ < n && s.write(b, cb));
+    if (k >= n)
+      bench.end(n);
+  }
+  s.on('drain', run);
+  s.on('data', () => {});
+  run();
+}

lib/_stream_transform.js

@@ -65,66 +65,25 @@
 
 const {
   ObjectSetPrototypeOf,
+  Symbol
 } = primordials;
 
 module.exports = Transform;
 const {
-  ERR_METHOD_NOT_IMPLEMENTED,
-  ERR_MULTIPLE_CALLBACK,
-  ERR_TRANSFORM_ALREADY_TRANSFORMING,
-  ERR_TRANSFORM_WITH_LENGTH_0
+  ERR_METHOD_NOT_IMPLEMENTED
 } = require('internal/errors').codes;
 const Duplex = require('_stream_duplex');
 ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
 ObjectSetPrototypeOf(Transform, Duplex);
 
-
-function afterTransform(er, data) {
-  const ts = this._transformState;
-  ts.transforming = false;
-
-  const cb = ts.writecb;
-
-  if (cb === null) {
-    return this.emit('error', new ERR_MULTIPLE_CALLBACK());
-  }
-
-  ts.writechunk = null;
-  ts.writecb = null;
-
-  if (data != null) // Single equals check for both `null` and `undefined`
-    this.push(data);
-
-  cb(er);
-
-  const rs = this._readableState;
-  rs.reading = false;
-  if (rs.needReadable || rs.length < rs.highWaterMark) {
-    this._read(rs.highWaterMark);
-  }
-}
-
+const kResume = Symbol('kResume');
 
 function Transform(options) {
   if (!(this instanceof Transform))
     return new Transform(options);
 
   Duplex.call(this, options);
 
-  this._transformState = {
-    afterTransform: afterTransform.bind(this),
-    needTransform: false,
-    transforming: false,
-    writecb: null,
-    writechunk: null,
-    writeencoding: null
-  };
-
-  // We have implemented the _read method, and done the other things
-  // that Readable wants before the first _read call, so unset the
-  // sync guard flag.
-  this._readableState.sync = false;
-
   if (options) {
     if (typeof options.transform === 'function')
       this._transform = options.transform;
@@ -133,89 +92,53 @@ function Transform(options) {
       this._flush = options.flush;
   }
 
-  // When the writable side finishes, then flush out anything remaining.
-  this.on('prefinish', prefinish);
+  this._readableState.sync = false;
+  this[kResume] = null;
 }
 
-function prefinish() {
-  if (typeof this._flush === 'function' && !this._readableState.destroyed) {
-    this._flush((er, data) => {
-      done(this, er, data);
+Transform.prototype._final = function(cb) {
+  if (typeof this._flush === 'function') {
+    this._flush((err) => {
+      if (err) {
+        cb(err);
+      } else {
+        this.push(null);
+        cb();
+      }
     });
   } else {
-    done(this, null, null);
-  }
-}
-
-Transform.prototype.push = function(chunk, encoding) {
-  this._transformState.needTransform = false;
-  return Duplex.prototype.push.call(this, chunk, encoding);
-};
-
-// This is the part where you do stuff!
-// override this function in implementation classes.
-// 'chunk' is an input chunk.
-//
-// Call `push(newChunk)` to pass along transformed output
-// to the readable side.  You may call 'push' zero or more times.
-//
-// Call `cb(err)` when you are done with this chunk.  If you pass
-// an error, then that'll put the hurt on the whole operation.  If you
-// never call cb(), then you'll never get another chunk.
-Transform.prototype._transform = function(chunk, encoding, cb) {
-  cb(new ERR_METHOD_NOT_IMPLEMENTED('_transform()'));
-};
-
-Transform.prototype._write = function(chunk, encoding, cb) {
-  const ts = this._transformState;
-  ts.writecb = cb;
-  ts.writechunk = chunk;
-  ts.writeencoding = encoding;
-  if (!ts.transforming) {
-    var rs = this._readableState;
-    if (ts.needTransform ||
-        rs.needReadable ||
-        rs.length < rs.highWaterMark)
-      this._read(rs.highWaterMark);
+    this.push(null);
+    cb();
   }
 };
 
-// Doesn't matter what the args are here.
-// _transform does all the work.
-// That we got here means that the readable side wants more data.
 Transform.prototype._read = function(n) {
-  const ts = this._transformState;
-
-  if (ts.writechunk !== null && !ts.transforming) {
-    ts.transforming = true;
-    this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
-  } else {
-    // Mark that we need a transform, so that any data that comes in
-    // will get processed, now that we've asked for it.
-    ts.needTransform = true;
+  if (this[kResume]) {
+    const resume = this[kResume];
+    this[kResume] = null;
+    resume();
   }
 };
 
-
-Transform.prototype._destroy = function(err, cb) {
-  Duplex.prototype._destroy.call(this, err, (err2) => {
-    cb(err2);
+Transform.prototype._write = function(chunk, encoding, callback) {
+  this._transform(chunk, encoding, (err, data) => {
+    if (err) {
+      return callback(err);
+    }
+
+    if (data !== undefined) {
+      this.push(data);
+    }
+
+    const r = this._readableState;
+    if (r.length < r.highWaterMark || r.length === 0) {
+      callback();
+    } else {
+      this[kResume] = callback;
+    }
   });
 };
 
-
-function done(stream, er, data) {
-  if (er)
-    return stream.emit('error', er);
-
-  if (data != null) // Single equals check for both `null` and `undefined`
-    stream.push(data);
-
-  // These two error cases are coherence checks that can likely not be tested.
-  if (stream._writableState.length)
-    throw new ERR_TRANSFORM_WITH_LENGTH_0();
-
-  if (stream._transformState.transforming)
-    throw new ERR_TRANSFORM_ALREADY_TRANSFORMING();
-  return stream.push(null);
-}
+Transform.prototype._transform = function(chunk, encoding, cb) {
+  cb(new ERR_METHOD_NOT_IMPLEMENTED('_transform()'));
+};

lib/internal/errors.js

@@ -1297,12 +1297,7 @@ E('ERR_TLS_SNI_FROM_SERVER',
 E('ERR_TRACE_EVENTS_CATEGORY_REQUIRED',
   'At least one category is required', TypeError);
 E('ERR_TRACE_EVENTS_UNAVAILABLE', 'Trace events are unavailable', Error);
-E('ERR_TRANSFORM_ALREADY_TRANSFORMING',
-  'Calling transform done when still transforming', Error);
 
-// This should probably be a `RangeError`.
-E('ERR_TRANSFORM_WITH_LENGTH_0',
-  'Calling transform done when writableState.length != 0', Error);
 E('ERR_TTY_INIT_FAILED', 'TTY initialization failed', SystemError);
 E('ERR_UNCAUGHT_EXCEPTION_CAPTURE_ALREADY_SET',
   '`process.setupUncaughtExceptionCapture()` was called while a capture ' +

test/parallel/test-stream-transform-constructor-set-methods.js

@@ -18,23 +18,17 @@ const _transform = common.mustCall((chunk, _, next) => {
   next();
 });
 
-const _final = common.mustCall((next) => {
-  next();
-});
-
 const _flush = common.mustCall((next) => {
   next();
 });
 
 const t2 = new Transform({
   transform: _transform,
-  flush: _flush,
-  final: _final
+  flush: _flush
 });
 
 strictEqual(t2._transform, _transform);
 strictEqual(t2._flush, _flush);
-strictEqual(t2._final, _final);
 
 t2.end(Buffer.from('blerg'));
 t2.resume();

test/parallel/test-stream-transform-final-sync.js

@@ -66,36 +66,27 @@ const t = new stream.Transform({
     assert.strictEqual(++state, chunk + 2);
     process.nextTick(next);
   }, 3),
-  final: common.mustCall(function(done) {
-    state++;
-    // finalCallback part 1
-    assert.strictEqual(state, 10);
-    state++;
-    // finalCallback part 2
-    assert.strictEqual(state, 11);
-    done();
-  }, 1),
   flush: common.mustCall(function(done) {
     state++;
     // fluchCallback part 1
-    assert.strictEqual(state, 12);
+    assert.strictEqual(state, 10);
     process.nextTick(function() {
       state++;
       // fluchCallback part 2
-      assert.strictEqual(state, 13);
+      assert.strictEqual(state, 11);
       done();
     });
   }, 1)
 });
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 12);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
   // endEvent
-  assert.strictEqual(state, 16);
+  assert.strictEqual(state, 14);
 }, 1));
 t.on('data', common.mustCall(function(d) {
   // dataListener
@@ -106,5 +97,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 15);
+  assert.strictEqual(state, 13);
 }, 1));

test/parallel/test-stream-transform-final.js

@@ -66,38 +66,27 @@ const t = new stream.Transform({
     assert.strictEqual(++state, chunk + 2);
     process.nextTick(next);
   }, 3),
-  final: common.mustCall(function(done) {
-    state++;
-    // finalCallback part 1
-    assert.strictEqual(state, 10);
-    setTimeout(function() {
-      state++;
-      // finalCallback part 2
-      assert.strictEqual(state, 11);
-      done();
-    }, 100);
-  }, 1),
   flush: common.mustCall(function(done) {
     state++;
     // flushCallback part 1
-    assert.strictEqual(state, 12);
+    assert.strictEqual(state, 10);
     process.nextTick(function() {
       state++;
       // flushCallback part 2
-      assert.strictEqual(state, 15);
+      assert.strictEqual(state, 11);
       done();
     });
   }, 1)
 });
 t.on('finish', common.mustCall(function() {
   state++;
   // finishListener
-  assert.strictEqual(state, 13);
+  assert.strictEqual(state, 12);
 }, 1));
 t.on('end', common.mustCall(function() {
   state++;
   // end event
-  assert.strictEqual(state, 16);
+  assert.strictEqual(state, 14);
 }, 1));
 t.on('data', common.mustCall(function(d) {
   // dataListener
@@ -108,5 +97,5 @@ t.write(4);
 t.end(7, common.mustCall(function() {
   state++;
   // endMethodCallback
-  assert.strictEqual(state, 14);
+  assert.strictEqual(state, 13);
 }, 1));

test/parallel/test-stream2-transform.js

@@ -45,7 +45,6 @@ const Transform = require('_stream_transform');
 
   assert.strictEqual(tx.readableLength, 10);
   assert.strictEqual(transformed, 10);
-  assert.strictEqual(tx._transformState.writechunk.length, 5);
   assert.deepStrictEqual(tx.writableBuffer.map(function(c) {
     return c.chunk.length;
   }), [6, 7, 8, 9, 10]);

test/parallel/test-zlib-flush-drain.js

@@ -36,6 +36,11 @@ deflater.on('drain', function() {
   drainCount++;
 });
 
+deflater.on('data', function() {
+  // Keep reading data or flush will
+  // wait until buffer is emptied.
+});
+
 process.once('exit', function() {
   assert.strictEqual(
     beforeFlush, true);