1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
'use strict';
function needError(stream, err) {
if (!err) {
return false;
}
const r = stream._readableState;
const w = stream._writableState;
if ((w && w.errorEmitted) || (r && r.errorEmitted)) {
return false;
}
if (w) {
w.errorEmitted = true;
}
if (r) {
r.errorEmitted = true;
}
return true;
}
// Undocumented cb() API, needed for core, not for public API
function destroy(err, cb) {
const r = this._readableState;
const w = this._writableState;
if ((w && w.destroyed) || (r && r.destroyed)) {
if (cb) {
cb(err);
} else if (needError(this, err)) {
process.nextTick(emitErrorNT, this, err);
}
return this;
}
// We set destroyed to true before firing error callbacks in order
// to make it re-entrance safe in case destroy() is called within callbacks
if (w) {
w.destroyed = true;
}
if (r) {
r.destroyed = true;
}
this._destroy(err || null, (err) => {
const emitClose = (w && w.emitClose) || (r && r.emitClose);
if (cb) {
if (emitClose) {
process.nextTick(emitCloseNT, this);
}
cb(err);
} else if (needError(this, err)) {
process.nextTick(emitClose ? emitErrorCloseNT : emitErrorNT, this, err);
} else if (emitClose) {
process.nextTick(emitCloseNT, this);
}
});
return this;
}
function emitErrorCloseNT(self, err) {
self.emit('error', err);
self.emit('close');
}
function emitCloseNT(self) {
self.emit('close');
}
function emitErrorNT(self, err) {
self.emit('error', err);
}
function undestroy() {
const r = this._readableState;
const w = this._writableState;
if (r) {
r.destroyed = false;
r.reading = false;
r.ended = false;
r.endEmitted = false;
r.errorEmitted = false;
}
if (w) {
w.destroyed = false;
w.ended = false;
w.ending = false;
w.finalCalled = false;
w.prefinished = false;
w.finished = false;
w.errorEmitted = false;
}
}
function errorOrDestroy(stream, err) {
// We have tests that rely on errors being emitted
// in the same tick, so changing this is semver major.
// For now when you opt-in to autoDestroy we allow
// the error to be emitted nextTick. In a future
// semver major update we should change the default to this.
const r = stream._readableState;
const w = stream._writableState;
if ((r && r.autoDestroy) || (w && w.autoDestroy))
stream.destroy(err);
else if (needError(stream, err))
stream.emit('error', err);
}
module.exports = {
destroy,
undestroy,
errorOrDestroy
};
|
