stream: emit 'pause' on unpipe

ronag · MylesBorins · commit 56da8dfd8682 · 2020-03-26T14:48:57.000-04:00
unpipe should use pause() instead of mutating state.flowing directly so that pausing side effects such as emitting 'pause' are properly performed. Fixes: #32470 PR-URL: #32476 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -818,7 +818,7 @@ Readable.prototype.unpipe = function(dest) {
 // remove all.
 var dests = state.pipes;
 state.pipes = [];
- state.flowing = false;
+ this.pause();
 
 for (const dest of dests)
 dest.emit('unpipe', this, { hasUnpiped: false });
@@ -832,7 +832,7 @@ Readable.prototype.unpipe = function(dest) {
 
 state.pipes.splice(index, 1);
 if (state.pipes.length === 0)
- state.flowing = false;
+ this.pause();
 
 dest.emit('unpipe', this, unpipeInfo);
 
diff --git a/test/parallel/test-stream-pipe-unpipe-streams.js b/test/parallel/test-stream-pipe-unpipe-streams.js
@@ -84,3 +84,13 @@ assert.strictEqual(source._readableState.pipes.length, 0);
 checkDestCleanup(dest2);
 source.unpipe();
 }
+
+{
+ const src = Readable({ read: () => {} });
+ const dst = Writable({ write: () => {} });
+ src.pipe(dst);
+ src.on('resume', common.mustCall(() => {
+ src.on('pause', common.mustCall());
+ src.unpipe(dst);
+ }));
+}

-Original file line number
+Diff line change
 // remove all.
 vardests=state.pipes;
 state.pipes=[];
 -state.flowing=false;
 +this.pause();
 for(constdestofdests)
 dest.emit('unpipe',this,{hasUnpiped: false});
 state.pipes.splice(index,1);
 if(state.pipes.length===0)
 -state.flowing=false;
 +this.pause();
 dest.emit('unpipe',this,unpipeInfo);