stream: fix y.pipe(x)+y.pipe(x)+y.unpipe(x)

Fix the uncommon situation when a readable stream is piped twice into
the same destination stream, and then unpiped once.

Previously, the `unpipe` event handlers weren’t able to tell whether
they were corresponding to the “right” conceptual pipe that was being
removed; this fixes this by adding a counter to the `unpipe` event
handler and only removing a single piping destination at most.

Fixes: https://github.com/nodejs/node/issues/12718
PR-URL: https://github.com/nodejs/node/pull/12746
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

1 files changed, 78 insertions, 0 deletions

diff --git a/test/parallel/test-stream-pipe-same-destination-twice.js b/test/parallel/test-stream-pipe-same-destination-twice.js
new file mode 100644
index 0000000000..1824c06064
--- /dev/null
+++ b/test/parallel/test-stream-pipe-same-destination-twice.js
@@ -0,0 +1,78 @@
+'use strict';
+const common = require('../common');
+
+// Regression test for https://github.com/nodejs/node/issues/12718.
+// Tests that piping a source stream twice to the same destination stream
+// works, and that a subsequent unpipe() call only removes the pipe *once*.
+const assert = require('assert');
+const { PassThrough, Writable } = require('stream');
+
+{
+  const passThrough = new PassThrough();
+  const dest = new Writable({
+    write: common.mustCall((chunk, encoding, cb) => {
+      assert.strictEqual(`${chunk}`, 'foobar');
+      cb();
+    })
+  });
+
+  passThrough.pipe(dest);
+  passThrough.pipe(dest);
+
+  assert.strictEqual(passThrough._events.data.length, 2);
+  assert.strictEqual(passThrough._readableState.pipesCount, 2);
+  assert.strictEqual(passThrough._readableState.pipes[0], dest);
+  assert.strictEqual(passThrough._readableState.pipes[1], dest);
+
+  passThrough.unpipe(dest);
+
+  assert.strictEqual(passThrough._events.data.length, 1);
+  assert.strictEqual(passThrough._readableState.pipesCount, 1);
+  assert.strictEqual(passThrough._readableState.pipes, dest);
+
+  passThrough.write('foobar');
+  passThrough.pipe(dest);
+}
+
+{
+  const passThrough = new PassThrough();
+  const dest = new Writable({
+    write: common.mustCall((chunk, encoding, cb) => {
+      assert.strictEqual(`${chunk}`, 'foobar');
+      cb();
+    }, 2)
+  });
+
+  passThrough.pipe(dest);
+  passThrough.pipe(dest);
+
+  assert.strictEqual(passThrough._events.data.length, 2);
+  assert.strictEqual(passThrough._readableState.pipesCount, 2);
+  assert.strictEqual(passThrough._readableState.pipes[0], dest);
+  assert.strictEqual(passThrough._readableState.pipes[1], dest);
+
+  passThrough.write('foobar');
+}
+
+{
+  const passThrough = new PassThrough();
+  const dest = new Writable({
+    write: common.mustNotCall()
+  });
+
+  passThrough.pipe(dest);
+  passThrough.pipe(dest);
+
+  assert.strictEqual(passThrough._events.data.length, 2);
+  assert.strictEqual(passThrough._readableState.pipesCount, 2);
+  assert.strictEqual(passThrough._readableState.pipes[0], dest);
+  assert.strictEqual(passThrough._readableState.pipes[1], dest);
+
+  passThrough.unpipe(dest);
+  passThrough.unpipe(dest);
+
+  assert.strictEqual(passThrough._events.data, undefined);
+  assert.strictEqual(passThrough._readableState.pipesCount, 0);
+
+  passThrough.write('foobar');
+}