http: support generic Duplex streams

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/common/README.md

@@ -8,6 +8,7 @@ This directory contains modules used to test the Node.js implementation.
 * [Common module API](#common-module-api)
 * [Countdown module](#countdown-module)
 * [DNS module](#dns-module)
+* [Duplex pair helper](#duplex-pair-helper)
 * [Fixtures module](#fixtures-module)
 * [WPT module](#wpt-module)
 
@@ -458,6 +459,14 @@ Reads a Domain String and returns a Buffer containing the domain.
 Takes in a parsed Object and writes its fields to a DNS packet as a Buffer
 object.
 
+## Duplex pair helper
+
+The `common/duplexpair` module exports a single function `makeDuplexPair`,
+which returns an object `{ clientSide, serverSide }` where each side is a
+`Duplex` stream connected to the other side.
+
+There is no difference between client or server side beyond their names.
+
 ## Fixtures Module
 
 The `common/fixtures` module provides convenience methods for working with

test/common/duplexpair.js

@@ -0,0 +1,45 @@
+/* eslint-disable required-modules */
+'use strict';
+const { Duplex } = require('stream');
+const assert = require('assert');
+
+const kCallback = Symbol('Callback');
+const kOtherSide = Symbol('Other');
+
+class DuplexSocket extends Duplex {
+ constructor() {
+ super();
+ this[kCallback] = null;
+ this[kOtherSide] = null;
+ }
+
+ _read() {
+ const callback = this[kCallback];
+ if (callback) {
+ this[kCallback] = null;
+ callback();
+ }
+ }
+
+ _write(chunk, encoding, callback) {
+ assert.notStrictEqual(this[kOtherSide], null);
+ assert.strictEqual(this[kOtherSide][kCallback], null);
+ this[kOtherSide][kCallback] = callback;
+ this[kOtherSide].push(chunk);
+ }
+
+ _final(callback) {
+ this[kOtherSide].on('end', callback);
+ this[kOtherSide].push(null);
+ }
+}
+
+function makeDuplexPair() {
+ const clientSide = new DuplexSocket();
+ const serverSide = new DuplexSocket();
+ clientSide[kOtherSide] = serverSide;
+ serverSide[kOtherSide] = clientSide;
+ return { clientSide, serverSide };
+}
+
+module.exports = makeDuplexPair;

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