stream: try to wait for flush to complete before 'finish'

Due to compat reasons Transform streams don't always wait
for flush to complete before finishing the stream.

Try to wait when possible, i.e. when the user does not
override _final.

Fixes: #34274

PR-URL: #34314
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Zeyu Yang <[email protected]>

lib/_stream_transform.js

@@ -106,24 +106,42 @@ function Transform(options) {
   this.on('prefinish', prefinish);
 }
 
-function prefinish() {
+function final(cb) {
   if (typeof this._flush === 'function' && !this.destroyed) {
     this._flush((er, data) => {
       if (er) {
-        this.destroy(er);
+        if (cb) {
+          cb(er);
+        } else {
+          this.destroy(er);
+        }
         return;
       }
 
       if (data != null) {
         this.push(data);
       }
       this.push(null);
+      if (cb) {
+        cb();
+      }
     });
   } else {
     this.push(null);
+    if (cb) {
+      cb();
+    }
+  }
+}
+
+function prefinish() {
+  if (this._final !== final) {
+    final.call(this);
   }
 }
 
+Transform.prototype._final = final;
+
 Transform.prototype._transform = function(chunk, encoding, callback) {
   throw new ERR_METHOD_NOT_IMPLEMENTED('_transform()');
 };

lib/zlib.js

@@ -323,6 +323,11 @@ ZlibBase.prototype._flush = function(callback) {
   this._transform(Buffer.alloc(0), '', callback);
 };
 
+// Force Transform compat behavior.
+ZlibBase.prototype._final = function(callback) {
+  callback();
+};
+
 // If a flush is scheduled while another flush is still pending, a way to figure
 // out which one is the "stronger" flush is needed.
 // This is currently only used to figure out which flush flag to use for the

test/parallel/test-stream-pipeline.js

@@ -1231,3 +1231,27 @@ const net = require('net');
     assert.strictEqual(res, 'helloworld');
   }));
 }
+
+{
+  let flushed = false;
+  const makeStream = () =>
+    new Transform({
+      transform: (chunk, enc, cb) => cb(null, chunk),
+      flush: (cb) =>
+        setTimeout(() => {
+          flushed = true;
+          cb(null);
+        }, 1),
+    });
+
+  const input = new Readable();
+  input.push(null);
+
+  pipeline(
+    input,
+    makeStream(),
+    common.mustCall(() => {
+      assert.strictEqual(flushed, true);
+    }),
+  );
+}