fix: support new handler API in custom dispatchers

.gitignore

@@ -92,3 +92,7 @@ test/request-timeout.10mb.bin
 # Claude files
 CLAUDE.md
 .claude
+
+# Local tooling
+.githuman/
+.pi/

lib/api/api-connect.js

@@ -4,8 +4,18 @@ const assert = require('node:assert')
 const { AsyncResource } = require('node:async_hooks')
 const { InvalidArgumentError, SocketError } = require('../core/errors')
 const util = require('../core/util')
+const WrapHandler = require('../handler/wrap-handler')
+const DispatcherBase = require('../dispatcher/dispatcher-base')
 const { addSignal, removeSignal } = require('./abort-signal')
 
+function getDispatchHandler (dispatcher, handler) {
+  if (dispatcher instanceof DispatcherBase && dispatcher.dispatch === DispatcherBase.prototype.dispatch) {
+    return handler
+  }
+
+  return WrapHandler.wrap(handler)
+}
+
 class ConnectHandler extends AsyncResource {
   constructor (opts, callback) {
     if (!opts || typeof opts !== 'object') {
@@ -96,8 +106,9 @@ function connect (opts, callback) {
   try {
     const connectHandler = new ConnectHandler(opts, callback)
     const connectOptions = { ...opts, method: 'CONNECT' }
+    const dispatchHandler = getDispatchHandler(this, connectHandler)
 
-    this.dispatch(connectOptions, connectHandler)
+    this.dispatch(connectOptions, dispatchHandler)
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-pipeline.js

@@ -13,8 +13,18 @@ const {
   RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
+const WrapHandler = require('../handler/wrap-handler')
+const DispatcherBase = require('../dispatcher/dispatcher-base')
 const { addSignal, removeSignal } = require('./abort-signal')
 
+function getDispatchHandler (dispatcher, handler) {
+  if (dispatcher instanceof DispatcherBase && dispatcher.dispatch === DispatcherBase.prototype.dispatch) {
+    return handler
+  }
+
+  return WrapHandler.wrap(handler)
+}
+
 function noop () {}
 
 const kResume = Symbol('resume')
@@ -242,7 +252,8 @@ class PipelineHandler extends AsyncResource {
 function pipeline (opts, handler) {
   try {
     const pipelineHandler = new PipelineHandler(opts, handler)
-    this.dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)
+    const dispatchHandler = getDispatchHandler(this, pipelineHandler)
+    this.dispatch({ ...opts, body: pipelineHandler.req }, dispatchHandler)
     return pipelineHandler.ret
   } catch (err) {
     return new PassThrough().destroy(err)

lib/api/api-request.js

@@ -5,6 +5,16 @@ const { AsyncResource } = require('node:async_hooks')
 const { Readable } = require('./readable')
 const { InvalidArgumentError, RequestAbortedError } = require('../core/errors')
 const util = require('../core/util')
+const WrapHandler = require('../handler/wrap-handler')
+const DispatcherBase = require('../dispatcher/dispatcher-base')
+
+function getDispatchHandler (dispatcher, handler) {
+  if (dispatcher instanceof DispatcherBase && dispatcher.dispatch === DispatcherBase.prototype.dispatch) {
+    return handler
+  }
+
+  return WrapHandler.wrap(handler)
+}
 
 function noop () {}
 
@@ -198,8 +208,9 @@ function request (opts, callback) {
 
   try {
     const handler = new RequestHandler(opts, callback)
+    const dispatchHandler = getDispatchHandler(this, handler)
 
-    this.dispatch(opts, handler)
+    this.dispatch(opts, dispatchHandler)
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-stream.js

@@ -5,8 +5,18 @@ const { finished } = require('node:stream')
 const { AsyncResource } = require('node:async_hooks')
 const { InvalidArgumentError, InvalidReturnValueError } = require('../core/errors')
 const util = require('../core/util')
+const WrapHandler = require('../handler/wrap-handler')
+const DispatcherBase = require('../dispatcher/dispatcher-base')
 const { addSignal, removeSignal } = require('./abort-signal')
 
+function getDispatchHandler (dispatcher, handler) {
+  if (dispatcher instanceof DispatcherBase && dispatcher.dispatch === DispatcherBase.prototype.dispatch) {
+    return handler
+  }
+
+  return WrapHandler.wrap(handler)
+}
+
 function noop () {}
 
 class StreamHandler extends AsyncResource {
@@ -195,8 +205,9 @@ function stream (opts, factory, callback) {
 
   try {
     const handler = new StreamHandler(opts, factory, callback)
+    const dispatchHandler = getDispatchHandler(this, handler)
 
-    this.dispatch(opts, handler)
+    this.dispatch(opts, dispatchHandler)
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/api/api-upgrade.js

@@ -4,9 +4,19 @@ const { InvalidArgumentError, SocketError } = require('../core/errors')
 const { AsyncResource } = require('node:async_hooks')
 const assert = require('node:assert')
 const util = require('../core/util')
+const WrapHandler = require('../handler/wrap-handler')
+const DispatcherBase = require('../dispatcher/dispatcher-base')
 const { kHTTP2Stream } = require('../core/symbols')
 const { addSignal, removeSignal } = require('./abort-signal')
 
+function getDispatchHandler (dispatcher, handler) {
+  if (dispatcher instanceof DispatcherBase && dispatcher.dispatch === DispatcherBase.prototype.dispatch) {
+    return handler
+  }
+
+  return WrapHandler.wrap(handler)
+}
+
 class UpgradeHandler extends AsyncResource {
   constructor (opts, callback) {
     if (!opts || typeof opts !== 'object') {
@@ -97,8 +107,9 @@ function upgrade (opts, callback) {
       method: opts.method || 'GET',
       upgrade: opts.protocol || 'Websocket'
     }
+    const dispatchHandler = getDispatchHandler(this, upgradeHandler)
 
-    this.dispatch(upgradeOpts, upgradeHandler)
+    this.dispatch(upgradeOpts, dispatchHandler)
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

package.json

@@ -97,7 +97,7 @@
     "test:websocket:autobahn": "node test/autobahn/client.js",
     "test:websocket:autobahn:report": "node test/autobahn/report.js",
     "test:wpt:setup": "node test/web-platform-tests/wpt-runner.mjs setup",
-    "test:wpt": "npm run test:wpt:setup && node test/web-platform-tests/wpt-runner.mjs run /fetch /mimesniff /xhr /websockets /serviceWorkers /eventsource",
+    "test:wpt": "npm run test:wpt:setup && node test/web-platform-tests/wpt-runner.mjs run /fetch /mimesniff /websockets /serviceWorkers /eventsource",
     "test:cache-tests": "node test/cache-interceptor/cache-tests.mjs --ci",
     "coverage": "npm run coverage:clean && cross-env NODE_V8_COVERAGE=./coverage/tmp npm run test:javascript && npm run coverage:report",
     "coverage:ci": "npm run coverage:clean && cross-env NODE_V8_COVERAGE=./coverage/tmp npm run test:javascript && npm run coverage:report:ci",

test/issue-4780.js

@@ -0,0 +1,62 @@
+'use strict'
+
+const { strictEqual } = require('node:assert')
+const { test } = require('node:test')
+const { Dispatcher } = require('..')
+const DispatcherBase = require('../lib/dispatcher/dispatcher-base')
+
+function createController () {
+  return {
+    abort () {},
+    pause () {},
+    resume () {}
+  }
+}
+
+class NewAPIDispatcher extends Dispatcher {
+  dispatch (opts, handler) {
+    const controller = createController()
+
+    handler.onRequestStart?.(controller, {})
+    handler.onResponseStart?.(controller, 200, { 'content-type': 'text/plain' }, 'OK')
+    handler.onResponseData?.(controller, Buffer.from('Hello, world!'))
+    handler.onResponseEnd?.(controller, {})
+
+    return true
+  }
+
+  close () {}
+
+  destroy () {}
+}
+
+class NewAPIDispatcherBase extends DispatcherBase {
+  dispatch (opts, handler) {
+    const controller = createController()
+
+    handler.onRequestStart?.(controller, {})
+    handler.onResponseStart?.(controller, 200, { 'content-type': 'text/plain' }, 'OK')
+    handler.onResponseData?.(controller, Buffer.from('Hello, world!'))
+    handler.onResponseEnd?.(controller, {})
+
+    return true
+  }
+}
+
+async function assertRequestSucceeds (dispatcher) {
+  const response = await dispatcher.request({
+    origin: 'http://example.com',
+    path: '/',
+    method: 'GET'
+  })
+
+  strictEqual(await response.body.text(), 'Hello, world!')
+}
+
+test('https://github.com/nodejs/undici/issues/4780 - request uses new handler API (Dispatcher)', async () => {
+  await assertRequestSucceeds(new NewAPIDispatcher())
+})
+
+test('https://github.com/nodejs/undici/issues/4780 - request uses new handler API (DispatcherBase)', async () => {
+  await assertRequestSucceeds(new NewAPIDispatcherBase())
+})