Feat/ balancedPool add (#421)

* feat: support array of base URLs for undici balancedPool in request options

* feat: enhance base URL option to support array for load balancing with undici.BalancedPool

* fix: correct typos in README.md

Author: gulbaki

README.md

@@ -69,6 +69,15 @@ target.listen({ port: 3001 }, (err) => {
 Set the base URL for all the forwarded requests. Will be required if `http2` is set to `true`
 Note that _every path will be discarded_.
 
+Set the base URL for all the forwarded requests.  
+*String or String[]*:  
+
+* **Single string** → a normal `undici.Pool` / `http.request` client is used.  
+* **Array with ≥ 2 elements** → **[`undici.BalancedPool`](https://undici.nodejs.org/#/docs/api/BalancedPool)** is automatically selected and requests are load-balanced round-robin across the given origins.
+
+When you provide an array, only the *origin* (`protocol://host:port`) part of each URL is considered; any path component is ignored.
+
+
 Custom URL protocols `unix+http:` and `unix+https:` can be used to forward requests to a unix
 socket server by using `querystring.escape(socketPath)` as the hostname.  This is not supported
 for http2 nor undici.  To illustrate:
@@ -125,6 +134,17 @@ proxy.register(require('@fastify/reply-from'), {
 })
 ```
 
+You can also use with BalancedPool:
+ ```js
+proxy.register(require('@fastify/reply-from'), {
+  base: [
+    'http://api-1.internal:8080',
+    'http://api-2.internal:8080',
+    'http://api-3.internal:8080'
+  ]
+})
+```
+
 #### `http`
 
 Set the `http` option to an Object to use

lib/request.js

@@ -37,6 +37,9 @@ function isUndiciInstance (obj) {
 
 function buildRequest (opts) {
   const isHttp2 = !!opts.http2
+  if (Array.isArray(opts.base) && opts.base.length === 1) {
+    opts.base = opts.base[0]
+  }
   const hasUndiciOptions = shouldUseUndici(opts)
   const requests = {
     'http:': http,
@@ -46,7 +49,8 @@ function buildRequest (opts) {
   }
   const http2Opts = getHttp2Opts(opts)
   const httpOpts = getHttpOpts(opts)
-  const baseUrl = opts.base && new URL(opts.base).origin
+  const baseUrl = Array.isArray(opts.base) ? null : (opts.base && new URL(opts.base).origin)
+  const isBalanced = Array.isArray(opts.base) && opts.base.length > 1
   const undiciOpts = opts.undici || {}
   const globalAgent = opts.globalAgent
   const destroyAgent = opts.destroyAgent
@@ -75,7 +79,10 @@ function buildRequest (opts) {
   if (isHttp2) {
     return { request: handleHttp2Req, close, retryOnError: 'ECONNRESET' }
   } else if (hasUndiciOptions) {
-    if (opts.base?.startsWith('unix+')) {
+    if (isBalanced) {
+      const origins = opts.base.map(u => new URL(u).origin)
+      undiciInstance = new undici.BalancedPool(origins, getUndiciOptions(opts.undici))
+    } else if (opts.base?.startsWith('unix+')) {
       const undiciOpts = getUndiciOptions(opts.undici)
       undiciOpts.socketPath = decodeURIComponent(new URL(opts.base).host)
       const protocol = opts.base.startsWith('unix+https') ? 'https' : 'http'
@@ -155,6 +162,8 @@ function buildRequest (opts) {
 
     if (undiciInstance) {
       pool = undiciInstance
+    } else if (pool instanceof undici.BalancedPool) {
+      delete req.origin
     } else if (!baseUrl && opts.url.protocol.startsWith('unix')) {
       done(new Error('unix socket not supported with undici yet'))
       return

lib/utils.js

@@ -58,6 +58,7 @@ function stripHttp1ConnectionHeaders (headers) {
 
 // issue ref: https://github.com/fastify/fast-proxy/issues/42
 function buildURL (source, reqBase) {
+  if (Array.isArray(reqBase)) reqBase = reqBase[0]
   let baseOrigin = reqBase ? new URL(reqBase).href : undefined
 
   // To make sure we don't accidentally override the base path

test/balanced-pool.test.js

@@ -0,0 +1,48 @@
+'use strict'
+const t = require('node:test')
+const http = require('node:http')
+const Fastify = require('fastify')
+const From = require('..')
+const { request } = require('undici')
+
+t.test('undici balanced pool http', async t => {
+  const hit = [0, 0]
+  const makeTarget = idx => http.createServer((req, res) => {
+    hit[idx]++
+    res.statusCode = 200
+    res.end('hello world')
+  })
+  const target1 = makeTarget(0)
+  const target2 = makeTarget(1)
+
+  await Promise.all([
+    new Promise(resolve => target1.listen(0, resolve)),
+    new Promise(resolve => target2.listen(0, resolve))
+  ])
+  const p1 = target1.address().port
+  const p2 = target2.address().port
+
+  const proxy = Fastify()
+  proxy.register(From, {
+    base: [`http://localhost:${p1}`, `http://localhost:${p2}`]
+  })
+  proxy.get('*', (_req, reply) => {
+    reply.from()
+  })
+
+  t.after(() => {
+    proxy.close()
+    target1.close()
+    target2.close()
+  })
+
+  await proxy.listen({ port: 0 })
+  const proxyPort = proxy.server.address().port
+
+  for (let i = 0; i < 10; i++) {
+    const res = await request(`http://localhost:${proxyPort}/hello`)
+    t.assert.strictEqual(res.statusCode, 200)
+    t.assert.strictEqual(await res.body.text(), 'hello world')
+  }
+  t.assert.ok(hit[0] > 0 && hit[1] > 0, `load distribution OK => [${hit[0]}, ${hit[1]}]`)
+})

types/index.d.ts

@@ -100,12 +100,13 @@ declare namespace fastifyReplyFrom {
   }
 
   export interface FastifyReplyFromOptions {
-    base?: string;
+    base?: string | string[];
     cacheURLs?: number;
     disableCache?: boolean;
     http?: HttpOptions;
     http2?: Http2Options | boolean;
     undici?: Pool.Options & { proxy?: string | URL | ProxyAgent.Options } | { request: Dispatcher['request'] };
+    balancedPoolOptions?: Pool.Options & Record<string, unknown>;
     contentTypesToEncode?: string[];
     retryMethods?: (HTTPMethods | 'TRACE')[];
     maxRetriesOn503?: number;