feat: allow connection header in request (nodejs#1829)

Author: metcoder95

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
@@ -1363,7 +1363,6 @@ function write (client, request) {
 
   if (method === 'HEAD') {
     // 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.
 
@@ -1377,6 +1376,10 @@ function write (client, request) {
     socket[kReset] = true
   }
 
+  if (reset) {
+    socket[kReset] = true
+  }
+
   if (client[kMaxRequests] && socket[kCounter]++ >= client[kMaxRequests]) {
     socket[kReset] = true
   }

lib/core/request.js

@@ -65,6 +65,7 @@ class Request {
     upgrade,
     headersTimeout,
     bodyTimeout,
+    reset,
     throwOnError
   }, handler) {
     if (typeof path !== 'string') {
@@ -97,6 +98,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 +144,8 @@ class Request {
 
     this.blocking = blocking == null ? false : blocking
 
+    this.reset = reset == null ? false : reset
+
     this.host = null
 
     this.contentLength = null
@@ -320,7 +327,12 @@ function processHeader (request, key, val) {
     key.length === 10 &&
     key.toLowerCase() === 'connection'
   ) {
-    throw new InvalidArgumentError('invalid connection header')
+    const value = val.toLowerCase()
+    if (value !== 'close' && value !== 'keep-alive') {
+      throw new InvalidArgumentError('invalid connection header')
+    } else if (value === 'close') {
+      request.reset = true
+    }
   } else if (
     key.length === 10 &&
     key.toLowerCase() === 'keep-alive'

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

@@ -50,7 +50,7 @@ test('invalid headers', (t) => {
     path: '/',
     method: 'GET',
     headers: {
-      connection: 'close'
+      connection: 'asd'
     }
   }, (err, data) => {
     t.type(err, errors.InvalidArgumentError)

test/request.js

@@ -149,3 +149,100 @@ 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(4)
+
+  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 new Promise((resolve, reject) => {
+      server.listen(0, (err) => {
+        if (err != null) reject(err)
+        else resolve()
+      })
+    })
+
+    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 new Promise((resolve, reject) => {
+      server.listen(0, (err) => {
+        if (err != null) reject(err)
+        else resolve()
+      })
+    })
+
+    await request({
+      method: 'GET',
+      origin: `http://localhost:${server.address().port}`,
+      reset: false
+    })
+  })
+
+  scope.test('Should react to manual set of "connection:close" header', 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 new Promise((resolve, reject) => {
+      server.listen(0, (err) => {
+        if (err != null) reject(err)
+        else resolve()
+      })
+    })
+
+    await request({
+      method: 'GET',
+      origin: `http://localhost:${server.address().port}`,
+      headers: {
+        connection: 'close'
+      }
+    })
+  })
+})

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;
   }