Skip to content

Commit f2f6872

Browse files
ronagronag
committed
stream: cleanup async handling
Cleanup async stream method handling. PR-URL: #39329 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 7a7ba82 commit f2f6872

4 files changed

Lines changed: 63 additions & 113 deletions

File tree

‎lib/internal/streams/destroy.js‎

Lines changed: 36 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -76,12 +76,16 @@ function destroy(err, cb) {
7676

7777
function _destroy(self, err, cb) {
7878
let called = false;
79-
const result = self._destroy(err || null, (err) => {
80-
const r = self._readableState;
81-
const w = self._writableState;
8279

80+
function onDestroy(err) {
81+
if (called) {
82+
return;
83+
}
8384
called = true;
8485

86+
const r = self._readableState;
87+
const w = self._writableState;
88+
8589
checkError(err, w, r);
8690

8791
if (w) {
@@ -100,64 +104,24 @@ function _destroy(self, err, cb) {
100104
} else {
101105
process.nextTick(emitCloseNT, self);
102106
}
103-
});
104-
if (result !== undefined && result !== null) {
105-
try {
107+
}
108+
try {
109+
const result = self._destroy(err || null, onDestroy);
110+
if (result != null) {
106111
const then = result.then;
107112
if (typeof then === 'function') {
108113
then.call(
109114
result,
110115
function() {
111-
if (called)
112-
return;
113-
114-
const r = self._readableState;
115-
const w = self._writableState;
116-
117-
if (w) {
118-
w.closed = true;
119-
}
120-
if (r) {
121-
r.closed = true;
122-
}
123-
124-
if (typeof cb === 'function') {
125-
process.nextTick(cb);
126-
}
127-
128-
process.nextTick(emitCloseNT, self);
116+
process.nextTick(onDestroy, null);
129117
},
130118
function(err) {
131-
const r = self._readableState;
132-
const w = self._writableState;
133-
err.stack; // eslint-disable-line no-unused-expressions
134-
135-
called = true;
136-
137-
if (w && !w.errored) {
138-
w.errored = err;
139-
}
140-
if (r && !r.errored) {
141-
r.errored = err;
142-
}
143-
144-
if (w) {
145-
w.closed = true;
146-
}
147-
if (r) {
148-
r.closed = true;
149-
}
150-
151-
if (typeof cb === 'function') {
152-
process.nextTick(cb, err);
153-
}
154-
155-
process.nextTick(emitErrorCloseNT, self, err);
119+
process.nextTick(onDestroy, err);
156120
});
157121
}
158-
} catch (err) {
159-
process.nextTick(emitErrorNT, self, err);
160122
}
123+
} catch (err) {
124+
onDestroy(err);
161125
}
162126
}
163127

@@ -291,74 +255,52 @@ function construct(stream, cb) {
291255
}
292256

293257
function constructNT(stream) {
294-
const r = stream._readableState;
295-
const w = stream._writableState;
296-
// With duplex streams we use the writable side for state.
297-
const s = w || r;
298-
299258
let called = false;
300-
const result = stream._construct((err) => {
259+
260+
function onConstruct(err) {
261+
if (called) {
262+
errorOrDestroy(stream, err ?? new ERR_MULTIPLE_CALLBACK());
263+
return;
264+
}
265+
called = true;
266+
267+
const r = stream._readableState;
268+
const w = stream._writableState;
269+
const s = w || r;
270+
301271
if (r) {
302272
r.constructed = true;
303273
}
304274
if (w) {
305275
w.constructed = true;
306276
}
307277

308-
if (called) {
309-
err = new ERR_MULTIPLE_CALLBACK();
310-
} else {
311-
called = true;
312-
}
313-
314278
if (s.destroyed) {
315279
stream.emit(kDestroy, err);
316280
} else if (err) {
317281
errorOrDestroy(stream, err, true);
318282
} else {
319283
process.nextTick(emitConstructNT, stream);
320284
}
321-
});
322-
if (result !== undefined && result !== null) {
323-
try {
285+
}
286+
287+
try {
288+
const result = stream._construct(onConstruct);
289+
if (result != null) {
324290
const then = result.then;
325291
if (typeof then === 'function') {
326292
then.call(
327293
result,
328294
function() {
329-
// If the callback was invoked, do nothing further.
330-
if (called)
331-
return;
332-
if (r) {
333-
r.constructed = true;
334-
}
335-
if (w) {
336-
w.constructed = true;
337-
}
338-
if (s.destroyed) {
339-
process.nextTick(() => stream.emit(kDestroy));
340-
} else {
341-
process.nextTick(emitConstructNT, stream);
342-
}
295+
process.nextTick(onConstruct, null);
343296
},
344297
function(err) {
345-
if (r) {
346-
r.constructed = true;
347-
}
348-
if (w) {
349-
w.constructed = true;
350-
}
351-
called = true;
352-
if (s.destroyed) {
353-
process.nextTick(() => stream.emit(kDestroy, err));
354-
} else {
355-
process.nextTick(errorOrDestroy, stream, err);
356-
}
298+
process.nextTick(onConstruct, err);
357299
});
358300
}
359-
} catch (err) {
360-
process.nextTick(emitErrorNT, stream, err);
361301
}
302+
} catch (err) {
303+
onConstruct(err);
362304
}
363305
}
364306

‎lib/internal/streams/readable.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -483,8 +483,10 @@ Readable.prototype.read = function(n) {
483483
// If the length is currently zero, then we *need* a readable event.
484484
if (state.length === 0)
485485
state.needReadable = true;
486+
486487
// Call internal read method
487488
this._read(state.highWaterMark);
489+
488490
state.sync = false;
489491
// If _read pushed data synchronously, then `reading` will be false,
490492
// and we need to re-evaluate how much data we can return to the user.

‎lib/internal/streams/writable.js‎

Lines changed: 22 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -660,9 +660,15 @@ function needFinish(state) {
660660
}
661661

662662
function callFinal(stream, state) {
663-
state.sync = true;
664-
state.pendingcb++;
665-
const result = stream._final((err) => {
663+
let called = false;
664+
665+
function onFinish(err) {
666+
if (called) {
667+
errorOrDestroy(stream, err ?? ERR_MULTIPLE_CALLBACK());
668+
return;
669+
}
670+
called = true;
671+
666672
state.pendingcb--;
667673
if (err) {
668674
const onfinishCallbacks = state[kOnFinished].splice(0);
@@ -679,33 +685,30 @@ function callFinal(stream, state) {
679685
state.pendingcb++;
680686
process.nextTick(finish, stream, state);
681687
}
682-
});
683-
if (result !== undefined && result !== null) {
684-
try {
688+
}
689+
690+
state.sync = true;
691+
state.pendingcb++;
692+
693+
try {
694+
const result = stream._final(onFinish);
695+
if (result != null) {
685696
const then = result.then;
686697
if (typeof then === 'function') {
687698
then.call(
688699
result,
689700
function() {
690-
if (state.prefinished || !needFinish(state))
691-
return;
692-
state.prefinish = true;
693-
process.nextTick(() => stream.emit('prefinish'));
694-
state.pendingcb++;
695-
process.nextTick(finish, stream, state);
701+
process.nextTick(onFinish, null);
696702
},
697703
function(err) {
698-
const onfinishCallbacks = state[kOnFinished].splice(0);
699-
for (let i = 0; i < onfinishCallbacks.length; i++) {
700-
process.nextTick(onfinishCallbacks[i], err);
701-
}
702-
process.nextTick(errorOrDestroy, stream, err, state.sync);
704+
process.nextTick(onFinish, err);
703705
});
704706
}
705-
} catch (err) {
706-
process.nextTick(errorOrDestroy, stream, err, state.sync);
707707
}
708+
} catch (err) {
709+
onFinish(stream, state, err);
708710
}
711+
709712
state.sync = false;
710713
}
711714

‎test/parallel/test-stream-construct-async-error.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,9 @@ const assert = require('assert');
9898

9999
const foo = new Foo();
100100
foo.write('test', common.mustCall());
101+
foo.on('error', common.expectsError({
102+
code: 'ERR_MULTIPLE_CALLBACK'
103+
}));
101104
}
102105

103106
{

0 commit comments

Comments
 (0)
⚡