stream: sync stream unpipe resume

ronag · ronag · commit 1cdf6f646771 · 2020-04-01T09:12:18.000+02:00
pipe() ondata should not control flow state if cleaned up. Fixes: nodejs#31190 PR-URL: nodejs#31191 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com> Backport-PR-URL: nodejs#32264
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -705,6 +705,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
         debug('false write response, pause', state.awaitDrain);
         state.awaitDrain++;
       }
+      if (!cleanedUp) {
+        src.pause();
+      }
       if (!ondrain) {
         // When the dest drains, it reduces the awaitDrain counter
         // on the source.  This would be more elegant with a .once()
@@ -713,7 +716,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
         ondrain = pipeOnDrain(src);
         dest.on('drain', ondrain);
       }
-      src.pause();
     }
   }
 
diff --git a/test/parallel/test-stream-readable-unpipe-resume.js b/test/parallel/test-stream-readable-unpipe-resume.js
@@ -0,0 +1,20 @@
+'use strict';
+
+const common = require('../common');
+const stream = require('stream');
+const fs = require('fs');
+
+const readStream = fs.createReadStream(process.execPath);
+
+const transformStream = new stream.Transform({
+  transform: common.mustCall(() => {
+    readStream.unpipe();
+    readStream.resume();
+  })
+});
+
+readStream.on('end', common.mustCall());
+
+readStream
+  .pipe(transformStream)
+  .resume();

Original file line number	Diff line number	Diff line change
`@@ -705,6 +705,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {`
`705`	`705`	`debug('false write response, pause', state.awaitDrain);`
`706`	`706`	`state.awaitDrain++;`
`707`	`707`	`}`
	`708`	`+ if (!cleanedUp) {`
	`709`	`+ src.pause();`
	`710`	`+ }`
`708`	`711`	`if (!ondrain) {`
`709`	`712`	`// When the dest drains, it reduces the awaitDrain counter`
`710`	`713`	`// on the source. This would be more elegant with a .once()`
`@@ -713,7 +716,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {`
`713`	`716`	`ondrain = pipeOnDrain(src);`
`714`	`717`	`dest.on('drain', ondrain);`
`715`	`718`	`}`
`716`		`- src.pause();`
`717`	`719`	`}`
`718`	`720`	`}`
`719`	`721`