http2: support generic Duplex streams

Support generic `Duplex` streams through using `StreamWrap`
on the server and client sides, and adding a `createConnection`
method option similar to what the HTTP/1 API provides.

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

Ref: #16256
PR-URL: #16269
Reviewed-By: Anatoli Papirovski <apapirovski@mac.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/http2.md

@@ -1598,6 +1598,9 @@ added: v8.4.0
     used to determine the padding. See [Using options.selectPadding][].
   * `settings` {[Settings Object][]} The initial settings to send to the
     remote peer upon connection.
+  * `createConnection` {Function} An optional callback that receives the `URL`
+    instance passed to `connect` and the `options` object, and returns any
+    [`Duplex`][] stream that is to be used as the connection for this session.
 * `listener` {Function}
 * Returns {Http2Session}
 

lib/internal/http2/core.js

@@ -13,6 +13,7 @@ const tls = require('tls');
 const util = require('util');
 const fs = require('fs');
 const errors = require('internal/errors');
+const { StreamWrap } = require('_stream_wrap');
 const { Duplex } = require('stream');
 const { URL } = require('url');
 const { onServerStream,
@@ -695,10 +696,14 @@ class Http2Session extends EventEmitter {
 
   // type     { number } either NGHTTP2_SESSION_SERVER or NGHTTP2_SESSION_CLIENT
   // options  { Object }
-  // socket   { net.Socket | tls.TLSSocket }
+  // socket   { net.Socket | tls.TLSSocket | stream.Duplex }
   constructor(type, options, socket) {
     super();
 
+    if (!socket._handle || !socket._handle._externalStream) {
+      socket = new StreamWrap(socket);
+    }
+
     // No validation is performed on the input parameters because this
     // constructor is not exported directly for users.
 
@@ -723,7 +728,8 @@ class Http2Session extends EventEmitter {
     this[kSocket] = socket;
 
     // Do not use nagle's algorithm
-    socket.setNoDelay();
+    if (typeof socket.setNoDelay === 'function')
+      socket.setNoDelay();
 
     // Disable TLS renegotiation on the socket
     if (typeof socket.disableRenegotiation === 'function')
@@ -2429,15 +2435,19 @@ function connect(authority, options, listener) {
   const host = authority.hostname || authority.host || 'localhost';
 
   let socket;
-  switch (protocol) {
-    case 'http:':
-      socket = net.connect(port, host);
-      break;
-    case 'https:':
-      socket = tls.connect(port, host, initializeTLSOptions(options, host));
-      break;
-    default:
-      throw new errors.Error('ERR_HTTP2_UNSUPPORTED_PROTOCOL', protocol);
+  if (typeof options.createConnection === 'function') {
+    socket = options.createConnection(authority, options);
+  } else {
+    switch (protocol) {
+      case 'http:':
+        socket = net.connect(port, host);
+        break;
+      case 'https:':
+        socket = tls.connect(port, host, initializeTLSOptions(options, host));
+        break;
+      default:
+        throw new errors.Error('ERR_HTTP2_UNSUPPORTED_PROTOCOL', protocol);
+    }
   }
 
   socket.on('error', socketOnError);

test/parallel/test-http2-generic-streams-sendfile.js

@@ -0,0 +1,40 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const fs = require('fs');
+const makeDuplexPair = require('../common/duplexpair');
+
+{
+  const server = http2.createServer();
+  server.on('stream', common.mustCall((stream, headers) => {
+    stream.respondWithFile(__filename);
+  }));
+
+  const { clientSide, serverSide } = makeDuplexPair();
+  server.emit('connection', serverSide);
+
+  const client = http2.connect('http://localhost:80', {
+    createConnection: common.mustCall(() => clientSide)
+  });
+
+  const req = client.request({ ':path': '/' });
+
+  req.on('response', common.mustCall((headers) => {
+    assert.strictEqual(headers[':status'], 200);
+  }));
+
+  req.setEncoding('utf8');
+  let data = '';
+  req.on('data', (chunk) => {
+    data += chunk;
+  });
+  req.on('end', common.mustCall(() => {
+    assert.strictEqual(data, fs.readFileSync(__filename, 'utf8'));
+    clientSide.destroy();
+    clientSide.end();
+  }));
+  req.end();
+}

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

@@ -0,0 +1,45 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const makeDuplexPair = require('../common/duplexpair');
+
+{
+  const testData = '<h1>Hello World</h1>';
+  const server = http2.createServer();
+  server.on('stream', common.mustCall((stream, headers) => {
+    stream.respond({
+      'content-type': 'text/html',
+      ':status': 200
+    });
+    stream.end(testData);
+  }));
+
+  const { clientSide, serverSide } = makeDuplexPair();
+  server.emit('connection', serverSide);
+
+  const client = http2.connect('http://localhost:80', {
+    createConnection: common.mustCall(() => clientSide)
+  });
+
+  const req = client.request({ ':path': '/' });
+
+  req.on('response', common.mustCall((headers) => {
+    assert.strictEqual(headers[':status'], 200);
+  }));
+
+  req.setEncoding('utf8');
+  // Note: This is checking that this small amount of data is passed through in
+  // a single chunk, which is unusual for our test suite but seems like a
+  // reasonable assumption here.
+  req.on('data', common.mustCall((data) => {
+    assert.strictEqual(data, testData);
+  }));
+  req.on('end', common.mustCall(() => {
+    clientSide.destroy();
+    clientSide.end();
+  }));
+  req.end();
+}