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: nodejs#41785 PR-URL: nodejs#41848 Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Luigi Pinca <[email protected]> Reviewed-By: James M Snell <[email protected]>
danielleadams · Mar 4, 2022 · fc41c2f · fc41c2f
1 parent 3ac9e80
commit fc41c2f
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 2 deletions.
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
@@ -830,8 +830,7 @@ function pipeOnDrain(src, dest) {
 
     if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&
       EE.listenerCount(src, 'data')) {
-      state.flowing = true;
-      flow(src);
+      src.resume();
     }
   };
 }

diff --git a/test/parallel/test-stream-readable-pause-and-resume.js b/test/parallel/test-stream-readable-pause-and-resume.js
@@ -56,3 +56,19 @@ function readAndPause() {
     assert(readable.isPaused());
   });
 }
+
+{
+  const { PassThrough } = require('stream');
+
+  const source3 = new PassThrough();
+  const target3 = new PassThrough();
+
+  const chunk = Buffer.allocUnsafe(1000);
+  while (target3.write(chunk));
+
+  source3.pipe(target3);
+  target3.on('drain', common.mustCall(() => {
+    assert(!source3.isPaused());
+  }));
+  target3.on('data', () => {});
+}