feat: onConnect w/ controller

lib/api/abort-signal.js

@@ -7,12 +7,17 @@ const kSignal = Symbol('kSignal')
 function abort (self) {
   if (self.abort) {
     self.abort(self[kSignal]?.reason)
+  } else if (self.controller?.abort) {
+    self.controller.abort(self[kSignal]?.reason)
   } else {
-    self.onError(self[kSignal]?.reason ?? new RequestAbortedError())
+    self.reason = self[kSignal]?.reason ?? new RequestAbortedError()
   }
+  removeSignal(self)
 }
 
 function addSignal (self, signal) {
+  self.reason = null
+
   self[kSignal] = null
   self[kListener] = null
 

lib/api/api-connect.js

@@ -1,7 +1,8 @@
 'use strict'
 
+const assert = require('node:assert')
 const { AsyncResource } = require('node:async_hooks')
-const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const { InvalidArgumentError, SocketError } = require('../core/errors')
 const util = require('../core/util')
 const { addSignal, removeSignal } = require('./abort-signal')
 
@@ -32,10 +33,13 @@ class ConnectHandler extends AsyncResource {
   }
 
   onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      abort(this.reason)
+      return
     }
 
+    assert(this.callback)
+
     this.abort = abort
     this.context = context
   }

lib/api/api-pipeline.js

@@ -147,12 +147,14 @@ class PipelineHandler extends AsyncResource {
   onConnect (abort, context) {
     const { ret, res } = this
 
-    assert(!res, 'pipeline cannot be retried')
-
-    if (ret.destroyed) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      abort(this.reason)
+      return
     }
 
+    assert(!res, 'pipeline cannot be retried')
+    assert(!ret.destroyed)
+
     this.abort = abort
     this.context = context
   }

lib/api/api-request.js

@@ -1,10 +1,8 @@
 'use strict'
 
+const assert = require('node:assert')
 const { Readable } = require('./readable')
-const {
-  InvalidArgumentError,
-  RequestAbortedError
-} = require('../core/errors')
+const { InvalidArgumentError } = require('../core/errors')
 const util = require('../core/util')
 const { getResolveErrorBodyCallback } = require('./util')
 const { AsyncResource } = require('node:async_hooks')
@@ -69,16 +67,19 @@ class RequestHandler extends AsyncResource {
   }
 
   onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      abort(this.reason)
+      return
     }
 
+    assert(this.callback)
+
     this.abort = abort
     this.context = context
   }
 
   onHeaders (statusCode, rawHeaders, resume, statusMessage) {
-    const { callback, opaque, abort, context, responseHeaders, highWaterMark } = this
+    const { callback, opaque, context, responseHeaders, highWaterMark } = this
 
     const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
 
@@ -92,7 +93,13 @@ class RequestHandler extends AsyncResource {
     const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
     const contentType = parsedHeaders['content-type']
     const contentLength = parsedHeaders['content-length']
-    const body = new Readable({ resume, abort, contentType, contentLength, highWaterMark })
+    const body = new Readable({
+      resume,
+      abort: this.abort,
+      contentType,
+      contentLength,
+      highWaterMark
+    })
 
     this.callback = null
     this.res = body

lib/api/api-stream.js

@@ -1,10 +1,10 @@
 'use strict'
 
+const assert = require('node:assert')
 const { finished, PassThrough } = require('node:stream')
 const {
   InvalidArgumentError,
-  InvalidReturnValueError,
-  RequestAbortedError
+  InvalidReturnValueError
 } = require('../core/errors')
 const util = require('../core/util')
 const { getResolveErrorBodyCallback } = require('./util')
@@ -70,10 +70,13 @@ class StreamHandler extends AsyncResource {
   }
 
   onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      abort(this.reason)
+      return
     }
 
+    assert(this.callback)
+
     this.abort = abort
     this.context = context
   }

lib/api/api-upgrade.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const { InvalidArgumentError, SocketError } = require('../core/errors')
 const { AsyncResource } = require('node:async_hooks')
 const util = require('../core/util')
 const { addSignal, removeSignal } = require('./abort-signal')
@@ -34,10 +34,13 @@ class UpgradeHandler extends AsyncResource {
   }
 
   onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      abort(this.reason)
+      return
     }
 
+    assert(this.callback)
+
     this.abort = abort
     this.context = null
   }

lib/core/request.js

