Skip to content

Commit f4c1935

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

File tree

6 files changed

+182
-84
lines changed

6 files changed

+182
-84
lines changed

lib/internal/streams/destroy.js

+6-16
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

+22-54
Original file line numberDiff line numberDiff line change
@@ -17,48 +17,23 @@ const {
1717
validateObject,
1818
} = require('internal/validators');
1919

20+
const {
21+
isClosed,
22+
isReadable,
23+
isReadableStream,
24+
isReadableEnded,
25+
isWritable,
26+
isWritableStream,
27+
isWritableFinished,
28+
willEmitClose: _willEmitClose,
29+
} = require('internal/streams/utils');
30+
2031
function isRequest(stream) {
2132
return stream.setHeader && typeof stream.abort === 'function';
2233
}
2334

24-
function isServerResponse(stream) {
25-
return (
26-
typeof stream._sent100 === 'boolean' &&
27-
typeof stream._removedConnection === 'boolean' &&
28-
typeof stream._removedContLen === 'boolean' &&
29-
typeof stream._removedTE === 'boolean' &&
30-
typeof stream._closed === 'boolean'
31-
);
32-
}
33-
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-
5335
const nop = () => {};
5436

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-
6237
function eos(stream, options, callback) {
6338
if (arguments.length === 2) {
6439
callback = options;
@@ -74,13 +49,12 @@ function eos(stream, options, callback) {
7449
callback = once(callback);
7550

7651
const readable = options.readable ||
77-
(options.readable !== false && isReadable(stream));
52+
(options.readable !== false && isReadableStream(stream));
7853
const writable = options.writable ||
79-
(options.writable !== false && isWritable(stream));
54+
(options.writable !== false && isWritableStream(stream));
8055

8156
const wState = stream._writableState;
8257
const rState = stream._readableState;
83-
const state = wState || rState;
8458

8559
const onlegacyfinish = () => {
8660
if (!stream.writable) onfinish();
@@ -89,16 +63,13 @@ function eos(stream, options, callback) {
8963
// TODO (ronag): Improve soft detection to include core modules and
9064
// common ecosystem modules that do properly emit 'close' but fail
9165
// this generic check.
92-
let willEmitClose = isServerResponse(stream) || (
93-
state &&
94-
state.autoDestroy &&
95-
state.emitClose &&
96-
state.closed === false &&
97-
isReadable(stream) === readable &&
98-
isWritable(stream) === writable
66+
let willEmitClose = (
67+
_willEmitClose(stream) &&
68+
isReadableStream(stream) === readable &&
69+
isWritableStream(stream) === writable
9970
);
10071

101-
let writableFinished = stream.writableFinished || wState?.finished;
72+
let writableFinished = isWritableFinished(stream);
10273
const onfinish = () => {
10374
writableFinished = true;
10475
// Stream should not be destroyed here. If it is that
@@ -110,7 +81,7 @@ function eos(stream, options, callback) {
11081
if (!readable || readableEnded) callback.call(stream);
11182
};
11283

113-
let readableEnded = stream.readableEnded || rState?.endEmitted;
84+
let readableEnded = isReadableEnded(stream);
11485
const onend = () => {
11586
readableEnded = true;
11687
// Stream should not be destroyed here. If it is that
@@ -126,7 +97,7 @@ function eos(stream, options, callback) {
12697
callback.call(stream, err);
12798
};
12899

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

131102
const onclose = () => {
132103
closed = true;
@@ -185,19 +156,16 @@ function eos(stream, options, callback) {
185156
}
186157
} else if (
187158
!readable &&
188-
(!willEmitClose || stream.readable) &&
159+
(!willEmitClose || isReadable(stream)) &&
189160
writableFinished
190161
) {
191162
process.nextTick(onclose);
192163
} else if (
193164
!writable &&
194-
(!willEmitClose || stream.writable) &&
165+
(!willEmitClose || isWritable(stream)) &&
195166
readableEnded
196167
) {
197168
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);
201169
} else if ((rState && stream.req && stream.aborted)) {
202170
process.nextTick(onclose);
203171
}

lib/internal/streams/pipeline.js

+4-4
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

0 commit comments

Comments
 (0)