Skip to content

Commit 0fd4a82

Browse files
committed
stream: unify stream utils
Unify stream helps into utils.
1 parent 4de6f20 commit 0fd4a82

File tree

6 files changed

+148
-66
lines changed

6 files changed

+148
-66
lines changed

lib/internal/streams/destroy.js

Lines changed: 6 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,11 @@ const {
1010
const {
1111
Symbol,
1212
} = primordials;
13+
const {
14+
kDestroyed,
15+
isDestroyed,
16+
isFinished,
17+
} = require('internal/streams/utils');
1318

1419
const kDestroy = Symbol('kDestroy');
1520
const kConstruct = Symbol('kConstruct');
@@ -364,8 +369,6 @@ function isRequest(stream) {
364369
return stream && stream.setHeader && typeof stream.abort === 'function';
365370
}
366371

367-
const kDestroyed = Symbol('kDestroyed');
368-
369372
function emitCloseLegacy(stream) {
370373
stream.emit('close');
371374
}
@@ -375,25 +378,13 @@ function emitErrorCloseLegacy(stream, err) {
375378
process.nextTick(emitCloseLegacy, stream);
376379
}
377380

378-
function isDestroyed(stream) {
379-
return stream.destroyed || stream[kDestroyed];
380-
}
381-
382-
function isReadable(stream) {
383-
return stream.readable && !stream.readableEnded && !isDestroyed(stream);
384-
}
385-
386-
function isWritable(stream) {
387-
return stream.writable && !stream.writableEnded && !isDestroyed(stream);
388-
}
389-
390381
// Normalize destroy for legacy.
391382
function destroyer(stream, err) {
392383
if (isDestroyed(stream)) {
393384
return;
394385
}
395386

396-
if (!err && (isReadable(stream) || isWritable(stream))) {
387+
if (!err && !isFinished(stream)) {
397388
err = new AbortError();
398389
}
399390

@@ -422,7 +413,6 @@ function destroyer(stream, err) {
422413

423414
module.exports = {
424415
kDestroy,
425-
isDestroyed,
426416
construct,
427417
destroyer,
428418
destroy,

lib/internal/streams/end-of-stream.js

Lines changed: 15 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,14 @@ const {
1717
validateObject,
1818
} = require('internal/validators');
1919

20+
const {
21+
isClosed,
22+
isReadableStream,
23+
isReadableEnded,
24+
isWritableStream,
25+
isWritableFinished,
26+
} = require('internal/streams/utils');
27+
2028
function isRequest(stream) {
2129
return stream.setHeader && typeof stream.abort === 'function';
2230
}
@@ -31,34 +39,8 @@ function isServerResponse(stream) {
3139
);
3240
}
3341

34-
function isReadable(stream) {
35-
return typeof stream.readable === 'boolean' ||
36-
typeof stream.readableEnded === 'boolean' ||
37-
!!stream._readableState;
38-
}
39-
40-
function isWritable(stream) {
41-
return typeof stream.writable === 'boolean' ||
42-
typeof stream.writableEnded === 'boolean' ||
43-
!!stream._writableState;
44-
}
45-
46-
function isWritableFinished(stream) {
47-
if (stream.writableFinished) return true;
48-
const wState = stream._writableState;
49-
if (!wState || wState.errored) return false;
50-
return wState.finished || (wState.ended && wState.length === 0);
51-
}
52-
5342
const nop = () => {};
5443

55-
function isReadableEnded(stream) {
56-
if (stream.readableEnded) return true;
57-
const rState = stream._readableState;
58-
if (!rState || rState.errored) return false;
59-
return rState.endEmitted || (rState.ended && rState.length === 0);
60-
}
61-
6244
function eos(stream, options, callback) {
6345
if (arguments.length === 2) {
6446
callback = options;
@@ -74,9 +56,9 @@ function eos(stream, options, callback) {
7456
callback = once(callback);
7557

7658
const readable = options.readable ||
77-
(options.readable !== false && isReadable(stream));
59+
(options.readable !== false && isReadableStream(stream, true));
7860
const writable = options.writable ||
79-
(options.writable !== false && isWritable(stream));
61+
(options.writable !== false && isWritableStream(stream, true));
8062

8163
const wState = stream._writableState;
8264
const rState = stream._readableState;
@@ -94,11 +76,11 @@ function eos(stream, options, callback) {
9476
state.autoDestroy &&
9577
state.emitClose &&
9678
state.closed === false &&
97-
isReadable(stream) === readable &&
98-
isWritable(stream) === writable
79+
isReadableStream(stream) === readable &&
80+
isWritableStream(stream) === writable
9981
);
10082

101-
let writableFinished = stream.writableFinished || wState?.finished;
83+
let writableFinished = isWritableFinished(stream);
10284
const onfinish = () => {
10385
writableFinished = true;
10486
// Stream should not be destroyed here. If it is that
@@ -110,7 +92,7 @@ function eos(stream, options, callback) {
11092
if (!readable || readableEnded) callback.call(stream);
11193
};
11294

113-
let readableEnded = stream.readableEnded || rState?.endEmitted;
95+
let readableEnded = isReadableEnded(stream);
11496
const onend = () => {
11597
readableEnded = true;
11698
// Stream should not be destroyed here. If it is that
@@ -126,7 +108,7 @@ function eos(stream, options, callback) {
126108
callback.call(stream, err);
127109
};
128110

129-
let closed = wState?.closed || rState?.closed;
111+
let closed = isClosed(stream);
130112

131113
const onclose = () => {
132114
closed = true;
@@ -195,9 +177,6 @@ function eos(stream, options, callback) {
195177
readableEnded
196178
) {
197179
process.nextTick(onclose);
198-
} else if (!wState && !rState && stream._closed === true) {
199-
// _closed is for OutgoingMessage which is not a proper Writable.
200-
process.nextTick(onclose);
201180
} else if ((rState && stream.req && stream.aborted)) {
202181
process.nextTick(onclose);
203182
}

lib/internal/streams/pipeline.js

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ const { validateCallback } = require('internal/validators');
2727

2828
const {
2929
isIterable,
30-
isReadable,
30+
isReadableStream,
3131
isStream,
3232
} = require('internal/streams/utils');
3333

@@ -87,7 +87,7 @@ function popCallback(streams) {
8787
function makeAsyncIterable(val) {
8888
if (isIterable(val)) {
8989
return val;
90-
} else if (isReadable(val)) {
90+
} else if (isReadableStream(val)) {
9191
// Legacy streams are not Iterable.
9292
return fromReadable(val);
9393
}
@@ -216,7 +216,7 @@ function pipeline(...streams) {
216216
throw new ERR_INVALID_RETURN_VALUE(
217217
'Iterable, AsyncIterable or Stream', 'source', ret);
218218
}
219-
} else if (isIterable(stream) || isReadable(stream)) {
219+
} else if (isIterable(stream) || isReadableStream(stream)) {
220220
ret = stream;
221221
} else {
222222
throw new ERR_INVALID_ARG_TYPE(
@@ -272,7 +272,7 @@ function pipeline(...streams) {
272272
destroys.push(destroyer(ret, false, true, finish));
273273
}
274274
} else if (isStream(stream)) {
275-
if (isReadable(ret)) {
275+
if (isReadableStream(ret)) {
276276
ret.pipe(stream);
277277

278278
// Compat. Before node v10.12.0 stdio used to throw an error so

lib/internal/streams/utils.js

Lines changed: 121 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,20 +3,44 @@
33
const {
44
SymbolAsyncIterator,
55
SymbolIterator,
6+
Symbol
67
} = primordials;
78

8-
function isReadable(obj) {
9-
return !!(obj && typeof obj.pipe === 'function' &&
10-
typeof obj.on === 'function');
9+
const kDestroyed = Symbol('kDestroyed');
10+
11+
function isReadableStream(obj, strict) {
12+
return !!(
13+
obj &&
14+
typeof obj.pipe === 'function' &&
15+
typeof obj.on === 'function' &&
16+
(!obj._writableState || obj._readableState?.readable !== false) && // Duplex
17+
(!obj._writableState || obj._readableState) && // Writable has .pipe.
18+
(
19+
!strict ||
20+
typeof obj.readable === 'boolean' ||
21+
typeof obj.readableEnded === 'boolean' ||
22+
obj._readableState
23+
)
24+
);
1125
}
1226

13-
function isWritable(obj) {
14-
return !!(obj && typeof obj.write === 'function' &&
15-
typeof obj.on === 'function');
27+
function isWritableStream(obj, strict) {
28+
return !!(
29+
obj &&
30+
typeof obj.write === 'function' &&
31+
typeof obj.on === 'function' &&
32+
(!obj._readableState || obj._writableState?.writable !== false) && // Duplex
33+
(
34+
!strict ||
35+
typeof obj.writable === 'boolean' ||
36+
typeof obj.writableEnded === 'boolean' ||
37+
obj._writableState
38+
)
39+
);
1640
}
1741

1842
function isStream(obj) {
19-
return isReadable(obj) || isWritable(obj);
43+
return isReadableStream(obj) || isWritableStream(obj);
2044
}
2145

2246
function isIterable(obj, isAsync) {
@@ -27,8 +51,97 @@ function isIterable(obj, isAsync) {
2751
typeof obj[SymbolIterator] === 'function';
2852
}
2953

54+
function isDestroyed(stream) {
55+
if (!isStream(stream)) return null;
56+
const wState = stream._writableState;
57+
const rState = stream._readableState;
58+
const state = wState || rState;
59+
return !!(stream.destroyed || stream[kDestroyed] || state?.destroyed);
60+
}
61+
62+
function isWritableFinished(stream) {
63+
if (!isStream(stream)) return null;
64+
if (stream.writableFinished === true) return true;
65+
const wState = stream._writableState;
66+
if (!wState || wState.errored) return false;
67+
return wState.finished || (wState.ended && wState.length === 0);
68+
}
69+
70+
function isReadableEnded(stream) {
71+
if (!isStream(stream)) return null;
72+
if (stream.readableEnded === true) return true;
73+
const rState = stream._readableState;
74+
if (!rState || rState.errored) return false;
75+
return rState.endEmitted || (rState.ended && rState.length === 0);
76+
}
77+
78+
function isFinished(stream, opts) {
79+
if (!isStream(stream)) {
80+
return null;
81+
}
82+
83+
if (isDestroyed(stream)) {
84+
return true;
85+
}
86+
87+
if (
88+
opts?.readable !== false &&
89+
isReadableStream(stream) &&
90+
!isReadableEnded(stream)
91+
) {
92+
return false;
93+
}
94+
95+
if (
96+
opts?.writable !== false &&
97+
isWritableStream(stream) &&
98+
!isWritableFinished(stream)
99+
) {
100+
return false;
101+
}
102+
103+
return true;
104+
}
105+
106+
function isClosed(stream) {
107+
if (!isStream(stream)) {
108+
return null;
109+
}
110+
111+
const wState = stream._writableState;
112+
const rState = stream._readableState;
113+
114+
if (
115+
typeof wState?.closed === 'boolean' ||
116+
typeof rState?.closed === 'boolean'
117+
) {
118+
return wState?.closed || rState?.closed;
119+
}
120+
121+
// OutgoingMessage
122+
if (
123+
!wState &&
124+
!rState &&
125+
typeof stream._closed === 'boolean' &&
126+
typeof stream._defaultKeepAlive === 'boolean' &&
127+
typeof stream._removedConnection === 'boolean' &&
128+
typeof stream._removedContLen === 'boolean'
129+
) {
130+
return stream._closed;
131+
}
132+
133+
return null;
134+
}
135+
30136
module.exports = {
137+
kDestroyed,
138+
isClosed,
139+
isDestroyed,
140+
isFinished,
31141
isIterable,
32-
isReadable,
142+
isReadableStream,
143+
isReadableEnded,
33144
isStream,
145+
isWritableStream,
146+
isWritableFinished,
34147
};

test/parallel/test-stream-finished.js

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -415,7 +415,7 @@ testClosed((opts) => new Writable({ write() {}, ...opts }));
415415
d._writableState = {};
416416
d._writableState.finished = true;
417417
finished(d, { readable: false, writable: true }, common.mustCall((err) => {
418-
assert.strictEqual(err, undefined);
418+
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
419419
}));
420420
d._writableState.errored = true;
421421
d.emit('close');

test/parallel/test-stream-pipeline.js

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1035,7 +1035,7 @@ const net = require('net');
10351035
const dst = new PassThrough();
10361036
dst.readable = false;
10371037
pipeline(src, dst, common.mustSucceed(() => {
1038-
assert.strictEqual(dst.destroyed, false);
1038+
assert.strictEqual(dst.destroyed, true);
10391039
}));
10401040
src.end();
10411041
}

0 commit comments

Comments
 (0)