From c18872bb2ef2fe330c883de83ad4a45029bd61e9 Mon Sep 17 00:00:00 2001 From: Dolan Halbrook Date: Tue, 22 Apr 2025 17:24:42 -0700 Subject: [PATCH] Add "clientTtl" option to close and remove connections from the pool after a specified time. --- docs/docs/api/Pool.md | 1 + lib/dispatcher/pool.js | 20 ++++++++++++++--- test/pool.js | 50 ++++++++++++++++++++++++++++++++++++++++++ types/pool.d.ts | 2 ++ 4 files changed, 70 insertions(+), 3 deletions(-) diff --git a/docs/docs/api/Pool.md b/docs/docs/api/Pool.md index 9c4328240af..ee0a0d3fe81 100644 --- a/docs/docs/api/Pool.md +++ b/docs/docs/api/Pool.md @@ -19,6 +19,7 @@ Extends: [`ClientOptions`](/docs/docs/api/Client.md#parameter-clientoptions) * **factory** `(origin: URL, opts: Object) => Dispatcher` - Default: `(origin, opts) => new Client(origin, opts)` * **connections** `number | null` (optional) - Default: `null` - The number of `Client` instances to create. When set to `null`, the `Pool` instance will create an unlimited amount of `Client` instances. +* **clientTtl** `number | null` (optional) - Default: `null` - The amount of time before a `Client` instance is removed from the `Pool` and closed. When set to `null`, `Client` instances will not be removed or closed based on age. ## Instance Properties diff --git a/lib/dispatcher/pool.js b/lib/dispatcher/pool.js index 14175cf4a18..00cf50c3012 100644 --- a/lib/dispatcher/pool.js +++ b/lib/dispatcher/pool.js @@ -5,7 +5,8 @@ const { kClients, kNeedDrain, kAddClient, - kGetDispatcher + kGetDispatcher, + kRemoveClient } = require('./pool-base') const Client = require('./client') const { @@ -35,6 +36,7 @@ class Pool extends PoolBase { autoSelectFamily, autoSelectFamilyAttemptTimeout, allowH2, + clientTtl, ...options } = {}) { if (connections != null && (!Number.isFinite(connections) || connections < 0)) { @@ -65,12 +67,20 @@ class Pool extends PoolBase { this[kConnections] = connections || null this[kUrl] = util.parseOrigin(origin) - this[kOptions] = { ...util.deepClone(options), connect, allowH2 } + this[kOptions] = { ...util.deepClone(options), connect, allowH2, clientTtl } this[kOptions].interceptors = options.interceptors ? { ...options.interceptors } : undefined this[kFactory] = factory + this.on('connect', (origin, targets) => { + if (clientTtl != null && clientTtl > 0) { + for (const target of targets) { + Object.assign(target, { ttl: Date.now() }) + } + } + }) + this.on('connectionError', (origin, targets, error) => { // If a connection error occurs, we remove the client from the pool, // and emit a connectionError event. They will not be re-used. @@ -87,8 +97,12 @@ class Pool extends PoolBase { } [kGetDispatcher] () { + const clientTtlOption = this[kOptions].clientTtl for (const client of this[kClients]) { - if (!client[kNeedDrain]) { + // check ttl of client and if it's stale, remove it from the pool + if (clientTtlOption != null && clientTtlOption > 0 && client.ttl && ((Date.now() - client.ttl) > clientTtlOption)) { + this[kRemoveClient](client) + } else if (!client[kNeedDrain]) { return client } } diff --git a/test/pool.js b/test/pool.js index 2b847af70d3..908aa5a5e8a 100644 --- a/test/pool.js +++ b/test/pool.js @@ -578,6 +578,56 @@ test('pool connect', async (t) => { await t.completed }) +test('pool connect with clientTtl specified', async (t) => { + t = tspl(t, { plan: 1 }) + + const server = createServer({ joinDuplicateHeaders: true }, (c) => { + t.fail() + }) + server.on('connect', (req, socket, firstBodyChunk) => { + socket.write('HTTP/1.1 200 Connection established\r\n\r\n') + + let data = firstBodyChunk.toString() + socket.on('data', (buf) => { + data += buf.toString() + }) + + socket.on('end', () => { + socket.end(data) + }) + }) + after(() => server.close()) + + server.listen(0, async () => { + const client = new Pool(`http://localhost:${server.address().port}`, { + clientTtl: 10 + }) + + const { socket } = await client.connect({ + path: '/' + }) + + t.strictEqual(socket.closed, false) + + let recvData = '' + socket.on('data', (d) => { + recvData += d + }) + + socket.on('end', () => { + t.strictEqual(recvData.toString(), 'Body') + }) + + socket.write('Body') + socket.end() + + await new Promise(resolve => setTimeout(resolve, 10)) + t.strictEqual(socket.closed, true) + }) + + await t.completed +}) + test('pool dispatch', async (t) => { t = tspl(t, { plan: 2 }) diff --git a/types/pool.d.ts b/types/pool.d.ts index 4814606a702..5198476eb9c 100644 --- a/types/pool.d.ts +++ b/types/pool.d.ts @@ -33,6 +33,8 @@ declare namespace Pool { factory?(origin: URL, opts: object): Dispatcher; /** The max number of clients to create. `null` if no limit. Default `null`. */ connections?: number | null; + /** The amount of time before a client is removed from the pool and closed. `null` if no time limit. Default `null` */ + clientTtl?: number | null; interceptors?: { Pool?: readonly Dispatcher.DispatchInterceptor[] } & Client.Options['interceptors'] }