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: https://github.com/nodejs/node/pull/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>

3 files changed, 27 insertions, 20 deletions

diff --git a/lib/_stream_duplex.js b/lib/_stream_duplex.js
index b832c973a1..fe2281df47 100644
--- a/lib/_stream_duplex.js
+++ b/lib/_stream_duplex.js
@@ -66,7 +66,6 @@ function Duplex(options) {
 
     if (options.allowHalfOpen === false) {
       this.allowHalfOpen = false;
-      this.once('end', onend);
     }
   }
 }
@@ -128,18 +127,3 @@ ObjectDefineProperties(Duplex.prototype, {
     }
   }
 });
-
-// The no-half-open enforcer
-function onend() {
-  // If the writable side ended, then we're ok.
-  if (this._writableState.ended)
-    return;
-
-  // No more data can be written.
-  // But allow more writes to happen in this tick.
-  process.nextTick(onEndNT, this);
-}
-
-function onEndNT(self) {
-  self.end();
-}
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index dfbd023d24..1df50ba200 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -1217,17 +1217,34 @@ function endReadableNT(state, stream) {
     state.endEmitted = true;
     stream.emit('end');
 
-    if (state.autoDestroy) {
+    if (stream.writable && stream.allowHalfOpen === false) {
+      process.nextTick(endWritableNT, state, stream);
+    } else if (state.autoDestroy) {
       // In case of duplex streams we need a way to detect
       // if the writable side is ready for autoDestroy as well
       const wState = stream._writableState;
-      if (!wState || (wState.autoDestroy && wState.finished)) {
+      const autoDestroy = !wState || (
+        wState.autoDestroy &&
+        // We don't expect the writable to ever 'finish'
+        // if writable is explicitly set to false.
+        (wState.finished || wState.writable === false)
+      );
+
+      if (autoDestroy) {
         stream.destroy();
       }
     }
   }
 }
 
+function endWritableNT(state, stream) {
+  const writable = stream.writable && !stream.writableEnded &&
+    !stream.destroyed;
+  if (writable) {
+    stream.end();
+  }
+}
+
 Readable.from = function(iterable, opts) {
   if (from === undefined) {
     from = require('internal/streams/from');
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index b24192101c..c3a7a35d2b 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -675,7 +675,13 @@ function finish(stream, state) {
     // In case of duplex streams we need a way to detect
     // if the readable side is ready for autoDestroy as well
     const rState = stream._readableState;
-    if (!rState || (rState.autoDestroy && rState.endEmitted)) {
+    const autoDestroy = !rState || (
+      rState.autoDestroy &&
+      // We don't expect the readable to ever 'end'
+      // if readable is explicitly set to false.
+      (rState.endEmitted || rState.readable === false)
+    );
+    if (autoDestroy) {
       stream.destroy();
     }
   }
@@ -748,7 +754,7 @@ ObjectDefineProperties(Writable.prototype, {
       // Compat. The user might manually disable writable side through
       // deprecated setter.
       return !!w && w.writable !== false && !w.destroyed && !w.errored &&
-        !w.ending;
+        !w.ending && !w.ended;
     },
     set(val) {
       // Backwards compatible.