http: use autoDestroy: true in incoming message

Enable the default `autoDestroy: true` option in IncomingMessage.

Refactor `_http_client` and `_http_server` to remove any manual
destroying/closing of IncomingMessage.
Refactor IncomingMessage `destroy` method to use the standard
implementation of the stream module and move the early termination
event emitting inside of it.

PR-URL: https://github.com/nodejs/node/pull/33035
Refs: https://github.com/nodejs/node/issues/30625
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

lib/_http_client.js

@@ -430,25 +430,13 @@ function socketCloseListener() {
   req.destroyed = true;
   if (res) {
     // Socket closed before we emitted 'end' below.
-    // TOOD(ronag): res.destroy(err)
     if (!res.complete) {
-      res.aborted = true;
+      res.destroy(connResetException('aborted'));
-      res.emit('aborted');
-      if (res.listenerCount('error') > 0) {
-        res.emit('error', connResetException('aborted'));
-      }
     }
     req._closed = true;
     req.emit('close');
     if (!res.aborted && res.readable) {
-      res.on('end', function() {
-        this.destroyed = true;
-        this.emit('close');
-      });
       res.push(null);
-    } else {
-      res.destroyed = true;
-      res.emit('close');
     }
   } else {
     if (!req.socket._hadError) {
@@ -697,7 +685,6 @@ function responseKeepAlive(req) {
 
   req.destroyed = true;
   if (req.res) {
-    req.res.destroyed = true;
     // Detach socket from IncomingMessage to avoid destroying the freed
     // socket in IncomingMessage.destroy().
     req.res.socket = null;
@@ -752,13 +739,10 @@ function requestOnPrefinish() {
 function emitFreeNT(req) {
   req._closed = true;
   req.emit('close');
-  if (req.res) {
-    req.res.emit('close');
-  }
-
   if (req.socket) {
     req.socket.emit('free');
   }
+
 }
 
 function tickOnSocket(req, socket) {

lib/_http_incoming.js

@@ -54,7 +54,7 @@ function IncomingMessage(socket) {
     };
   }
 
-  Stream.Readable.call(this, { autoDestroy: false, ...streamOptions });
+  Stream.Readable.call(this, streamOptions);
 
   this._readableState.readingMore = true;
 
@@ -160,19 +160,20 @@ IncomingMessage.prototype._read = function _read(n) {
     readStart(this.socket);
 };
 
-
 // It's possible that the socket will be destroyed, and removed from
 // any messages, before ever calling this.  In that case, just skip
 // it, since something else is destroying this connection anyway.
-IncomingMessage.prototype.destroy = function destroy(error) {
+IncomingMessage.prototype._destroy = function _destroy(err, cb) {
-  // TODO(ronag): Implement in terms of _destroy
+  if (!this.readableEnded || !this.complete) {
-  this.destroyed = true;
+    this.aborted = true;
-  if (this.socket)
+    this.emit('aborted');
-    this.socket.destroy(error);
+  }
-  return this;
+  if (this.socket && !this.readableEnded) {
+    this.socket.destroy(err);
+  }
+  this.listenerCount('error') > 0 ? cb(err) : cb();
 };
 
-
 IncomingMessage.prototype._addHeaderLines = _addHeaderLines;
 function _addHeaderLines(headers, n) {
   if (headers && headers.length) {

lib/_http_server.js

@@ -575,14 +575,7 @@ function socketOnClose(socket, state) {
 function abortIncoming(incoming) {
   while (incoming.length) {
     const req = incoming.shift();
-    // TODO(ronag): req.destroy(err)
+    req.destroy(connResetException('aborted'));
-    req.aborted = true;
-    req.destroyed = true;
-    req.emit('aborted');
-    if (req.listenerCount('error') > 0) {
-      req.emit('error', connResetException('aborted'));
-    }
-    req.emit('close');
   }
   // Abort socket._httpMessage ?
 }
@@ -741,14 +734,9 @@ function clearIncoming(req) {
   if (parser && parser.incoming === req) {
     if (req.readableEnded) {
       parser.incoming = null;
-      req.destroyed = true;
-      req.emit('close');
     } else {
       req.on('end', clearIncoming);
     }
-  } else {
-    req.destroyed = true;
-    req.emit('close');
   }
 }