@@ -89,7 +89,7 @@ class Request {
 
     this.method = method
 
-    this.abort = null
+    this.controller = null
 
     if (body == null) {
       this.body = null
@@ -105,8 +105,8 @@ class Request {
       }
 
       this.errorHandler = err => {
-        if (this.abort) {
-          this.abort(err)
+        if (this.controller) {
+          this.controller.abort(err)
         } else {
           this.error = err
         }
@@ -196,7 +196,7 @@ class Request {
       try {
         return this[kHandler].onBodySent(chunk)
       } catch (err) {
-        this.abort(err)
+        this.controller.abort(err)
       }
     }
   }
@@ -210,59 +210,102 @@ class Request {
       try {
         return this[kHandler].onRequestSent()
       } catch (err) {
-        this.abort(err)
+        this.controller.abort(err)
       }
     }
   }
 
   onConnect (abort) {
+    assert(abort)
     assert(!this.aborted)
     assert(!this.completed)
 
+    // TODO(fix): Hack for compatibility with the old API.
+    // Remove and use proper class with encapsulation in
+    // semver major.
+    this.controller = abort.controller = {
+      paused: false,
+      aborted: false,
+      _internal: null,
+      _abort: abort,
+      abort (err) {
+        this.aborted = true
+        this._abort(err)
+      },
+      pause () {
+        this.paused = true
+        this._internal?.pause?.()
+      },
+      resume () {
+        this.paused = false
+        this._internal?.resume?.()
+      }
+    }
+
+    // TODO (fix): This should be enabled. Needs some thinking
+    // in regards to pipeline retries.
+    // if (this.controller) {
+    //   this.controller.abort()
+    //   this.controller = null
+    // }
+
     if (this.error) {
       abort(this.error)
     } else {
-      this.abort = abort
-      return this[kHandler].onConnect(abort)
+      this[kHandler].onConnect(abort)
     }
   }
 
-  onResponseStarted () {
-    return this[kHandler].onResponseStarted?.()
+  onStart () {
+    assert(!this.aborted)
+    assert(!this.completed)
+    assert(this.controller)
+
+    try {
+      this[kHandler].onResponseStarted?.()
+    } catch (err) {
+      this.controller.abort(err)
+    }
   }
 
-  onHeaders (statusCode, headers, resume, statusText) {
+  onHeaders (statusCode, statusText, headers) {
     assert(!this.aborted)
     assert(!this.completed)
+    assert(this.controller)
 
     if (channels.headers.hasSubscribers) {
       channels.headers.publish({ request: this, response: { statusCode, headers, statusText } })
     }
 
     try {
-      return this[kHandler].onHeaders(statusCode, headers, resume, statusText)
+      if (this[kHandler].onHeaders?.(statusCode, headers, () => this.controller.resume(), statusText) === false) {
+        this.controller.pause()
+      }
     } catch (err) {
-      this.abort(err)
+      this.controller.abort(err)
     }
   }
 
   onData (chunk) {
     assert(!this.aborted)
     assert(!this.completed)
+    assert(this.controller)
 
     try {
-      return this[kHandler].onData(chunk)
+      if (this[kHandler].onData?.(chunk) === false) {
+        this.controller.pause()
+      }
     } catch (err) {
-      this.abort(err)
-      return false
+      this.controller.abort(err)
     }
   }
 
   onUpgrade (statusCode, headers, socket) {
     assert(!this.aborted)
     assert(!this.completed)
+    assert(this.controller)
 
-    return this[kHandler].onUpgrade(statusCode, headers, socket)
+    this[kHandler].onUpgrade(statusCode, headers, socket)
   }
 
   onComplete (trailers) {
@@ -276,9 +319,8 @@ class Request {
     }
 
     try {
-      return this[kHandler].onComplete(trailers)
+      this[kHandler].onComplete(trailers)
     } catch (err) {
-      // TODO (fix): This might be a bad idea?
       this.onError(err)
     }
   }
@@ -295,7 +337,7 @@ class Request {
     }
     this.aborted = true
 
-    return this[kHandler].onError(error)
+    this[kHandler].onError(error)
   }
 
   onFinally () {

lib/dispatcher/client-h1.js

@@ -325,7 +325,12 @@ class Parser {
     if (!request) {
       return -1
     }
-    request.onResponseStarted()
+
+    request.controller._internal = this
+
+    request.onStart()
+
+    return request.controller.paused ? constants.ERROR.PAUSED : 0
   }
 
   onHeaderField (buf) {
@@ -504,7 +509,7 @@ class Parser {
       socket[kReset] = true
     }
 
-    const pause = request.onHeaders(statusCode, headers, this.resume, statusText) === false
+    request.onHeaders(statusCode, statusText, headers)
 
     if (request.aborted) {
       return -1
@@ -523,7 +528,7 @@ class Parser {
       client[kResume]()
     }
 
-    return pause ? constants.ERROR.PAUSED : 0
+    return request.controller.paused ? constants.ERROR.PAUSED : 0
   }
 
   onBody (buf) {
@@ -553,9 +558,9 @@ class Parser {
 
     this.bytesRead += buf.length
 
-    if (request.onData(buf) === false) {
-      return constants.ERROR.PAUSED
-    }
+    request.onData(buf)
+
+    return request.controller.paused ? constants.ERROR.PAUSED : 0
   }
 
   onMessageComplete () {