stream: cleanup() when unpiping all streams.

This PR makes sure the object emitted as the 'unpipe' event in the destination stream is not shared between destination, as it would be muted. Refs: #12746 PR-URL: #18266 Reviewed-By: Luigi Pinca <[email protected]> Reviewed-By: Anatoli Papirovski <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: James M Snell <[email protected]> Reviewed-By: Ruben Bridgewater <[email protected]> Reviewed-By: Anna Henningsen <[email protected]>

Author: 陈刚

lib/_stream_readable.js

@@ -754,7 +754,7 @@ Readable.prototype.unpipe = function(dest){
 state.flowing=false;
 
 for(vari=0;i<len;i++)
-dests[i].emit('unpipe',this,unpipeInfo);
+dests[i].emit('unpipe',this,{hasUnpiped: false});
 returnthis;
 }
 

test/parallel/test-stream-pipe-unpipe-streams.js

@@ -31,3 +31,57 @@ source.unpipe(dest2);
 source.unpipe(dest1);
 
 assert.strictEqual(source._readableState.pipes,null);
+
+{
+// test `cleanup()` if we unpipe all streams.
+constsource=Readable({read: ()=>{}});
+constdest1=Writable({write: ()=>{}});
+constdest2=Writable({write: ()=>{}});
+
+letdestCount=0;
+constsrcCheckEventNames=['end','data'];
+constdestCheckEventNames=['close','finish','drain','error','unpipe'];
+
+constcheckSrcCleanup=common.mustCall(()=>{
+assert.strictEqual(source._readableState.pipes,null);
+assert.strictEqual(source._readableState.pipesCount,0);
+assert.strictEqual(source._readableState.flowing,false);
+
+srcCheckEventNames.forEach((eventName)=>{
+assert.strictEqual(
+source.listenerCount(eventName),0,
+`source's '${eventName}' event listeners not removed`
+);
+});
+});
+
+functioncheckDestCleanup(dest){
+constcurrentDestId=++destCount;
+source.pipe(dest);
+
+constunpipeChecker=common.mustCall(()=>{
+assert.deepStrictEqual(
+dest.listeners('unpipe'),[unpipeChecker],
+`destination{${currentDestId}} should have a 'unpipe' event `+
+'listener which is `unpipeChecker`'
+);
+dest.removeListener('unpipe',unpipeChecker);
+destCheckEventNames.forEach((eventName)=>{
+assert.strictEqual(
+dest.listenerCount(eventName),0,
+`destination{${currentDestId}}'s '${eventName}' event `+
+'listeners not removed'
+);
+});
+
+if(--destCount===0)
+checkSrcCleanup();
+});
+
+dest.on('unpipe',unpipeChecker);
+}
+
+checkDestCleanup(dest1);
+checkDestCleanup(dest2);
+source.unpipe();
+}