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 <jasnell@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Colin Ihrig <cjihrig@gmail.com>

Author: addaleax

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();
+ });
+}