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 <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Luigi Pinca <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: ronag

lib/internal/streams/readable.js

@@ -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();
 }
 };
 }

test/parallel/test-stream-readable-pause-and-resume.js

@@ -56,3 +56,19 @@ function readAndPause(){
 assert(readable.isPaused());
 });
 }
+
+{
+const{ PassThrough }=require('stream');
+
+constsource3=newPassThrough();
+consttarget3=newPassThrough();
+
+constchunk=Buffer.allocUnsafe(1000);
+while(target3.write(chunk));
+
+source3.pipe(target3);
+target3.on('drain',common.mustCall(()=>{
+assert(!source3.isPaused());
+}));
+target3.on('data',()=>{});
+}