stream: simplify Transform stream implementation

Significantly simplified Transform stream implementation by using mostly standard stream code. PR-URL: #32763 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Anna Henningsen <[email protected]>

Author: ronag

lib/_stream_transform.js

@@ -65,66 +65,32 @@
 
 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;
 constDuplex=require('_stream_duplex');
 ObjectSetPrototypeOf(Transform.prototype,Duplex.prototype);
 ObjectSetPrototypeOf(Transform,Duplex);
 
-
-functionafterTransform(er,data){
-constts=this._transformState;
-ts.transforming=false;
-
-constcb=ts.writecb;
-
-if(cb===null){
-returnthis.emit('error',newERR_MULTIPLE_CALLBACK());
-}
-
-ts.writechunk=null;
-ts.writecb=null;
-
-if(data!=null)// Single equals check for both `null` and `undefined`
-this.push(data);
-
-cb(er);
-
-constrs=this._readableState;
-rs.reading=false;
-if(rs.needReadable||rs.length<rs.highWaterMark){
-this._read(rs.highWaterMark);
-}
-}
-
+constkCallback=Symbol('kCallback');
 
 functionTransform(options){
 if(!(thisinstanceofTransform))
 returnnewTransform(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;
 
+this[kCallback]=null;
+
 if(options){
 if(typeofoptions.transform==='function')
 this._transform=options.transform;
@@ -133,89 +99,67 @@ function Transform(options){
 this._flush=options.flush;
 }
 
-// When the writable side finishes, then flush out anything remaining.
+// TODO(ronag): Unfortunately _final is invoked asynchronously.
+// Use `prefinish` hack. `prefinish` is emitted synchronously when
+// and only when `_final` is not defined. Implementing `_final`
+// to a Transform should be an error.
 this.on('prefinish',prefinish);
 }
 
 functionprefinish(){
-if(typeofthis._flush==='function'&&!this._readableState.destroyed){
+if(typeofthis._flush==='function'&&!this.destroyed){
 this._flush((er,data)=>{
-done(this,er,data);
+if(er){
+this.destroy(er);
+return;
+}
+
+if(data!=null){
+this.push(data);
+}
+this.push(null);
 });
 }else{
-done(this,null,null);
+this.push(null);
 }
 }
 
-Transform.prototype.push=function(chunk,encoding){
-this._transformState.needTransform=false;
-returnDuplex.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){
+Transform.prototype._transform=function(chunk,encoding,callback){
 thrownewERR_METHOD_NOT_IMPLEMENTED('_transform()');
 };
 
-Transform.prototype._write=function(chunk,encoding,cb){
-constts=this._transformState;
-ts.writecb=cb;
-ts.writechunk=chunk;
-ts.writeencoding=encoding;
-if(!ts.transforming){
-constrs=this._readableState;
-if(ts.needTransform||
-rs.needReadable||
-rs.length<rs.highWaterMark)
-this._read(rs.highWaterMark);
-}
+Transform.prototype._write=function(chunk,encoding,callback){
+constrState=this._readableState;
+constwState=this._writableState;
+constlength=rState.length;
+
+this._transform(chunk,encoding,(err,val)=>{
+if(err){
+callback(err);
+return;
+}
+
+if(val!=null){
+this.push(val);
+}
+
+if(
+wState.ended||// Backwards compat.
+length===rState.length||// Backwards compat.
+rState.length<rState.highWaterMark||
+rState.length===0
+){
+callback();
+}else{
+this[kCallback]=callback;
+}
+});
 };
 
-// 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){
-constts=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;
+Transform.prototype._read=function(){
+if(this[kCallback]){
+constcallback=this[kCallback];
+this[kCallback]=null;
+callback();
 }
 };
-
-
-Transform.prototype._destroy=function(err,cb){
-Duplex.prototype._destroy.call(this,err,(err2)=>{
-cb(err2);
-});
-};
-
-
-functiondone(stream,er,data){
-if(er)
-returnstream.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)
-thrownewERR_TRANSFORM_WITH_LENGTH_0();
-
-if(stream._transformState.transforming)
-thrownewERR_TRANSFORM_ALREADY_TRANSFORMING();
-returnstream.push(null);
-}

lib/internal/errors.js

@@ -1363,12 +1363,8 @@ 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-stream2-transform.js

@@ -45,10 +45,9 @@ 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){
 returnc.chunk.length;
-}),[6,7,8,9,10]);
+}),[5,6,7,8,9,10]);
 }
 
 {

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

@@ -28,6 +28,9 @@ const ws = deflater._writableState;
 constbeforeFlush=ws.needDrain;
 letafterFlush=ws.needDrain;
 
+deflater.on('data',()=>{
+});
+
 deflater.flush(function(err){
 afterFlush=ws.needDrain;
 });