158: Added a hook to get the url base based on URL object (#157)

* Added a hook to get the url base based on request params. Useful for gradual migrate services based on the request data

* updated readme with the getUpstream documentation

* added getUpstream test for undici and http

* increase test coverage

* increase test coverage

* added disableCache to readme and ts interface

Co-authored-by: Yohay <yohayg@gmail.com>

README.md

@@ -171,6 +171,12 @@ proxy.register(require('fastify-reply-from'), {
 
 The number of parsed URLs that will be cached. Default: `100`.
 
+#### `disableCache`
+
+This option will disable the URL caching. 
+This cache is dedicated to reduce the amount of URL object generation.
+Generating URLs is a main bottleneck of this module, please disable this cache with caution.
+
 ---
 
 ### `reply.from(source, [opts])`
@@ -219,6 +225,12 @@ It must return the new headers object.
 Called to rewrite the headers of the request, before them being sent to the other server.
 It must return the new headers object.
 
+#### `getUpstream(originalReq, base)`
+
+Called to get upstream destination, before the request is being sent. Useful when you want to decide which target server to call based on the request data.
+Helpful for a gradual rollout of new services.
+It must return the upstream destination.
+
 #### `queryString`
 
 Replaces the original querystring of the request with what is specified.

index.d.ts

@@ -49,6 +49,10 @@ export interface FastifyReplyFromHooks {
     req: Http2ServerRequest | IncomingMessage,
     headers: Http2IncomingHttpHeaders | IncomingHttpHeaders
   ) => Http2IncomingHttpHeaders | IncomingHttpHeaders;
+  getUpstream?: (
+      req: Http2ServerRequest | IncomingMessage,
+      base: string
+  ) => string;
 }
 
 declare module "fastify" {
@@ -76,6 +80,7 @@ interface HttpOptions {
 export interface FastifyReplyFromOptions {
   base?: string;
   cacheURLs?: number;
+  disableCache?: boolean;
   http?: HttpOptions;
   http2?: Http2Options | boolean;
   undici?: unknown; // undici has no TS declarations yet

index.js

@@ -16,7 +16,7 @@ const {
 const { TimeoutError } = buildRequest
 
 module.exports = fp(function from (fastify, opts, next) {
-  const cache = lru(opts.cacheURLs || 100)
+  const cache = opts.disableCache ? undefined : lru(opts.cacheURLs || 100)
   const base = opts.base
   const { request, close } = buildRequest({
     http: opts.http,
@@ -30,15 +30,22 @@ module.exports = fp(function from (fastify, opts, next) {
     const onResponse = opts.onResponse
     const rewriteHeaders = opts.rewriteHeaders || headersNoOp
     const rewriteRequestHeaders = opts.rewriteRequestHeaders || requestHeadersNoOp
+    const getUpstream = opts.getUpstream || upstreamNoOp
     const onError = opts.onError || onErrorDefault
 
     if (!source) {
       source = req.url
     }
 
     // we leverage caching to avoid parsing the destination URL
-    const url = cache.get(source) || buildURL(source, base)
-    cache.set(source, url)
+    const dest = getUpstream(req, base)
+    let url
+    if (cache) {
+      url = cache.get(source) || buildURL(source, dest)
+      cache.set(source, url)
+    } else {
+      url = buildURL(source, dest)
+    }
 
     const sourceHttp2 = req.httpVersionMajor === 2
     const headers = sourceHttp2 ? filterPseudoHeaders(req.headers) : req.headers
@@ -171,6 +178,10 @@ function requestHeadersNoOp (originalReq, headers) {
   return headers
 }
 
+function upstreamNoOp (req, base) {
+  return base
+}
+
 function onErrorDefault (reply, { error }) {
   reply.send(error)
 }

package.json

@@ -5,7 +5,7 @@
   "main": "index.js",
   "types": "index.d.ts",
   "scripts": {
-    "coverage": "tap -j8 test/*.js --cov",
+    "coverage": "tap -j8 test/*.js --cov --coverage-report=html",
     "lint:fix": "standard --fix",
     "test": "standard | snazzy && tap --ts test/* && npm run typescript",
     "test:ci": "standard | snazzy && tap --ts test/* --coverage-report=lcovonly && npm run typescript",

test/get-upstream-http.js

@@ -0,0 +1,70 @@
+'use strict'
+
+const t = require('tap')
+const Fastify = require('fastify')
+const From = require('..')
+const http = require('http')
+const get = require('simple-get').concat
+
+const instance = Fastify()
+const instanceWithoutBase = Fastify()
+instance.register(From, {
+  base: 'http://localhost',
+  http: true,
+  disableCache: true
+})
+
+instanceWithoutBase.register(From, {
+  http: true,
+  disableCache: true
+})
+
+t.plan(13)
+t.tearDown(instance.close.bind(instance))
+t.tearDown(instanceWithoutBase.close.bind(instanceWithoutBase))
+
+const target = http.createServer((req, res) => {
+  t.pass('request proxied')
+  t.equal(req.method, 'GET')
+  res.end(req.headers.host)
+})
+
+instance.get('/test', (request, reply) => {
+  reply.from('/test', {
+    getUpstream: (req, base) => {
+      t.pass('getUpstream called')
+      return `${base}:${target.address().port}`
+    }
+  })
+})
+
+instanceWithoutBase.get('/test2', (request, reply) => {
+  reply.from('/test2', {
+    getUpstream: () => {
+      t.pass('getUpstream called')
+      return `http://localhost:${target.address().port}`
+    }
+  })
+})
+
+t.tearDown(target.close.bind(target))
+
+instance.listen(0, (err) => {
+  t.error(err)
+  instanceWithoutBase.listen(0, (err) => {
+    t.error(err)
+    target.listen(0, (err) => {
+      t.error(err)
+
+      get(`http://localhost:${instance.server.address().port}/test`, (err, res) => {
+        t.error(err)
+        t.equal(res.statusCode, 200)
+      })
+
+      get(`http://localhost:${instanceWithoutBase.server.address().port}/test2`, (err, res) => {
+        t.error(err)
+        t.equal(res.statusCode, 200)
+      })
+    })
+  })
+})

test/get-upstream-undici.js

@@ -0,0 +1,45 @@
+'use strict'
+
+const t = require('tap')
+const Fastify = require('fastify')
+const From = require('..')
+const http = require('http')
+const get = require('simple-get').concat
+
+const instance = Fastify()
+instance.register(From, {
+  disableCache: true
+})
+
+t.plan(7)
+t.tearDown(instance.close.bind(instance))
+
+const target = http.createServer((req, res) => {
+  t.pass('request proxied')
+  t.equal(req.method, 'GET')
+  res.end(req.headers.host)
+})
+
+instance.get('/test', (request, reply) => {
+  reply.from('/test', {
+    getUpstream: () => {
+      t.pass('getUpstream called')
+      return `http://localhost:${target.address().port}`
+    }
+  })
+})
+
+t.tearDown(target.close.bind(target))
+
+instance.listen(0, (err) => {
+  t.error(err)
+
+  target.listen(0, (err) => {
+    t.error(err)
+
+    get(`http://localhost:${instance.server.address().port}/test`, (err, res) => {
+      t.error(err)
+      t.equal(res.statusCode, 200)
+    })
+  })
+})

test/index.test-d.ts

@@ -36,6 +36,7 @@ const fullOptions: FastifyReplyFromOptions = {
     }
   },
   cacheURLs: 100,
+  disableCache: false,
   keepAliveMsecs: 60 * 1000,
   maxFreeSockets: 2048,
   maxSockets: 2048,
@@ -67,6 +68,10 @@ async function main() {
           rewriteRequestHeaders(req, headers) {
               expectType<http.IncomingMessage | http2.Http2ServerRequest>(req);
               return headers;
+          },
+          getUpstream(req, base) {
+              expectType<http.IncomingMessage | http2.Http2ServerRequest>(req);
+              return base;
           }
       });
   });
@@ -90,6 +95,9 @@ async function main() {
           rewriteRequestHeaders(req, headers: IncomingHttpHeaders) {
               return headers;
           },
+          getUpstream(req, base) {
+              return base;
+          },
           onError(reply: FastifyReply<RawServerBase>, error) {
               return reply.send(error.error);
           }