Dispatch compose (nodejs#2795)

* feat: new interceptors API

* WIP: compose

.gitignore

@@ -79,3 +79,5 @@ fuzz-results-*.json
 # Bundle output
 undici-fetch.js
 /test/imports/undici-import.js
+
+.tap

index.js

@@ -14,12 +14,7 @@ const MockClient = require('./lib/mock/mock-client')
 const MockAgent = require('./lib/mock/mock-agent')
 const MockPool = require('./lib/mock/mock-pool')
 const mockErrors = require('./lib/mock/mock-errors')
-const ProxyAgent = require('./lib/proxy-agent')
-const RetryAgent = require('./lib/retry-agent')
-const RetryHandler = require('./lib/handler/RetryHandler')
 const { getGlobalDispatcher, setGlobalDispatcher } = require('./lib/global')
-const DecoratorHandler = require('./lib/handler/DecoratorHandler')
-const RedirectHandler = require('./lib/handler/RedirectHandler')
 
 Object.assign(Dispatcher.prototype, api)
 
@@ -28,12 +23,12 @@ module.exports.Client = Client
 module.exports.Pool = Pool
 module.exports.BalancedPool = BalancedPool
 module.exports.Agent = Agent
-module.exports.ProxyAgent = ProxyAgent
-module.exports.RetryAgent = RetryAgent
-module.exports.RetryHandler = RetryHandler
 
-module.exports.DecoratorHandler = DecoratorHandler
-module.exports.RedirectHandler = RedirectHandler
+module.exports.interceptor = {
+  redirect: require('./lib/interceptor/redirect'),
+  retry: require('./lib/interceptor/retry'),
+  proxy: require('./lib/interceptor/proxy')
+}
 
 module.exports.buildConnector = buildConnector
 module.exports.errors = errors

lib/api/api-connect.js

@@ -3,7 +3,7 @@
 const { AsyncResource } = require('node:async_hooks')
 const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
 const util = require('../core/util')
-const RedirectHandler = require('../handler/RedirectHandler')
+const redirect = require('../interceptor/redirect')
 const { addSignal, removeSignal } = require('./abort-signal')
 
 class ConnectHandler extends AsyncResource {
@@ -91,19 +91,9 @@ function connect (opts, callback) {
   }
 
   try {
-    const connectHandler = new ConnectHandler(opts, callback)
-    const connectOptions = { ...opts, method: 'CONNECT' }
-
-    if (opts?.maxRedirections != null && (!Number.isInteger(opts?.maxRedirections) || opts?.maxRedirections < 0)) {
-      throw new InvalidArgumentError('maxRedirections must be a positive number')
-    }
-
-    if (opts?.maxRedirections > 0) {
-      RedirectHandler.buildDispatch(this, opts.maxRedirections)(connectOptions, connectHandler)
-      return
-    }
-
-    this.dispatch(connectOptions, connectHandler)
+    this
+      .compose(redirect(opts))
+      .dispatch({ ...opts, method: opts?.method || 'CONNECT' }, new ConnectHandler(opts, callback))
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-pipeline.js

@@ -13,7 +13,7 @@ const {
   RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
-const RedirectHandler = require('../handler/RedirectHandler')
+const redirect = require('../interceptor/redirect')
 const { addSignal, removeSignal } = require('./abort-signal')
 
 const kResume = Symbol('resume')
@@ -241,15 +241,9 @@ function pipeline (opts, handler) {
   try {
     const pipelineHandler = new PipelineHandler(opts, handler)
 
-    if (opts?.maxRedirections != null && (!Number.isInteger(opts?.maxRedirections) || opts?.maxRedirections < 0)) {
-      throw new InvalidArgumentError('maxRedirections must be a positive number')
-    }
-
-    if (opts?.maxRedirections > 0) {
-      RedirectHandler.buildDispatch(this, opts.maxRedirections)({ ...opts, body: pipelineHandler.req }, pipelineHandler)
-    } else {
-      this.dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)
-    }
+    this
+      .compose(redirect(opts))
+      .dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)
 
     return pipelineHandler.ret
   } catch (err) {

lib/api/api-request.js

@@ -7,7 +7,7 @@ const {
   RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
-const RedirectHandler = require('../handler/RedirectHandler')
+const redirect = require('../interceptor/redirect')
 const { getResolveErrorBodyCallback } = require('./util')
 const { addSignal, removeSignal } = require('./abort-signal')
 
@@ -167,18 +167,9 @@ function request (opts, callback) {
   }
 
   try {
-    const handler = new RequestHandler(opts, callback)
-
-    if (opts?.maxRedirections != null && (!Number.isInteger(opts?.maxRedirections) || opts?.maxRedirections < 0)) {
-      throw new InvalidArgumentError('maxRedirections must be a positive number')
-    }
-
-    if (opts?.maxRedirections > 0) {
-      RedirectHandler.buildDispatch(this, opts.maxRedirections)(opts, handler)
-      return
-    }
-
-    this.dispatch(opts, handler)
+    this
+      .compose(redirect(opts))
+      .dispatch(opts, new RequestHandler(opts, callback))
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-stream.js

@@ -7,7 +7,7 @@ const {
   RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
-const RedirectHandler = require('../handler/RedirectHandler')
+const redirect = require('../interceptor/redirect')
 const { getResolveErrorBodyCallback } = require('./util')
 const { AsyncResource } = require('node:async_hooks')
 const { addSignal, removeSignal } = require('./abort-signal')
@@ -208,18 +208,9 @@ function stream (opts, factory, callback) {
   }
 
   try {
-    const handler = new StreamHandler(opts, factory, callback)
-
-    if (opts?.maxRedirections != null && (!Number.isInteger(opts?.maxRedirections) || opts?.maxRedirections < 0)) {
-      throw new InvalidArgumentError('maxRedirections must be a positive number')
-    }
-
-    if (opts?.maxRedirections > 0) {
-      RedirectHandler.buildDispatch(this, opts.maxRedirections)(opts, handler)
-      return
-    }
-
-    this.dispatch(opts, handler)
+    this
+      .compose(redirect(opts))
+      .dispatch(opts, new StreamHandler(opts, factory, callback))
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-upgrade.js

@@ -4,7 +4,7 @@ const assert = require('node:assert')
 const { AsyncResource } = require('node:async_hooks')
 const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
 const util = require('../core/util')
-const RedirectHandler = require('../handler/RedirectHandler')
+const redirect = require('../interceptor/redirect')
 const { addSignal, removeSignal } = require('./abort-signal')
 
 class UpgradeHandler extends AsyncResource {
@@ -88,23 +88,9 @@ function upgrade (opts, callback) {
   }
 
   try {
-    const upgradeHandler = new UpgradeHandler(opts, callback)
-    const upgradeOpts = {
-      ...opts,
-      method: opts.method || 'GET',
-      upgrade: opts.protocol || 'Websocket'
-    }
-
-    if (opts?.maxRedirections != null && (!Number.isInteger(opts?.maxRedirections) || opts?.maxRedirections < 0)) {
-      throw new InvalidArgumentError('maxRedirections must be a positive number')
-    }
-
-    if (opts?.maxRedirections > 0) {
-      RedirectHandler.buildDispatch(this, opts.maxRedirections)(upgradeOpts, upgradeHandler)
-      return
-    }
-
-    this.dispatch(upgradeOpts, upgradeHandler)
+    this
+      .compose(redirect(opts))
+      .dispatch({ ...opts, method: opts?.method || 'GET', upgrade: opts?.protocol || 'Websocket' }, new UpgradeHandler(opts, callback))
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/dispatcher.js

@@ -2,7 +2,11 @@
 
 const EventEmitter = require('node:events')
 
+const kDispatcherVersion = Symbol.for('undici.dispatcher.version')
+
 class Dispatcher extends EventEmitter {
+  [kDispatcherVersion] = 1
+
   dispatch () {
     throw new Error('not implemented')
   }
@@ -14,6 +18,26 @@ class Dispatcher extends EventEmitter {
   destroy () {
     throw new Error('not implemented')
   }
+
+  compose (...interceptors) {
+    let dispatcher = this
+    for (const interceptor of interceptors) {
+      if (interceptor == null) {
+        continue
+      }
+
+      if (typeof interceptor !== 'function') {
+        throw new Error('invalid interceptor')
+      }
+
+      dispatcher = interceptor(dispatcher) ?? dispatcher
+
+      if (dispatcher[kDispatcherVersion] !== 1) {
+        throw new Error('invalid dispatcher')
+      }
+    }
+    return dispatcher
+  }
 }
 
 module.exports = Dispatcher

lib/proxy-agent.js → lib/interceptor/proxy.js

@@ -1,12 +1,12 @@
 'use strict'
 
-const { kProxy, kClose, kDestroy } = require('./core/symbols')
+const { kProxy, kClose, kDestroy } = require('../core/symbols')
 const { URL } = require('node:url')
-const Agent = require('./agent')
-const Pool = require('./pool')
-const DispatcherBase = require('./dispatcher-base')
-const { InvalidArgumentError, RequestAbortedError } = require('./core/errors')
-const buildConnector = require('./core/connect')
+const Agent = require('../agent')
+const Pool = require('../pool')
+const DispatcherBase = require('../dispatcher-base')
+const { InvalidArgumentError, RequestAbortedError } = require('../core/errors')
+const buildConnector = require('../core/connect')
 
 const kAgent = Symbol('proxy agent')
 const kClient = Symbol('proxy client')
@@ -39,10 +39,10 @@ function defaultFactory (origin, opts) {
 }
 
 class ProxyAgent extends DispatcherBase {
-  constructor (opts) {
+  constructor (dispatcher, opts) {
     super(opts)
     this[kProxy] = buildProxyOptions(opts)
-    this[kAgent] = new Agent(opts)
+    this[kAgent] = dispatcher
 
     if (typeof opts === 'string') {
       opts = { uri: opts }
@@ -183,4 +183,24 @@ function throwIfProxyAuthIsSent (headers) {
   }
 }
 
-module.exports = ProxyAgent
+module.exports = (opts) => {
+  if (typeof opts === 'string') {
+    opts = { uri: opts }
+  }
+
+  if (!opts || !opts.uri) {
+    throw new InvalidArgumentError('Proxy opts.uri is mandatory')
+  }
+
+  const { clientFactory = defaultFactory } = opts
+
+  if (typeof clientFactory !== 'function') {
+    throw new InvalidArgumentError('Proxy opts.clientFactory must be a function.')
+  }
+
+  if (opts.auth && opts.token) {
+    throw new InvalidArgumentError('opts.auth cannot be used in combination with opts.token')
+  }
+
+  return (dispatcher) => new ProxyAgent(dispatcher, opts)
+}