add backward compatibility for the undici v6

index.js

@@ -7,6 +7,7 @@ const { threadId, MessageChannel, parentPort } = require('worker_threads')
 const inject = require('light-my-request')
 const Hooks = require('./lib/hooks')
 const DispatchController = require('./lib/dispatch-controller')
+const WrapHandler = require('./lib/wrap-handler')
 
 const kAddress = Symbol('undici-thread-interceptor.address')
 
@@ -30,6 +31,8 @@ function createThreadInterceptor (opts) {
         url = new URL(opts.path, url)
       }
 
+      handler = handler.onRequestStart ? handler : new WrapHandler(handler)
+
       // Hostnames are case-insensitive
       const roundRobin = routes.get(url.hostname.toLowerCase())
       if (!roundRobin) {

lib/wrap-handler.js

@@ -0,0 +1,46 @@
+'use strict'
+
+class WrapHandler {
+  #handler
+
+  constructor (handler) {
+    this.#handler = handler
+  }
+
+  onRequestStart (controller, context) {
+    this.#handler.onConnect?.((err) => controller.abort(err), context)
+  }
+
+  onResponseStart (controller, statusCode, statusMessage, headers) {
+    const rawHeaders = []
+    for (const [key, val] of Object.entries(headers)) {
+      rawHeaders.push(Buffer.from(key), Buffer.from(val))
+    }
+
+    this.#handler.onHeaders?.(
+      statusCode,
+      rawHeaders,
+      () => {},
+      statusMessage
+    )
+  }
+
+  onResponseData (controller, data) {
+    this.#handler.onData?.(data)
+  }
+
+  onResponseEnd (controller, trailers) {
+    const rawTrailers = []
+    for (const [key, val] of Object.entries(trailers)) {
+      rawTrailers.push(Buffer.from(key), Buffer.from(val))
+    }
+
+    this.#handler.onComplete?.(rawTrailers)
+  }
+
+  onResponseError (controller, err) {
+    this.#handler.onError?.(err)
+  }
+}
+
+module.exports = WrapHandler

test/handler.test.js

@@ -1,11 +1,14 @@
 'use strict'
 
+// This is needed to set the nodejs Global Dispatcher
+fetch('http://google.com')
+
 const { test } = require('node:test')
 const { deepStrictEqual, strictEqual } = require('node:assert')
 const { join } = require('node:path')
 const { Worker } = require('node:worker_threads')
 const { createThreadInterceptor } = require('../')
-const { Agent, request } = require('undici')
+const { Agent, getGlobalDispatcher, request } = require('undici')
 
 class TestHandler {
   #handler
@@ -82,3 +85,23 @@ test('support undici v7 handler interface', async (t) => {
   strictEqual(statusCode, 200)
   deepStrictEqual(await body.json(), { hello: 'world' })
 })
+
+test('support undici v6 handler interface', async (t) => {
+  const worker = new Worker(join(__dirname, 'fixtures', 'worker1.js'))
+  t.after(() => worker.terminate())
+
+  const interceptor = createThreadInterceptor({
+    domain: '.local',
+  })
+  interceptor.route('myserver', worker)
+
+  const agent = getGlobalDispatcher()
+    .compose(interceptor)
+
+  const { statusCode, body } = await request('http://myserver.local', {
+    dispatcher: agent
+  })
+
+  strictEqual(statusCode, 200)
+  deepStrictEqual(await body.json(), { hello: 'world' })
+})