Skip to content

Commit 241ffd3

Browse files
starkwangstarkwang
committed
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. It also adds a deprecation warning of `_readableState.pipesCount`.
1 parent 8850ef2 commit 241ffd3

File tree

7 files changed

+60
-64
lines changed

7 files changed

+60
-64
lines changed

‎doc/api/deprecations.md‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2498,6 +2498,20 @@ Type: Runtime
24982498
Passing a callback to [`worker.terminate()`][] is deprecated. Use the returned
24992499
`Promise` instead, or a listener to the worker’s `'exit'` event.
25002500
2501+
<a id="DEP0133"></a>
2502+
### DEP0133: _readableState.pipesCount
2503+
<!-- YAML
2504+
changes:
2505+
- version: REPLACEME
2506+
pr-url: https://github.com/nodejs/node/pull/28583
2507+
description: Documentation-only.
2508+
-->
2509+
2510+
Type: Documentation-only
2511+
2512+
`_readableState.pipesCount` is deprecated. Please use
2513+
`_readableState.pipes.length` instead.
2514+
25012515
[`--http-parser=legacy`]: cli.html#cli_http_parser_library
25022516
[`--pending-deprecation`]: cli.html#cli_pending_deprecation
25032517
[`Buffer.allocUnsafeSlow(size)`]: buffer.html#buffer_class_method_buffer_allocunsafeslow_size

‎lib/_stream_readable.js‎

Lines changed: 17 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ const EE = require('events');
3030
const Stream = require('stream');
3131
const { Buffer } = require('buffer');
3232

33+
const internalUtil = require('internal/util');
3334
const debug = require('internal/util/debuglog').debuglog('stream');
3435
const BufferList = require('internal/streams/buffer_list');
3536
const destroyImpl = require('internal/streams/destroy');
@@ -97,8 +98,7 @@ function ReadableState(options, stream, isDuplex) {
9798
// array.shift()
9899
this.buffer = new BufferList();
99100
this.length = 0;
100-
this.pipes = null;
101-
this.pipesCount = 0;
101+
this.pipes = [];
102102
this.flowing = null;
103103
this.ended = false;
104104
this.endEmitted = false;
@@ -148,6 +148,13 @@ function ReadableState(options, stream, isDuplex) {
148148
}
149149
}
150150

