Skip to content

Commit

Permalink
stream: try to wait for flush to complete before 'finish'
Browse files Browse the repository at this point in the history
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 <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Zeyu Yang <himself65@outlook.com>
  • Loading branch information
ronag authored and cjihrig committed Jul 22, 2020
1 parent 1032927 commit d9089cf
Show file tree
Hide file tree
Showing 3 changed files with 49 additions and 2 deletions.
22 changes: 20 additions & 2 deletions lib/_stream_transform.js
Original file line number Diff line number Diff line change
Expand Up @@ -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()');
};
Expand Down
5 changes: 5 additions & 0 deletions lib/zlib.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
24 changes: 24 additions & 0 deletions test/parallel/test-stream-pipeline.js
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}),
);
}

0 comments on commit d9089cf

Please sign in to comment.