Skip to content

Commit 388cef6

Browse files
committed
stream: align stream.Duplex with net.Socket
stream.Duplex and net.Socket slightly differs in behavior. Especially when it comes to the case where one side never becomes readable or writable. This aligns Duplex with the behavior of Socket. PR-URL: #32139 Reviewed-By: Anto Aravinth <anto.aravinth.cse@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 05f1df5 commit 388cef6

File tree

5 files changed

+73
-22
lines changed

5 files changed

+73
-22
lines changed

lib/_stream_duplex.js

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,6 @@ function Duplex(options) {
6666

6767
if (options.allowHalfOpen === false) {
6868
this.allowHalfOpen = false;
69-
this.once('end', onend);
7069
}
7170
}
7271
}
@@ -128,18 +127,3 @@ ObjectDefineProperties(Duplex.prototype, {
128127
}
129128
}
130129
});
131-
132-
// The no-half-open enforcer
133-
function onend() {
134-
// If the writable side ended, then we're ok.
135-
if (this._writableState.ended)
136-
return;
137-
138-
// No more data can be written.
139-
// But allow more writes to happen in this tick.
140-
process.nextTick(onEndNT, this);
141-
}
142-
143-
function onEndNT(self) {
144-
self.end();
145-
}

lib/_stream_readable.js

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1217,17 +1217,34 @@ function endReadableNT(state, stream) {
12171217
state.endEmitted = true;
12181218
stream.emit('end');
12191219

1220-
if (state.autoDestroy) {
1220+
if (stream.writable && stream.allowHalfOpen === false) {
1221+
process.nextTick(endWritableNT, state, stream);
1222+
} else if (state.autoDestroy) {
12211223
// In case of duplex streams we need a way to detect
12221224
// if the writable side is ready for autoDestroy as well
12231225
const wState = stream._writableState;
1224-
if (!wState || (wState.autoDestroy && wState.finished)) {
1226+
const autoDestroy = !wState || (
1227+
wState.autoDestroy &&
1228+
// We don't expect the writable to ever 'finish'
1229+
// if writable is explicitly set to false.
1230+
(wState.finished || wState.writable === false)
1231+
);
1232+
1233+
if (autoDestroy) {
12251234
stream.destroy();
12261235
}
12271236
}
12281237
}
12291238
}
12301239

1240+
function endWritableNT(state, stream) {
1241+
const writable = stream.writable && !stream.writableEnded &&
1242+
!stream.destroyed;
1243+
if (writable) {
1244+
stream.end();
1245+
}
1246+
}
1247+
12311248
Readable.from = function(iterable, opts) {
12321249
if (from === undefined) {
12331250
from = require('internal/streams/from');

lib/_stream_writable.js

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -675,7 +675,13 @@ function finish(stream, state) {
675675
// In case of duplex streams we need a way to detect
676676
// if the readable side is ready for autoDestroy as well
677677
const rState = stream._readableState;
678-
if (!rState || (rState.autoDestroy && rState.endEmitted)) {
678+
const autoDestroy = !rState || (
679+
rState.autoDestroy &&
680+
// We don't expect the readable to ever 'end'
681+
// if readable is explicitly set to false.
682+
(rState.endEmitted || rState.readable === false)
683+
);
684+
if (autoDestroy) {
679685
stream.destroy();
680686
}
681687
}
@@ -748,7 +754,7 @@ ObjectDefineProperties(Writable.prototype, {
748754
// Compat. The user might manually disable writable side through
749755
// deprecated setter.
750756
return !!w && w.writable !== false && !w.destroyed && !w.errored &&
751-
!w.ending;
757+
!w.ending && !w.ended;
752758
},
753759
set(val) {
754760
// Backwards compatible.

test/parallel/test-stream-duplex-destroy.js

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,3 +194,47 @@ const assert = require('assert');
194194

195195
new MyDuplex();
196196
}
197+
198+
{
199+
const duplex = new Duplex({
200+
writable: false,
201+
autoDestroy: true,
202+
write(chunk, enc, cb) { cb(); },
203+
read() {},
204+
});
205+
duplex.push(null);
206+
duplex.resume();
207+
duplex.on('close', common.mustCall());
208+
}
209+
210+
{
211+
const duplex = new Duplex({
212+
readable: false,
213+
autoDestroy: true,
214+
write(chunk, enc, cb) { cb(); },
215+
read() {},
216+
});
217+
duplex.end();
218+
duplex.on('close', common.mustCall());
219+
}
220+
221+
{
222+
const duplex = new Duplex({
223+
allowHalfOpen: false,
224+
autoDestroy: true,
225+
write(chunk, enc, cb) { cb(); },
226+
read() {},
227+
});
228+
duplex.push(null);
229+
duplex.resume();
230+
const orgEnd = duplex.end;
231+
duplex.end = common.mustNotCall();
232+
duplex.on('end', () => {
233+
// Ensure end() is called in next tick to allow
234+
// any pending writes to be invoked first.
235+
process.nextTick(() => {
236+
duplex.end = common.mustCall(orgEnd);
237+
});
238+
});
239+
duplex.on('close', common.mustCall());
240+
}

test/parallel/test-stream-duplex-end.js

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ const Duplex = require('stream').Duplex;
2222
});
2323
assert.strictEqual(stream.allowHalfOpen, false);
2424
stream.on('finish', common.mustCall());
25-
assert.strictEqual(stream.listenerCount('end'), 1);
25+
assert.strictEqual(stream.listenerCount('end'), 0);
2626
stream.resume();
2727
stream.push(null);
2828
}
@@ -35,7 +35,7 @@ const Duplex = require('stream').Duplex;
3535
assert.strictEqual(stream.allowHalfOpen, false);
3636
stream._writableState.ended = true;
3737
stream.on('finish', common.mustNotCall());
38-
assert.strictEqual(stream.listenerCount('end'), 1);
38+
assert.strictEqual(stream.listenerCount('end'), 0);
3939
stream.resume();
4040
stream.push(null);
4141
}

0 commit comments

Comments
 (0)