http: support generic Duplex streams

Support generic `Duplex` streams through more duck typing
on the server and client sides.

Since HTTP is, as a protocol, independent of its underlying transport
layer, Node.js should not enforce any restrictions on what streams
its HTTP parser may use.

Ref: #16256
PR-URL: #16267
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Colin Ihrig <[email protected]>

doc/api/http.md

@@ -797,11 +797,14 @@ added: v0.1.0
 
 * `socket` {net.Socket}
 
-When a new TCP stream is established. `socket` is an object of type
-[`net.Socket`][]. Usually users will not want to access this event. In
-particular, the socket will not emit `'readable'` events because of how
-the protocol parser attaches to the socket. The `socket` can also be
-accessed at `request.connection`.
+This event is emitted when a new TCP stream is established. `socket` is
+typically an object of type [`net.Socket`][]. Usually users will not want to
+access this event. In particular, the socket will not emit `'readable'` events
+because of how the protocol parser attaches to the socket. The `socket` can
+also be accessed at `request.connection`.
+
+*Note*: This event can also be explicitly emitted by users to inject connections
+into the HTTP server. In that case, any [`Duplex`][] stream can be passed.
 
 ### Event: 'request'
 <!-- YAML
@@ -1769,7 +1772,7 @@ changes:
     use for the request when the `agent` option is not used. This can be used to
     avoid creating a custom `Agent` class just to override the default
     `createConnection` function. See [`agent.createConnection()`][] for more
-    details.
+    details. Any [`Duplex`][] stream is a valid return value.
   * `timeout` {number}: A number specifying the socket timeout in milliseconds.
     This will set the timeout before the socket is connected.
 * `callback` {Function}
@@ -1869,6 +1872,7 @@ const req = http.request(options, (res) => {
 [`'request'`]: #http_event_request
 [`'response'`]: #http_event_response
 [`Agent`]: #http_class_http_agent
+[`Duplex`]: stream.html#stream_class_stream_duplex
 [`EventEmitter`]: events.html#events_class_eventemitter
 [`TypeError`]: errors.html#errors_class_typeerror
 [`URL`]: url.html#url_the_whatwg_url_api

lib/_http_client.js

@@ -565,7 +565,10 @@ function responseKeepAlive(res, req) {
   if (!req.shouldKeepAlive) {
     if (socket.writable) {
       debug('AGENT socket.destroySoon()');
-      socket.destroySoon();
+      if (typeof socket.destroySoon === 'function')
+        socket.destroySoon();
+      else
+        socket.end();
     }
     assert(!socket.writable);
   } else {

lib/_http_server.js

@@ -303,7 +303,7 @@ function connectionListener(socket) {
   // If the user has added a listener to the server,
   // request, or response, then it's their responsibility.
   // otherwise, destroy on timeout by default
-  if (this.timeout)
+  if (this.timeout && typeof socket.setTimeout === 'function')
     socket.setTimeout(this.timeout);
   socket.on('timeout', socketOnTimeout);
 
@@ -354,11 +354,13 @@ function connectionListener(socket) {
   socket.on = socketOnWrap;
 
   // We only consume the socket if it has never been consumed before.
-  var external = socket._handle._externalStream;
-  if (!socket._handle._consumed && external) {
-    parser._consumed = true;
-    socket._handle._consumed = true;
-    parser.consume(external);
+  if (socket._handle) {
+    var external = socket._handle._externalStream;
+    if (!socket._handle._consumed && external) {
+      parser._consumed = true;
+      socket._handle._consumed = true;
+      parser.consume(external);
+    }
   }
   parser[kOnExecute] =
     onParserExecute.bind(undefined, this, socket, parser, state);
@@ -533,9 +535,13 @@ function resOnFinish(req, res, socket, state, server) {
   res.detachSocket(socket);
 
   if (res._last) {
-    socket.destroySoon();
+    if (typeof socket.destroySoon === 'function') {
+      socket.destroySoon();
+    } else {
+      socket.end();
+    }
   } else if (state.outgoing.length === 0) {
-    if (server.keepAliveTimeout) {
+    if (server.keepAliveTimeout && typeof socket.setTimeout === 'function') {
       socket.setTimeout(0);
       socket.setTimeout(server.keepAliveTimeout);
       state.keepAliveTimeoutSet = true;

test/parallel/test-http-generic-streams.js

@@ -0,0 +1,60 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+const MakeDuplexPair = require('../common/duplexpair');
+
+// Test 1: Simple HTTP test, no keep-alive.
+{
+  const testData = 'Hello, World!\n';
+  const server = http.createServer(common.mustCall((req, res) => {
+    res.statusCode = 200;
+    res.setHeader('Content-Type', 'text/plain');
+    res.end(testData);
+  }));
+
+  const { clientSide, serverSide } = MakeDuplexPair();
+  server.emit('connection', serverSide);
+
+  const req = http.request({
+    createConnection: common.mustCall(() => clientSide)
+  }, common.mustCall((res) => {
+    res.setEncoding('utf8');
+    res.on('data', common.mustCall((data) => {
+      assert.strictEqual(data, testData);
+    }));
+  }));
+  req.end();
+}
+
+// Test 2: Keep-alive for 2 requests.
+{
+  const testData = 'Hello, World!\n';
+  const server = http.createServer(common.mustCall((req, res) => {
+    res.statusCode = 200;
+    res.setHeader('Content-Type', 'text/plain');
+    res.end(testData);
+  }, 2));
+
+  const { clientSide, serverSide } = MakeDuplexPair();
+  server.emit('connection', serverSide);
+
+  function doRequest(cb) {
+    const req = http.request({
+      createConnection: common.mustCall(() => clientSide),
+      headers: { Connection: 'keep-alive' }
+    }, common.mustCall((res) => {
+      res.setEncoding('utf8');
+      res.on('data', common.mustCall((data) => {
+        assert.strictEqual(data, testData);
+      }));
+      res.on('end', common.mustCall(cb));
+    }));
+    req.shouldKeepAlive = true;
+    req.end();
+  }
+
+  doRequest(() => {
+    doRequest();
+  });
+}