feat: allow connection header in request

docs/api/Dispatcher.md

@@ -192,6 +192,7 @@ Returns: `Boolean` - `false` if dispatcher is busy and further dispatch calls wo
 * **origin** `string | URL`
 * **path** `string`
 * **method** `string`
+* **reset** `boolean` (optional) - Default: `false` - Indicates whether the request should attempt to create a long-living connection by sending the `connection: keep-alive` header, or close it immediately after response by sending `connection: close`.
 * **body** `string | Buffer | Uint8Array | stream.Readable | Iterable | AsyncIterable | null` (optional) - Default: `null`
 * **headers** `UndiciHeaders | string[]` (optional) - Default: `null`.
 * **query** `Record<string, any> | null` (optional) - Default: `null` - Query string params to be embedded in the request URL. Note that both keys and values of query are encoded using `encodeURIComponent`. If for some reason you need to send them unencoded, embed query params into path directly instead.

lib/client.js

@@ -1295,7 +1295,7 @@ function _resume (client, sync) {
 }
 
 function write (client, request) {
-  const { body, method, path, host, upgrade, headers, blocking } = request
+  const { body, method, path, host, upgrade, headers, blocking, reset } = request
 
   // https://tools.ietf.org/html/rfc7231#section-4.3.1
   // https://tools.ietf.org/html/rfc7231#section-4.3.2
@@ -1361,16 +1361,11 @@ function write (client, request) {
     return false
   }
 
-  if (method === 'HEAD') {
+  if (upgrade || method === 'CONNECT' || method === 'HEAD' || reset) {
     // https://github.com/mcollina/undici/issues/258
-
     // Close after a HEAD request to interop with misbehaving servers
     // that may send a body in the response.
 
-    socket[kReset] = true
-  }
-
-  if (upgrade || method === 'CONNECT') {
     // On CONNECT or upgrade, block pipeline from dispatching further
     // requests on this connection.
 

lib/core/request.js

@@ -54,19 +54,22 @@ try {
 }
 
 class Request {
-  constructor (origin, {
-    path,
-    method,
-    body,
-    headers,
-    query,
-    idempotent,
-    blocking,
-    upgrade,
-    headersTimeout,
-    bodyTimeout,
-    throwOnError
-  }, handler) {
+  constructor (origin, options, handler) {
+    const {
+      path,
+      method,
+      body,
+      headers,
+      query,
+      idempotent,
+      blocking,
+      upgrade,
+      headersTimeout,
+      bodyTimeout,
+      reset,
+      throwOnError
+    } = options
+
     if (typeof path !== 'string') {
       throw new InvalidArgumentError('path must be a string')
     } else if (
@@ -97,6 +100,10 @@ class Request {
       throw new InvalidArgumentError('invalid bodyTimeout')
     }
 
+    if (reset != null && typeof reset !== 'boolean') {
+      throw new InvalidArgumentError('invalid reset')
+    }
+
     this.headersTimeout = headersTimeout
 
     this.bodyTimeout = bodyTimeout
@@ -139,6 +146,8 @@ class Request {
 
     this.blocking = blocking == null ? false : blocking
 
+    this.reset = reset == null ? false : reset
+
     this.host = null
 
     this.contentLength = null
@@ -250,6 +259,7 @@ class Request {
     if (channels.trailers.hasSubscribers) {
       channels.trailers.publish({ request: this, trailers })
     }
+
     return this[kHandler].onComplete(trailers)
   }
 
@@ -320,7 +330,10 @@ function processHeader (request, key, val) {
     key.length === 10 &&
     key.toLowerCase() === 'connection'
   ) {
-    throw new InvalidArgumentError('invalid connection header')
+    if (val.toLowerCase() === 'close') {
+      request.reset = true
+    }
+    request.headers += `${key}: ${val}\r\n`
   } else if (
     key.length === 10 &&
     key.toLowerCase() === 'keep-alive'

lib/pool.js

@@ -25,8 +25,8 @@ function defaultFactory (origin, opts) {
 
 class Pool extends PoolBase {
   constructor (origin, {
-    connections,
     factory = defaultFactory,
+    connections,
     connect,
     connectTimeout,
     tls,

test/client.js

@@ -88,6 +88,77 @@ test('basic get', (t) => {
   })
 })
 
+test('basic get with custom request.reset=true', (t) => {
+  t.plan(26)
+
+  const server = createServer((req, res) => {
+    t.equal('/', req.url)
+    t.equal('GET', req.method)
+    t.equal(`localhost:${server.address().port}`, req.headers.host)
+    t.equal(req.headers.connection, 'close')
+    t.equal(undefined, req.headers.foo)
+    t.equal('bar', req.headers.bar)
+    t.equal(undefined, req.headers['content-length'])
+    res.setHeader('Content-Type', 'text/plain')
+    res.end('hello')
+  })
+  t.teardown(server.close.bind(server))
+
+  const reqHeaders = {
+    foo: undefined,
+    bar: 'bar'
+  }
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`, {})
+    t.teardown(client.close.bind(client))
+
+    t.equal(client[kUrl].origin, `http://localhost:${server.address().port}`)
+
+    const signal = new EE()
+    client.request({
+      signal,
+      path: '/',
+      method: 'GET',
+      reset: true,
+      headers: reqHeaders
+    }, (err, data) => {
+      t.error(err)
+      const { statusCode, headers, body } = data
+      t.equal(statusCode, 200)
+      t.equal(signal.listenerCount('abort'), 1)
+      t.equal(headers['content-type'], 'text/plain')
+      const bufs = []
+      body.on('data', (buf) => {
+        bufs.push(buf)
+      })
+      body.on('end', () => {
+        t.equal(signal.listenerCount('abort'), 0)
+        t.equal('hello', Buffer.concat(bufs).toString('utf8'))
+      })
+    })
+    t.equal(signal.listenerCount('abort'), 1)
+
+    client.request({
+      path: '/',
+      reset: true,
+      method: 'GET',
+      headers: reqHeaders
+    }, (err, { statusCode, headers, body }) => {
+      t.error(err)
+      t.equal(statusCode, 200)
+      t.equal(headers['content-type'], 'text/plain')
+      const bufs = []
+      body.on('data', (buf) => {
+        bufs.push(buf)
+      })
+      body.on('end', () => {
+        t.equal('hello', Buffer.concat(bufs).toString('utf8'))
+      })
+    })
+  })
+})
+
 test('basic get with query params', (t) => {
   t.plan(4)
 

test/invalid-headers.js

@@ -4,7 +4,7 @@ const { test } = require('tap')
 const { Client, errors } = require('..')
 
 test('invalid headers', (t) => {
-  t.plan(10)
+  t.plan(9)
 
   const client = new Client('http://localhost:3000')
   t.teardown(client.destroy.bind(client))
@@ -46,16 +46,6 @@ test('invalid headers', (t) => {
     t.type(err, errors.InvalidArgumentError)
   })
 
-  client.request({
-    path: '/',
-    method: 'GET',
-    headers: {
-      connection: 'close'
-    }
-  }, (err, data) => {
-    t.type(err, errors.InvalidArgumentError)
-  })
-
   client.request({
     path: '/',
     method: 'GET',

test/request.js

@@ -149,3 +149,61 @@ test('Absolute URL as pathname should be included in req.path', async (t) => {
   t.equal(noSlashPath2Arg.statusCode, 200)
   t.end()
 })
+
+test('DispatchOptions#reset', scope => {
+  scope.plan(3)
+
+  scope.test('Should throw if invalid reset option', t => {
+    t.plan(1)
+
+    t.rejects(request({
+      method: 'GET',
+      origin: 'http://somehost.xyz',
+      reset: 0
+    }), 'invalid reset')
+  })
+
+  scope.test('Should include "connection:close" if reset true', async t => {
+    const server = createServer((req, res) => {
+      t.equal('GET', req.method)
+      t.equal(`localhost:${server.address().port}`, req.headers.host)
+      t.equal(req.headers.connection, 'close')
+      res.statusCode = 200
+      res.end('hello')
+    })
+
+    t.plan(3)
+
+    t.teardown(server.close.bind(server))
+
+    await server.listen(0)
+
+    await request({
+      method: 'GET',
+      origin: `http://localhost:${server.address().port}`,
+      reset: true
+    })
+  })
+
+  scope.test('Should include "connection:keep-alive" if reset false', async t => {
+    const server = createServer((req, res) => {
+      t.equal('GET', req.method)
+      t.equal(`localhost:${server.address().port}`, req.headers.host)
+      t.equal(req.headers.connection, 'keep-alive')
+      res.statusCode = 200
+      res.end('hello')
+    })
+
+    t.plan(3)
+
+    t.teardown(server.close.bind(server))
+
+    await server.listen(0)
+
+    await request({
+      method: 'GET',
+      origin: `http://localhost:${server.address().port}`,
+      reset: false
+    })
+  })
+})

test/types/api.test-d.ts

@@ -5,7 +5,7 @@ import { Dispatcher, request, stream, pipeline, connect, upgrade } from '../..'
 // request
 expectAssignable<Promise<Dispatcher.ResponseData>>(request(''))
 expectAssignable<Promise<Dispatcher.ResponseData>>(request('', { }))
-expectAssignable<Promise<Dispatcher.ResponseData>>(request('', { method: 'GET' }))
+expectAssignable<Promise<Dispatcher.ResponseData>>(request('', { method: 'GET', reset: false }))
 
 // stream
 expectAssignable<Promise<Dispatcher.StreamData>>(stream('', { method: 'GET' }, data => {

test/types/dispatcher.test-d.ts

@@ -14,7 +14,7 @@ expectAssignable<Dispatcher>(new Dispatcher())
   expectAssignable<boolean>(dispatcher.dispatch({ origin: '', path: '', method: 'GET' }, {}))
   expectAssignable<boolean>(dispatcher.dispatch({ origin: '', path: '', method: 'GET', headers: [] }, {}))
   expectAssignable<boolean>(dispatcher.dispatch({ origin: '', path: '', method: 'GET', headers: {} }, {}))
-  expectAssignable<boolean>(dispatcher.dispatch({ origin: '', path: '', method: 'GET', headers: null }, {}))
+  expectAssignable<boolean>(dispatcher.dispatch({ origin: '', path: '', method: 'GET', headers: null, reset: true }, {}))
   expectAssignable<boolean>(dispatcher.dispatch({ origin: new URL('http://localhost'), path: '', method: 'GET' }, {}))
 
   // connect
@@ -30,7 +30,7 @@ expectAssignable<Dispatcher>(new Dispatcher())
   expectAssignable<Promise<Dispatcher.ResponseData>>(dispatcher.request({ origin: '', path: '', method: 'GET', maxRedirections: 0, query: { pageNum: 1, id: 'abc' } }))
   expectAssignable<Promise<Dispatcher.ResponseData>>(dispatcher.request({ origin: '', path: '', method: 'GET', maxRedirections: 0, throwOnError: true }))
   expectAssignable<Promise<Dispatcher.ResponseData>>(dispatcher.request({ origin: new URL('http://localhost'), path: '', method: 'GET' }))
-  expectAssignable<void>(dispatcher.request({ origin: '', path: '', method: 'GET' }, (err, data) => {
+  expectAssignable<void>(dispatcher.request({ origin: '', path: '', method: 'GET', reset: true }, (err, data) => {
     expectAssignable<Error | null>(err)
     expectAssignable<Dispatcher.ResponseData>(data)
   }))
@@ -59,7 +59,7 @@ expectAssignable<Dispatcher>(new Dispatcher())
     return new Writable()
   }))
   expectAssignable<void>(dispatcher.stream(
-    { origin: '', path: '', method: 'GET' },
+    { origin: '', path: '', method: 'GET', reset: false },
     data => {
       expectAssignable<Dispatcher.StreamFactoryData>(data)
       return new Writable()

types/dispatcher.d.ts

@@ -111,6 +111,8 @@ declare namespace Dispatcher {
     headersTimeout?: number | null;
     /** The timeout after which a request will time out, in milliseconds. Monitors time between receiving body data. Use 0 to disable it entirely. Defaults to 30 seconds. */
     bodyTimeout?: number | null;
+    /** Whether the request should stablish a keep-alive or not. Default `false` */
+    reset?: boolean;
     /** Whether Undici should throw an error upon receiving a 4xx or 5xx response from the server. Defaults to false */
     throwOnError?: boolean;
   }