151+
Object.defineProperty(ReadableState.prototype, 'pipesCount', {
152+
get: internalUtil.deprecate(function() {
153+
return this.pipes.length;
154+
}, '_readableState.pipesCount is deprecated. ' +
155+
'Use _readableState.pipes.length instead.', 'DEP0133'),
156+
});
157+
151158
function Readable(options) {
152159
if (!(this instanceof Readable))
153160
return new Readable(options);
@@ -635,19 +642,8 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
635642
const src = this;
636643
const state = this._readableState;
637644

638-
switch (state.pipesCount) {
639-
case 0:
640-
state.pipes = dest;
641-
break;
642-
case 1:
643-
state.pipes = [state.pipes, dest];
644-
break;
645-
default:
646-
state.pipes.push(dest);
647-
break;
648-
}
649-
state.pipesCount += 1;
650-
debug('pipe count=%d opts=%j', state.pipesCount, pipeOpts);
645+
state.pipes.push(dest);
646+
debug('pipe count=%d opts=%j', state.pipes.length, pipeOpts);
651647

652648
const doEnd = (!pipeOpts || pipeOpts.end !== false) &&
653649
dest !== process.stdout &&
@@ -717,9 +713,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
717713
// to get stuck in a permanently paused state if that write
718714
// also returned false.
719715
// => Check whether `dest` is still a piping destination.
720-
if (((state.pipesCount === 1 && state.pipes === dest) ||
721-
(state.pipesCount > 1 && state.pipes.includes(dest))) &&
722-
!cleanedUp) {
716+
if (state.pipes.length > 0 && state.pipes.includes(dest) && !cleanedUp) {
723717
debug('false write response, pause', state.awaitDrain);
724718
state.awaitDrain++;
725719
}
@@ -789,38 +783,16 @@ Readable.prototype.unpipe = function(dest) {
789783
const unpipeInfo = { hasUnpiped: false };
790784

791785
// If we're not piping anywhere, then do nothing.
792-
if (state.pipesCount === 0)
793-
return this;
794-
795-
// Just one destination. most common case.
796-
if (state.pipesCount === 1) {
797-
// Passed in one, but it's not the right one.
798-
if (dest && dest !== state.pipes)
799-
return this;
800-
801-
if (!dest)
802-
dest = state.pipes;
803-
804-
// got a match.
805-
state.pipes = null;
806-
state.pipesCount = 0;
807-
state.flowing = false;
808-
if (dest)
809-
dest.emit('unpipe', this, unpipeInfo);
786+
if (state.pipes.length === 0)
810787
return this;
811-
}
812-
813-
// Slow case with multiple pipe destinations.
814788

815789
if (!dest) {
816790
// remove all.
817791
var dests = state.pipes;
818-
var len = state.pipesCount;
819-
state.pipes = null;
820-
state.pipesCount = 0;
792+
state.pipes = [];
821793
state.flowing = false;
822794

823-
for (var i = 0; i < len; i++)
795+
for (var i = 0; i < dests.length; i++)
824796
dests[i].emit('unpipe', this, { hasUnpiped: false });
825797
return this;
826798
}
@@ -831,9 +803,8 @@ Readable.prototype.unpipe = function(dest) {
831803
return this;
832804

833805
state.pipes.splice(index, 1);
834-
state.pipesCount -= 1;
835-
if (state.pipesCount === 1)
836-
state.pipes = state.pipes[0];
806+
if (state.pipes.length === 0)
807+
state.flowing = false;
837808

838809
dest.emit('unpipe', this, unpipeInfo);
839810

‎test/parallel/test-stream-pipe-same-destination-twice.js‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,15 @@ const { PassThrough, Writable } = require('stream');
2020
passThrough.pipe(dest);
2121

2222
assert.strictEqual(passThrough._events.data.length, 2);
23-
assert.strictEqual(passThrough._readableState.pipesCount, 2);
23+
assert.strictEqual(passThrough._readableState.pipes.length, 2);
2424
assert.strictEqual(passThrough._readableState.pipes[0], dest);
2525
assert.strictEqual(passThrough._readableState.pipes[1], dest);
2626

2727
passThrough.unpipe(dest);
2828

2929
assert.strictEqual(passThrough._events.data.length, 1);
30-
assert.strictEqual(passThrough._readableState.pipesCount, 1);
31-
assert.strictEqual(passThrough._readableState.pipes, dest);
30+
assert.strictEqual(passThrough._readableState.pipes.length, 1);
31+
assert.deepStrictEqual(passThrough._readableState.pipes, [dest]);
3232

3333
passThrough.write('foobar');
3434
passThrough.pipe(dest);
@@ -47,7 +47,7 @@ const { PassThrough, Writable } = require('stream');
4747
passThrough.pipe(dest);
4848

4949
assert.strictEqual(passThrough._events.data.length, 2);
50-
assert.strictEqual(passThrough._readableState.pipesCount, 2);
50+
assert.strictEqual(passThrough._readableState.pipes.length, 2);
5151
assert.strictEqual(passThrough._readableState.pipes[0], dest);
5252
assert.strictEqual(passThrough._readableState.pipes[1], dest);
5353

@@ -64,15 +64,15 @@ const { PassThrough, Writable } = require('stream');
6464
passThrough.pipe(dest);
6565

6666
assert.strictEqual(passThrough._events.data.length, 2);
67-
assert.strictEqual(passThrough._readableState.pipesCount, 2);
67+
assert.strictEqual(passThrough._readableState.pipes.length, 2);
6868
assert.strictEqual(passThrough._readableState.pipes[0], dest);
6969
assert.strictEqual(passThrough._readableState.pipes[1], dest);
7070

7171
passThrough.unpipe(dest);
7272
passThrough.unpipe(dest);
7373

7474
assert.strictEqual(passThrough._events.data, undefined);
75-
assert.strictEqual(passThrough._readableState.pipesCount, 0);
75+
assert.strictEqual(passThrough._readableState.pipes.length, 0);
7676

7777
passThrough.write('foobar');
7878
}

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

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,15 +22,15 @@ assert.strictEqual(source._readableState.pipes.length, 2);
2222

2323
source.unpipe(dest2);
2424

25-
assert.strictEqual(source._readableState.pipes, dest1);
25+
assert.deepStrictEqual(source._readableState.pipes, [dest1]);
2626
assert.notStrictEqual(source._readableState.pipes, dest2);
2727

2828
dest2.on('unpipe', common.mustNotCall());
2929
source.unpipe(dest2);
3030

3131
source.unpipe(dest1);
3232

33-
assert.strictEqual(source._readableState.pipes, null);
33+
assert.strictEqual(source._readableState.pipes.length, 0);
3434

3535
{
3636
// Test `cleanup()` if we unpipe all streams.
@@ -43,8 +43,7 @@ assert.strictEqual(source._readableState.pipes, null);
4343
const destCheckEventNames = ['close', 'finish', 'drain', 'error', 'unpipe'];
4444

4545
const checkSrcCleanup = common.mustCall(() => {
46-
assert.strictEqual(source._readableState.pipes, null);
47-
assert.strictEqual(source._readableState.pipesCount, 0);
46+
assert.strictEqual(source._readableState.pipes.length, 0);
4847
assert.strictEqual(source._readableState.flowing, false);
4948

5049
srcCheckEventNames.forEach((eventName) => {

‎test/parallel/test-stream-pipescount-deprecation-warning.js‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
'use strict';
2+
3+
const common = require('../common');
4+
const { Readable } = require('stream');
5+
6+
const warning = '_readableState.pipesCount is deprecated. ' +
7+
'Use _readableState.pipes.length instead.';
8+
9+
common.expectWarning('DeprecationWarning', warning, 'DEP0133');
10+
11+
const readable = new Readable();
12+
readable._readableState.pipesCount;

‎test/parallel/test-stream-unpipe-event.js‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ class NeverEndReadable extends Readable {
2323
dest.on('unpipe', common.mustCall());
2424
src.pipe(dest);
2525
setImmediate(() => {
26-
assert.strictEqual(src._readableState.pipesCount, 0);
26+
assert.strictEqual(src._readableState.pipes.length, 0);
2727
});
2828
}
2929

@@ -34,7 +34,7 @@ class NeverEndReadable extends Readable {
3434
dest.on('unpipe', common.mustNotCall('unpipe should not have been emitted'));
3535
src.pipe(dest);
3636
setImmediate(() => {
37-
assert.strictEqual(src._readableState.pipesCount, 1);
37+
assert.strictEqual(src._readableState.pipes.length, 1);
3838
});
3939
}
4040

@@ -46,7 +46,7 @@ class NeverEndReadable extends Readable {
4646
src.pipe(dest);
4747
src.unpipe(dest);
4848
setImmediate(() => {
49-
assert.strictEqual(src._readableState.pipesCount, 0);
49+
assert.strictEqual(src._readableState.pipes.length, 0);
5050
});
5151
}
5252

@@ -57,7 +57,7 @@ class NeverEndReadable extends Readable {
5757
dest.on('unpipe', common.mustCall());
5858
src.pipe(dest, { end: false });
5959
setImmediate(() => {
60-
assert.strictEqual(src._readableState.pipesCount, 0);
60+
assert.strictEqual(src._readableState.pipes.length, 0);
6161
});
6262
}
6363

@@ -68,7 +68,7 @@ class NeverEndReadable extends Readable {
6868
dest.on('unpipe', common.mustNotCall('unpipe should not have been emitted'));
6969
src.pipe(dest, { end: false });
7070
setImmediate(() => {
71-
assert.strictEqual(src._readableState.pipesCount, 1);
71+
assert.strictEqual(src._readableState.pipes.length, 1);
7272
});
7373
}
7474

@@ -80,6 +80,6 @@ class NeverEndReadable extends Readable {
8080
src.pipe(dest, { end: false });
8181
src.unpipe(dest);
8282
setImmediate(() => {
83-
assert.strictEqual(src._readableState.pipesCount, 0);
83+
assert.strictEqual(src._readableState.pipes.length, 0);
8484
});
8585
}

‎test/parallel/test-stream2-basic.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -171,10 +171,10 @@ class TestWriter extends EE {
171171
w[0].on('write', function() {
172172
if (--writes === 0) {
173173
r.unpipe();
174-
assert.strictEqual(r._readableState.pipes, null);
174+
assert.deepStrictEqual(r._readableState.pipes, []);
175175
w[0].end();
176176
r.pipe(w[1]);
177-
assert.strictEqual(r._readableState.pipes, w[1]);
177+
assert.deepStrictEqual(r._readableState.pipes, [w[1]]);
178178
}
179179
});
180180

0 commit comments

Comments
 (0)