http: revert incomingmessage autodestroy

lib/_http_client.js

@@ -430,13 +430,25 @@ function socketCloseListener() {
   req.destroyed = true;
   if (res) {
     // Socket closed before we emitted 'end' below.
+    // TOOD(ronag): res.destroy(err)
     if (!res.complete) {
-      res.destroy(connResetException('aborted'));
+      res.aborted = true;
+      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) {
@@ -685,6 +697,7 @@ 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;
@@ -739,6 +752,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');
   }

lib/_http_incoming.js

@@ -27,7 +27,7 @@ const {
   Symbol
 } = primordials;
 
-const { Readable, finished } = require('stream');
+const Stream = require('stream');
 
 const kHeaders = Symbol('kHeaders');
 const kHeadersCount = Symbol('kHeadersCount');
@@ -54,7 +54,7 @@ function IncomingMessage(socket) {
     };
   }
 
-  Readable.call(this, streamOptions);
+  Stream.Readable.call(this, { autoDestroy: false, ...streamOptions });
 
   this._readableState.readingMore = true;
 
@@ -89,8 +89,8 @@ function IncomingMessage(socket) {
   // read by the user, so there's no point continuing to handle it.
   this._dumped = false;
 }
-ObjectSetPrototypeOf(IncomingMessage.prototype, Readable.prototype);
-ObjectSetPrototypeOf(IncomingMessage, Readable);
+ObjectSetPrototypeOf(IncomingMessage.prototype, Stream.Readable.prototype);
+ObjectSetPrototypeOf(IncomingMessage, Stream.Readable);
 
 ObjectDefineProperty(IncomingMessage.prototype, 'connection', {
   get: function() {
@@ -160,31 +160,19 @@ 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(err, cb) {
-  if (!this.readableEnded || !this.complete) {
-    this.aborted = true;
-    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) {
-    this.socket.destroy(err);
-    const cleanup = finished(this.socket, (e) => {
-      cleanup();
-      onError(this, e || err, cb);
-    });
-  } else {
-    onError(this, err, cb);
-  }
+IncomingMessage.prototype.destroy = function destroy(error) {
+  // TODO(ronag): Implement in terms of _destroy
+  this.destroyed = true;
+  if (this.socket)
+    this.socket.destroy(error);
+  return this;
 };
 
+
 IncomingMessage.prototype._addHeaderLines = _addHeaderLines;
 function _addHeaderLines(headers, n) {
   if (headers && headers.length) {
@@ -361,16 +349,6 @@ IncomingMessage.prototype._dump = function _dump() {
   }
 };
 
-function onError(self, error, cb) {
-  // This is to keep backward compatible behavior.
-  // An error is emitted only if there are listeners attached to the event.
-  if (self.listenerCount('error') === 0) {
-    cb();
-  } else {
-    cb(error);
-  }
-}
-
 module.exports = {
   IncomingMessage,
   readStart,

lib/_http_server.js

@@ -575,7 +575,14 @@ function socketOnClose(socket, state) {
 function abortIncoming(incoming) {
   while (incoming.length) {
     const req = incoming.shift();
-    req.destroy(connResetException('aborted'));
+    // TODO(ronag): req.destroy(err)
+    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 ?
 }
@@ -734,9 +741,14 @@ 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');
   }
 }