feat: new hooks

lib/api/readable.js

@@ -40,6 +40,13 @@ class BodyReadable extends Readable {
     contentLength,
     highWaterMark = 64 * 1024 // Same as nodejs fs streams.
   }) {
+    if (typeof resume !== 'function') {
+      throw new InvalidArgumentError('resume must be a function')
+    }
+    if (typeof abort !== 'function') {
+      throw new InvalidArgumentError('abort must be a function')
+    }
+
     super({
       autoDestroy: true,
       read: resume,

lib/core/request.js

@@ -15,12 +15,12 @@ const {
   isIterable,
   isBlobLike,
   serializePathWithQuery,
-  assertRequestHandler,
   getServerName,
   normalizedMethodRecords
 } = require('./util')
 const { channels } = require('./diagnostics.js')
 const { headerNameLowerCasedRecord } = require('./constants')
+const DecoratorHandler = require('../handler/decorator-handler.js')
 
 // Verifies that a given path is valid does not contain control chars \x00 to \x20
 const invalidPathRegex = /[^\u0021-\u00ff]/
@@ -183,11 +183,10 @@ class Request {
       throw new InvalidArgumentError('headers must be an object or an array')
     }
 
-    assertRequestHandler(handler, method, upgrade)
-
     this.servername = servername || getServerName(this.host) || null
 
-    this[kHandler] = handler
+    // TODO (perf): Avoid this extra allocation used for compat.
+    this[kHandler] = handler instanceof DecoratorHandler ? handler : new DecoratorHandler(handler)
 
     if (channels.create.hasSubscribers) {
       channels.create.publish({ request: this })

lib/handler/decorator-handler.js

@@ -1,8 +1,67 @@
 'use strict'
 
 const assert = require('node:assert')
+const util = require('../core/util')
+const kAssignResume = Symbol('assignResume')
+
+function toRawHeaders (headers) {
+  const rawHeaders = []
+  if (headers != null) {
+    for (const [key, value] of Object.entries(headers)) {
+      rawHeaders.push(Buffer.from(key), Buffer.from(value))
+    }
+  }
+  return rawHeaders
+}
+
+class CompatController {
+  #paused
+  #abort
+  #resume
+  #reason = null
+  #aborted = false
+
+  constructor (abort) {
+    this.#abort = abort
+  }
+
+  get aborted () {
+    return this.#aborted
+  }
+
+  get reason () {
+    return this.#reason
+  }
+
+  get paused () {
+    return this.#paused
+  }
+
+  abort (reason) {
+    this.#reason = reason
+    this.#aborted = true
+    this.#abort(reason)
+  }
+
+  resume () {
+    this.#paused = false
+    this.#resume?.()
+  }
+
+  pause () {
+    this.#paused = true
+  }
+
+  [kAssignResume] (resume) {
+    this.#resume = resume
+    if (this.#paused === false) {
+      this.#resume()
+    }
+  }
+}
 
 module.exports = class DecoratorHandler {
+  #controller
   #handler
   #onCompleteCalled = false
   #onErrorCalled = false
@@ -14,49 +73,140 @@ module.exports = class DecoratorHandler {
     this.#handler = handler
   }
 
-  onConnect (...args) {
-    return this.#handler.onConnect?.(...args)
+  // New API
+
+  onResponseStart (controller) {
+    assert(!this.#onCompleteCalled)
+    assert(!this.#onErrorCalled)
+
+    this.#controller = controller
+
+    this.#handler.onResponseStart?.(controller)
+
+    if (this.#handler.onConnect) {
+      this.#handler.onConnect((reason) => this.#controller.abort(reason))
+    }
   }
 
-  onError (...args) {
-    this.#onErrorCalled = true
-    return this.#handler.onError?.(...args)
+  onResponseHeaders (headers, statusCode) {
+    assert(this.#controller)
+    assert(!this.#onCompleteCalled)
+    assert(!this.#onErrorCalled)
+
+    this.#handler.onResponseHeaders?.(headers, statusCode)
+
+    if (this.#handler.onHeaders) {
+      this.#handler.onHeaders(statusCode, toRawHeaders(headers), () => this.#controller.resume())
+    }
   }
 
-  onUpgrade (...args) {
+  onResponseData (data) {
+    assert(this.#controller)
     assert(!this.#onCompleteCalled)
     assert(!this.#onErrorCalled)
 
-    return this.#handler.onUpgrade?.(...args)
+    this.#handler.onResponseData?.(data)
+
+    if (this.#handler.onData?.(data) === false) {
+      this.#controller.pause()
+    }
   }
 
-  onResponseStarted (...args) {
+  onResponseEnd (trailers) {
+    assert(this.#controller)
     assert(!this.#onCompleteCalled)
     assert(!this.#onErrorCalled)
 
-    return this.#handler.onResponseStarted?.(...args)
+    this.#onCompleteCalled = true
+
+    this.#handler.onResponseEnd?.(trailers)
+
+    if (this.#handler.onComplete) {
+      this.#handler.onComplete(toRawHeaders(trailers))
+    }
+  }
+
+  onResponseError (err) {
+    this.#onErrorCalled = true
+
+    this.#handler.onResponseError?.(err)
+    this.#handler.onError?.(err)
+  }
+
+  // Legacy API
+
+  onConnect (abort) {
+    this.#controller = new CompatController(abort)
+    this.#handler.onConnect?.(abort)
+
+    if (this.#handler.onResponseStart) {
+      this.#handler.onResponseStart(this.#controller)
+    }
   }
 
-  onHeaders (...args) {
+  onHeaders (statusCode, headers, resume, statusText) {
+    assert(this.#controller)
     assert(!this.#onCompleteCalled)
     assert(!this.#onErrorCalled)
 
-    return this.#handler.onHeaders?.(...args)
+    if (this.#controller) {
+      this.#controller[kAssignResume](resume)
+    }
+
+    if (this.#handler.onResponseHeaders) {
+      this.#handler.onResponseHeaders(util.parseHeaders(headers), statusCode)
+    }
+
+    if (this.#handler.onHeaders?.(statusCode, headers, resume, statusText) === false) {
+      this.#controller.pause()
+    }
+
+    return !this.#controller.paused
   }
 
-  onData (...args) {
+  onData (data) {
+    assert(this.#controller)
     assert(!this.#onCompleteCalled)
     assert(!this.#onErrorCalled)
 
-    return this.#handler.onData?.(...args)
+    if (this.#handler.onResponseData) {
+      this.#handler.onResponseData(data)
+    }
+
+    if (this.#handler.onData?.(data) === false) {
+      this.#controller.pause()
+    }
+
+    return !this.#controller.paused
   }
 
-  onComplete (...args) {
+  onComplete (trailers) {
+    assert(this.#controller)
     assert(!this.#onCompleteCalled)
     assert(!this.#onErrorCalled)
 
     this.#onCompleteCalled = true
-    return this.#handler.onComplete?.(...args)
+
+    if (this.#handler.onResponseEnd) {
+      this.#handler.onResponseEnd(util.parseHeaders(trailers))
+    }
+
+    this.#handler.onComplete?.(trailers)
+  }
+
+  onError (...args) {
+    this.#onErrorCalled = true
+    this.#handler.onError?.(...args)
+    this.#handler.onResponseError?.(...args)
+  }
+
+  // Old API
+
+  onResponseStarted (...args) {
+    assert(!this.#onCompleteCalled)
+    assert(!this.#onErrorCalled)
+
+    return this.#handler.onResponseStarted?.(...args)
   }
 
   onBodySent (...args) {
@@ -65,4 +215,11 @@ module.exports = class DecoratorHandler {
 
     return this.#handler.onBodySent?.(...args)
   }
+
+  onUpgrade (...args) {
+    assert(!this.#onCompleteCalled)
+    assert(!this.#onErrorCalled)
+
+    return this.#handler.onUpgrade?.(...args)
+  }
 }

test/decorator-handler.js

@@ -4,17 +4,6 @@ const { tspl } = require('@matteo.collina/tspl')
 const { describe, test } = require('node:test')
 const DecoratorHandler = require('../lib/handler/decorator-handler')
 
-const methods = [
-  'onConnect',
-  'onError',
-  'onUpgrade',
-  'onHeaders',
-  'onResponseStarted',
-  'onData',
-  'onComplete',
-  'onBodySent'
-]
-
 describe('DecoratorHandler', () => {
   test('should throw if provided handler is not an object', (t) => {
     t = tspl(t, { plan: 4 })
@@ -31,44 +20,4 @@ describe('DecoratorHandler', () => {
     const decorator = new DecoratorHandler(handler)
     t.strictEqual(Object.keys(decorator).length, 0)
   })
-
-  methods.forEach((method) => {
-    test(`should have delegate ${method}-method`, (t) => {
-      t = tspl(t, { plan: 1 })
-      const decorator = new DecoratorHandler({})
-      t.equal(typeof decorator[method], 'function')
-    })
-
-    test(`should delegate ${method}-method`, (t) => {
-      t = tspl(t, { plan: 1 })
-      const handler = { [method]: () => method }
-      const decorator = new DecoratorHandler(handler)
-      t.equal(decorator[method](), method)
-    })
-
-    test(`should delegate ${method}-method with arguments`, (t) => {
-      t = tspl(t, { plan: 1 })
-      const handler = { [method]: (...args) => args }
-      const decorator = new DecoratorHandler(handler)
-      t.deepStrictEqual(decorator[method](1, 2, 3), [1, 2, 3])
-    })
-
-    test(`can be extended and should delegate ${method}-method`, (t) => {
-      t = tspl(t, { plan: 1 })
-
-      class ExtendedHandler extends DecoratorHandler {
-        [method] () {
-          return method
-        }
-      }
-      const decorator = new ExtendedHandler({})
-      t.equal(decorator[method](), method)
-    })
-
-    test(`calling the method ${method}-method should not throw if the method is not defined in the handler`, (t) => {
-      t = tspl(t, { plan: 1 })
-      const decorator = new DecoratorHandler({})
-      t.doesNotThrow(() => decorator[method]())
-    })
-  })
 })