net: add support for finished after .destroy()

Calling `finished(socket, cb)` would previously not
invoked the callback if the socket was already detroyed.

PR-URL: #36635
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Rich Trott <[email protected]>
Reviewed-By: James M Snell <[email protected]>

Author: ronag

lib/_http_incoming.js

@@ -174,12 +174,8 @@ IncomingMessage.prototype._destroy = function _destroy(err, cb) {
     this.emit('aborted');
   }
 
-  // If aborted and the underlying socket is not already destroyed,
-  // destroy it.
-  // We have to check if the socket is already destroyed because finished
-  // does not call the callback when this methdod is invoked from `_http_client`
-  // in `test/parallel/test-http-client-spurious-aborted.js`
-  if (this.socket && !this.socket.destroyed && this.aborted) {
+  // If aborted destroy socket.
+  if (this.socket && this.aborted) {
     this.socket.destroy(err);
     const cleanup = finished(this.socket, (e) => {
       cleanup();

lib/internal/streams/end-of-stream.js

@@ -9,6 +9,14 @@ const {
 } = require('internal/errors').codes;
 const { once } = require('internal/util');
 
+function isSocket(stream) {
+  return (
+    typeof stream.connect === 'function' &&
+    typeof stream.setNoDelay === 'function' &&
+    typeof stream.setKeepAlive === 'function'
+  );
+}
+
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
@@ -75,7 +83,7 @@ function eos(stream, options, callback) {
   let willEmitClose = (
     state &&
     state.autoDestroy &&
-    state.emitClose &&
+    (state.emitClose || isSocket(stream)) &&
     state.closed === false &&
     isReadable(stream) === readable &&
     isWritable(stream) === writable

lib/net.js

@@ -664,6 +664,12 @@ Socket.prototype._destroy = function(exception, cb) {
 
     this._handle.close(() => {
       debug('emit close');
+      if (this._writableState) {
+        this._writableState.closed = true;
+      }
+      if (this._readableState) {
+        this._readableState.closed = true;
+      }
       this.emit('close', isException);
     });
     this._handle.onread = noop;
@@ -1651,6 +1657,12 @@ Server.prototype._emitCloseIfDrained = function() {
 
 function emitCloseNT(self) {
   debug('SERVER: emit close');
+  if (self._writableState) {
+    self._writableState.closed = true;
+  }
+  if (self._readableState) {
+    self._readableState.closed = true;
+  }
   self.emit('close');
 }
 

test/parallel/test-net-finished.js

@@ -0,0 +1,25 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+const { finished } = require('stream');
+
+const server = net.createServer(function(con) {
+  con.on('close', common.mustCall());
+});
+
+server.listen(0, common.mustCall(function() {
+  const con = net.connect({
+    port: this.address().port
+  })
+  .on('connect', () => {
+    finished(con, common.mustCall());
+    con.destroy();
+    finished(con, common.mustCall());
+    con.on('close', common.mustCall(() => {
+      finished(con, common.mustCall(() => {
+        server.close();
+      }));
+    }));
+  })
+  .end();
+}));