Commit 07b6bf4
stream: resume stream on drain
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
PR-URL: #41848
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>1 parent 5085813 commit 07b6bf4
2 files changed
Lines changed: 17 additions & 2 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -853,8 +853,7 @@ function pipeOnDrain(src, dest) { | |||
| 853 | 853 | ||
| 854 | 854 | if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) && | |
| 855 | 855 | EE.listenerCount(src, 'data')) { | |
| 856 | - state.flowing = true; | ||
| 857 | - flow(src); | ||
| 856 | + src.resume(); | ||
| 858 | 857 | } | |
| 859 | 858 | }; | |
| 860 | 859 | } | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -56,3 +56,19 @@ function readAndPause() { | |||
| 56 | 56 | assert(readable.isPaused()); | |
| 57 | 57 | }); | |
| 58 | 58 | } | |
| 59 | + | ||
| 60 | + { | ||
| 61 | + const { PassThrough } = require('stream'); | ||
| 62 | + | ||
| 63 | + const source3 = new PassThrough(); | ||
| 64 | + const target3 = new PassThrough(); | ||
| 65 | + | ||
| 66 | + const chunk = Buffer.allocUnsafe(1000); | ||
| 67 | + while (target3.write(chunk)); | ||
| 68 | + | ||
| 69 | + source3.pipe(target3); | ||
| 70 | + target3.on('drain', common.mustCall(() => { | ||
| 71 | + assert(!source3.isPaused()); | ||
| 72 | + })); | ||
| 73 | + target3.on('data', () => {}); | ||
| 74 | + } | ||
0 commit comments