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: https://github.com/nodejs/node/pull/12746
PR-URL: https://github.com/nodejs/node/pull/18266
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Anatoli Papirovski <apapirovski@mac.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

lib/_stream_readable.js

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

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.
+  const source = Readable({ read: () => {} });
+  const dest1 = Writable({ write: () => {} });
+  const dest2 = Writable({ write: () => {} });
+
+  let destCount = 0;
+  const srcCheckEventNames = ['end', 'data'];
+  const destCheckEventNames = ['close', 'finish', 'drain', 'error', 'unpipe'];
+
+  const checkSrcCleanup = 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`
+      );
+    });
+  });
+
+  function checkDestCleanup(dest) {
+    const currentDestId = ++destCount;
+    source.pipe(dest);
+
+    const unpipeChecker = 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();
+}