Skip to content

Commit

Permalink
stream: resume stream on drain
Browse files Browse the repository at this point in the history
Previously we would just resume "flowing" the stream without
reseting the "paused" state. Fixes this by properly using
pause/resume methods for .pipe.

Fixes: #41785
  • Loading branch information
ronag committed Feb 4, 2022
1 parent 7faf763 commit 216bb20
Show file tree
Hide file tree
Showing 3 changed files with 42 additions and 2 deletions.
3 changes: 1 addition & 2 deletions lib/internal/streams/readable.js
Expand Up @@ -853,8 +853,7 @@ function pipeOnDrain(src, dest) {

if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&
EE.listenerCount(src, 'data')) {
state.flowing = true;
flow(src);
src.resume();
}
};
}
Expand Down
17 changes: 17 additions & 0 deletions test/parallel/test-stream-readable-pause-and-resume.js
Expand Up @@ -56,3 +56,20 @@ function readAndPause() {
assert(readable.isPaused());
});
}

{
const { PassThrough } = require('stream');

const source3 = new PassThrough();
const target3 = new PassThrough();

const chunk = Buffer.allocUnsafe(1000);
let chunks = 1;
while (target3.write(chunk)) chunks++;

source3.pipe(target3);
target3.on('drain', common.mustCall(() => {
assert(!source3.isPaused());
}));
target3.on('data', () => {});
}
24 changes: 24 additions & 0 deletions tmp.js
@@ -0,0 +1,24 @@
const { PassThrough } = require('stream');

// THIRD EXPERIMENT
console.info('\n********** THIRD EXPERIMENT **********');
const source3 = new PassThrough();
const target3 = new PassThrough();

// stall target3
const chunk = Buffer.allocUnsafe(1000);
let chunks = 1;
while (target3.write(chunk)) chunks++;
console.info(`${chunks} chunks of ${chunk.length} bytes to stall target3`);

// `Readable.pipe()` PAUSES the source if the target needs drain (only in
// version >= v14.17.0) and it does not resume it after drain
console.info(`source3 before pipe. Paused: ${source3.isPaused()}`);
source3.pipe(target3);
console.info(`source3 after pipe. Paused: ${source3.isPaused()}`);
target3.on('drain', () => {
console.info('target3 drained');
console.info(`source3 after drain. Paused: ${source3.isPaused()}`);
console.info(source3.readableFlowing);
});
target3.on('data', () => {});

0 comments on commit 216bb20

Please sign in to comment.