src: unconsume stream fix in internal http impl

When emitting a 'connection' event on a httpServer, the function
connectionListener is called. Then, a new parser is created, and
'consume' method is called on the socket's externalStream. However,
if this stream was already consumed and unconsumed, the process
crashes with a cpp assert from the 'Consume' method in stream_base.h.
This commit makes sure that no SIGABRT will be raised and the process
will stay alive (after emitting the socket).

PR-URL: nodejs#11015
Reviewed-By: Fedor Indutny <fedor.indutny@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

lib/_http_server.js

@@ -328,9 +328,11 @@ function connectionListener(socket) {
   // Override on to unconsume on `data`, `readable` listeners
   socket.on = socketOnWrap;
 
+  // We only consume the socket if it has never been consumed before.
   var external = socket._handle._externalStream;
-  if (external) {
+  if (!socket._handle._consumed && external) {
     parser._consumed = true;
+    socket._handle._consumed = true;
     parser.consume(external);
   }
   parser[kOnExecute] =

test/parallel/test-http-server-unconsume-consume.js

@@ -0,0 +1,22 @@
+'use strict';
+const common = require('../common');
+const http = require('http');
+
+const testServer = http.createServer((req, res) => {
+  common.fail('Should not be called');
+  res.end();
+});
+testServer.on('connect', common.mustCall((req, socket, head) => {
+  socket.write('HTTP/1.1 200 Connection Established' + '\r\n' +
+      'Proxy-agent: Node-Proxy' + '\r\n' +
+      '\r\n');
+  // This shouldn't raise an assertion in StreamBase::Consume.
+  testServer.emit('connection', socket);
+  testServer.close();
+}));
+testServer.listen(0, common.mustCall(() => {
+  http.request({
+      port: testServer.address().port,
+      method: 'CONNECT'
+  }, (res) => {}).end();
+}));