http: align OutgoingMessage and ClientRequest destroy

lib/_http_client.js

@@ -29,6 +29,7 @@ const {
   ObjectAssign,
   ObjectKeys,
   ObjectSetPrototypeOf,
+  Symbol
 } = primordials;
 
 const net = require('net');
@@ -65,6 +66,7 @@ const {
 } = require('internal/dtrace');
 
 const INVALID_PATH_REGEX = /[^\u0021-\u00ff]/;
+const kOnSocket = Symbol('kOnSocket');
 
 function validateHost(host, name) {
   if (host !== null && host !== undefined && typeof host !== 'string') {
@@ -337,25 +339,47 @@ ClientRequest.prototype._implicitHeader = function _implicitHeader() {
 };
 
 ClientRequest.prototype.abort = function abort() {
-  if (!this.aborted) {
-    process.nextTick(emitAbortNT, this);
+  if (this.aborted) {
+    return;
   }
   this.aborted = true;
+  process.nextTick(emitAbortNT, this);
+  this.destroy();
+};
+
+ClientRequest.prototype.destroy = function destroy(err) {
+  if (this.destroyed) {
+    return;
+  }
+  this.destroyed = true;
 
   // If we're aborting, we don't care about any more response data.
   if (this.res) {
     this.res._dump();
   }
 
-  // In the event that we don't have a socket, we will pop out of
-  // the request queue through handling in onSocket.
-  if (this.socket) {
-    // in-progress
-    this.socket.destroy();
+  if (!this.socket) {
+    this.once(kOnSocket, (socket) => {
+      if (this.agent) {
+        socket.emit('free');
+      } else {
+        socket.destroy();
+      }
+
+      if (!this.aborted && !err) {
+        err = connResetException('socket hang up');
+      }
+
+      if (err) {
+        this.emit('error', err);
+      }
+      this.emit('close');
+    });
+  } else {
+    this.socket.destroy(err);
   }
 };
 
-
 function emitAbortNT(req) {
   req.emit('abort');
 }
@@ -425,7 +449,9 @@ function socketErrorListener(err) {
   const req = socket._httpMessage;
   debug('SOCKET ERROR:', err.message, err.stack);
 
-  if (req) {
+  // If writableFinished then the error came from the readable/response
+  // side and will be emitted there.
+  if (req && !req.writableFinished) {
     // For Safety. Some additional errors might fire later on
     // and we need to make sure we don't double-fire the error event.
     req.socket._hadError = true;
@@ -686,6 +712,12 @@ function emitFreeNT(req) {
 }
 
 function tickOnSocket(req, socket) {
+  req.emit(kOnSocket, socket);
+
+  if (req.destroyed) {
+    return;
+  }
+
   const parser = parsers.alloc();
   req.socket = socket;
   parser.initialize(HTTPParser.RESPONSE,
@@ -746,23 +778,9 @@ function listenSocketTimeout(req) {
 }
 
 ClientRequest.prototype.onSocket = function onSocket(socket) {
-  process.nextTick(onSocketNT, this, socket);
+  process.nextTick(tickOnSocket, this, socket);
 };
 
-function onSocketNT(req, socket) {
-  if (req.aborted) {
-    // If we were aborted while waiting for a socket, skip the whole thing.
-    if (!req.agent) {
-      socket.destroy();
-    } else {
-      req.emit('close');
-      socket.emit('free');
-    }
-  } else {
-    tickOnSocket(req, socket);
-  }
-}
-
 ClientRequest.prototype._deferToConnect = _deferToConnect;
 function _deferToConnect(method, arguments_, cb) {
   // This function is for calls that need to happen once the socket is

lib/_http_outgoing.js

@@ -93,6 +93,7 @@ function OutgoingMessage() {
   this.outputSize = 0;
 
   this.writable = true;
+  this.destroyed = false;
 
   this._last = false;
   this.chunkedEncoding = false;
@@ -277,6 +278,11 @@ OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback) {
 // any messages, before ever calling this.  In that case, just skip
 // it, since something else is destroying this connection anyway.
 OutgoingMessage.prototype.destroy = function destroy(error) {
+  if (this.destroyed) {
+    return;
+  }
+  this.destroyed = true;
+
   if (this.socket) {
     this.socket.destroy(error);
   } else {

lib/internal/streams/destroy.js

@@ -163,7 +163,6 @@ function isRequest(stream) {
 
 // Normalize destroy for legacy.
 function destroyer(stream, err) {
-  // request.destroy just do .end - .abort is what we want
   if (isRequest(stream)) return stream.abort();
   if (isRequest(stream.req)) return stream.req.abort();
   if (typeof stream.destroy === 'function') return stream.destroy(err);

test/parallel/test-http-client-abort-destroy.js

@@ -0,0 +1,71 @@
+'use strict';
+const common = require('../common');
+const http = require('http');
+const assert = require('assert');
+
+{
+  // abort
+
+  const server = http.createServer(common.mustCall((req, res) => {
+    res.end('Hello');
+  }));
+
+  server.listen(0, common.mustCall(() => {
+    const options = { port: server.address().port };
+    const req = http.get(options, common.mustCall((res) => {
+      res.on('data', (data) => {
+        req.abort();
+        assert.strictEqual(req.aborted, true);
+        assert.strictEqual(req.destroyed, true);
+        server.close();
+      });
+    }));
+    req.on('error', common.mustNotCall());
+    assert.strictEqual(req.aborted, false);
+    assert.strictEqual(req.destroyed, false);
+  }));
+}
+
+{
+  // destroy + res
+
+  const server = http.createServer(common.mustCall((req, res) => {
+    res.end('Hello');
+  }));
+
+  server.listen(0, common.mustCall(() => {
+    const options = { port: server.address().port };
+    const req = http.get(options, common.mustCall((res) => {
+      res.on('data', (data) => {
+        req.destroy();
+        assert.strictEqual(req.aborted, false);
+        assert.strictEqual(req.destroyed, true);
+        server.close();
+      });
+    }));
+    req.on('error', common.mustNotCall());
+    assert.strictEqual(req.aborted, false);
+    assert.strictEqual(req.destroyed, false);
+  }));
+}
+
+
+{
+  // destroy
+
+  const server = http.createServer(common.mustNotCall((req, res) => {
+  }));
+
+  server.listen(0, common.mustCall(() => {
+    const options = { port: server.address().port };
+    const req = http.get(options, common.mustNotCall());
+    req.on('error', common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ECONNRESET');
+    }));
+    assert.strictEqual(req.aborted, false);
+    assert.strictEqual(req.destroyed, false);
+    req.destroy();
+    assert.strictEqual(req.aborted, false);
+    assert.strictEqual(req.destroyed, true);
+  }));
+}

test/parallel/test-http-client-abort-keep-alive-queued-tcp-socket.js

@@ -3,34 +3,45 @@ const common = require('../common');
 const assert = require('assert');
 const http = require('http');
 
-let socketsCreated = 0;
 
-class Agent extends http.Agent {
-  createConnection(options, oncreate) {
-    const socket = super.createConnection(options, oncreate);
-    socketsCreated++;
-    return socket;
+for (const destroyer of ['destroy', 'abort']) {
+  let socketsCreated = 0;
+
+  class Agent extends http.Agent {
+    createConnection(options, oncreate) {
+      const socket = super.createConnection(options, oncreate);
+      socketsCreated++;
+      return socket;
+    }
   }
-}
 
-const server = http.createServer((req, res) => res.end());
+  const server = http.createServer((req, res) => res.end());
 
-server.listen(0, common.mustCall(() => {
-  const port = server.address().port;
-  const agent = new Agent({
-    keepAlive: true,
-    maxSockets: 1
-  });
+  server.listen(0, common.mustCall(() => {
+    const port = server.address().port;
+    const agent = new Agent({
+      keepAlive: true,
+      maxSockets: 1
+    });
 
-  http.get({ agent, port }, (res) => res.resume());
+    http.get({ agent, port }, (res) => res.resume());
 
-  const req = http.get({ agent, port }, common.mustNotCall());
-  req.abort();
+    const req = http.get({ agent, port }, common.mustNotCall());
+    req[destroyer]();
 
-  http.get({ agent, port }, common.mustCall((res) => {
-    res.resume();
-    assert.strictEqual(socketsCreated, 1);
-    agent.destroy();
-    server.close();
+    if (destroyer === 'destroy') {
+      req.on('error', common.mustCall((err) => {
+        assert.strictEqual(err.code, 'ECONNRESET');
+      }));
+    } else {
+      req.on('error', common.mustNotCall());
+    }
+
+    http.get({ agent, port }, common.mustCall((res) => {
+      res.resume();
+      assert.strictEqual(socketsCreated, 1);
+      agent.destroy();
+      server.close();
+    }));
   }));
-}));
+}

test/parallel/test-http-client-close-event.js

@@ -14,12 +14,12 @@ server.listen(0, common.mustCall(() => {
   const req = http.get({ port: server.address().port }, common.mustNotCall());
   let errorEmitted = false;
 
-  req.on('error', (err) => {
+  req.on('error', common.mustCall((err) => {
     errorEmitted = true;
     assert.strictEqual(err.constructor, Error);
     assert.strictEqual(err.message, 'socket hang up');
     assert.strictEqual(err.code, 'ECONNRESET');
-  });
+  }));
 
   req.on('close', common.mustCall(() => {
     assert.strictEqual(errorEmitted, true);

test/parallel/test-http-outgoing-proto.js

@@ -122,3 +122,10 @@ assert.throws(() => {
   name: 'TypeError',
   message: 'Invalid character in trailer content ["404"]'
 });
+
+{
+  const outgoingMessage = new OutgoingMessage();
+  assert.strictEqual(outgoingMessage.destroyed, false);
+  outgoingMessage.destroy();
+  assert.strictEqual(outgoingMessage.destroyed, true);
+}