stream: simplify `.pipe()` and `.unpipe()` in Readable

Now we are using `pipes` and `pipesCount` in Readable state and the
`pipes` value can be a stream or an array of streams. This change
reducing them into one `pipes` value, which is an array of streams.

PR-URL: https://github.com/nodejs/node/pull/28583
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

1 files changed, 16 insertions, 46 deletions

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index d6db718875..d2de4122bb 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -97,8 +97,7 @@ function ReadableState(options, stream, isDuplex) {
   // array.shift()
   this.buffer = new BufferList();
   this.length = 0;
-  this.pipes = null;
-  this.pipesCount = 0;
+  this.pipes = [];
   this.flowing = null;
   this.ended = false;
   this.endEmitted = false;
@@ -148,6 +147,13 @@ function ReadableState(options, stream, isDuplex) {
   }
 }
 
+// Legacy getter for `pipesCount`
+Object.defineProperty(ReadableState.prototype, 'pipesCount', {
+  get() {
+    return this.pipes.length;
+  }
+});
+
 function Readable(options) {
   if (!(this instanceof Readable))
     return new Readable(options);
@@ -635,19 +641,8 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   const src = this;
   const state = this._readableState;
 
-  switch (state.pipesCount) {
-    case 0:
-      state.pipes = dest;
-      break;
-    case 1:
-      state.pipes = [state.pipes, dest];
-      break;
-    default:
-      state.pipes.push(dest);
-      break;
-  }
-  state.pipesCount += 1;
-  debug('pipe count=%d opts=%j', state.pipesCount, pipeOpts);
+  state.pipes.push(dest);
+  debug('pipe count=%d opts=%j', state.pipes.length, pipeOpts);
 
   const doEnd = (!pipeOpts || pipeOpts.end !== false) &&
               dest !== process.stdout &&
@@ -717,9 +712,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       // to get stuck in a permanently paused state if that write
       // also returned false.
       // => Check whether `dest` is still a piping destination.
-      if (((state.pipesCount === 1 && state.pipes === dest) ||
-           (state.pipesCount > 1 && state.pipes.includes(dest))) &&
-          !cleanedUp) {
+      if (state.pipes.length > 0 && state.pipes.includes(dest) && !cleanedUp) {
         debug('false write response, pause', state.awaitDrain);
         state.awaitDrain++;
       }
@@ -789,38 +782,16 @@ Readable.prototype.unpipe = function(dest) {
   const unpipeInfo = { hasUnpiped: false };
 
   // If we're not piping anywhere, then do nothing.
-  if (state.pipesCount === 0)
+  if (state.pipes.length === 0)
     return this;
 
-  // Just one destination.  most common case.
-  if (state.pipesCount === 1) {
-    // Passed in one, but it's not the right one.
-    if (dest && dest !== state.pipes)
-      return this;
-
-    if (!dest)
-      dest = state.pipes;
-
-    // got a match.
-    state.pipes = null;
-    state.pipesCount = 0;
-    state.flowing = false;
-    if (dest)
-      dest.emit('unpipe', this, unpipeInfo);
-    return this;
-  }
-
-  // Slow case with multiple pipe destinations.
-
   if (!dest) {
     // remove all.
     var dests = state.pipes;
-    var len = state.pipesCount;
-    state.pipes = null;
-    state.pipesCount = 0;
+    state.pipes = [];
     state.flowing = false;
 
-    for (var i = 0; i < len; i++)
+    for (var i = 0; i < dests.length; i++)
       dests[i].emit('unpipe', this, { hasUnpiped: false });
     return this;
   }
@@ -831,9 +802,8 @@ Readable.prototype.unpipe = function(dest) {
     return this;
 
   state.pipes.splice(index, 1);
-  state.pipesCount -= 1;
-  if (state.pipesCount === 1)
-    state.pipes = state.pipes[0];
+  if (state.pipes.length === 0)
+    state.flowing = false;
 
   dest.emit('unpipe', this, unpipeInfo);