Abort signal

lib/client.js

@@ -67,13 +67,17 @@ class Parser extends HTTPParser {
     assert(!this.read)
     assert(!this.body)
 
-    let body = request.callback(null, {
-      statusCode,
-      headers: parseHeaders(headers),
-      opaque: request.opaque
-    }, resumeSocket)
-    request.callback = null
-    request.opaque = null
+    let body
+
+    if (request.callback) {
+      body = request.callback(null, {
+        statusCode,
+        headers: parseHeaders(headers),
+        opaque: request.opaque
+      }, resumeSocket)
+      request.callback = null
+      request.opaque = null
+    }
 
     if (body && skipBody) {
       body(null, null)
@@ -82,6 +86,17 @@ class Parser extends HTTPParser {
 
     if (body) {
       this.body = body
+
+      if (request.signal) {
+        const onAbort = () => {
+          body(new RequestAbortedError(), null)
+        }
+        if ('addEventListener' in request.signal) {
+          request.signal.addEventListener('abort', onAbort)
+        } else {
+          request.signal.once('abort', onAbort)
+        }
+      }
     } else {
       this.next()
     }
@@ -702,9 +717,18 @@ function resume (client) {
     body,
     chunked,
     rawHeaders,
-    idempotent
+    idempotent,
+    callback,
+    signal
   } = client[kQueue][client[kInflight]]
 
+  if (!callback) {
+    // Request was aborted.
+    client[kQueue].splice(client[kInflight], 1)
+    resume(client)
+    return
+  }
+
   if (!idempotent && client.running) {
     // Non-idempotent request cannot be retried.
     // Ensure that no other requests are inflight and
@@ -767,19 +791,30 @@ function resume (client) {
     const onDrain = () => {
       body.resume()
     }
-    const onClose = () => {
+    const onAbort = () => {
       onFinished(new RequestAbortedError())
     }
+    const onSocketClose = () => {
+      onFinished(new SocketError('other side close'))
+    }
     const onFinished = (err) => {
+      if (signal) {
+        if ('removeEventListener' in signal) {
+          signal.removeEventListener('abort', onAbort)
+        } else {
+          signal.removeListener('abort', onAbort)
+        }
+      }
+
       socket
         .removeListener('drain', onDrain)
         .removeListener('error', onFinished)
-        .removeListener('close', onClose)
+        .removeListener('close', onSocketClose)
       body
         .removeListener('data', onData)
         .removeListener('end', onFinished)
         .removeListener('error', onFinished)
-        .removeListener('close', onClose)
+        .removeListener('close', onAbort)
         .on('error', nop)
 
       if (err) {
@@ -805,16 +840,24 @@ function resume (client) {
       }
     }
 
+    if (signal) {
+      if ('addEventListener' in signal) {
+        signal.addEventListener('abort', onAbort)
+      } else {
+        signal.on('abort', onAbort)
+      }
+    }
+
     body
       .on('data', onData)
       .on('end', onFinished)
       .on('error', onFinished)
-      .on('close', onClose)
+      .on('close', onAbort)
 
     socket
       .on('drain', onDrain)
       .on('error', onFinished)
-      .on('close', onClose)
+      .on('close', onSocketClose)
       .uncork()
 
     client[kWriting] = true

lib/request.js

@@ -1,7 +1,10 @@
 'use strict'
 
 const { AsyncResource } = require('async_hooks')
-const { InvalidArgumentError } = require('./errors')
+const {
+  InvalidArgumentError,
+  RequestAbortedError
+} = require('./errors')
 
 const methods = [
   'ACL',
@@ -54,7 +57,11 @@ class Request extends AsyncResource {
   constructor (opts, callback) {
     super('UNDICI_REQ')
 
-    const { path, method, body, headers, idempotent, opaque } = opts
+    if (!opts) {
+      throw new InvalidArgumentError('no options passed')
+    }
+
+    const { path, method, body, headers, idempotent, opaque, signal } = opts
 
     if (!(typeof path === 'string' && path[0] === '/')) {
       throw new InvalidArgumentError('path must be a valid path')
@@ -64,10 +71,16 @@ class Request extends AsyncResource {
       throw new InvalidArgumentError('method must be a valid method')
     }
 
+    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
+      throw new InvalidArgumentError('signal must implement .on(name, callback)')
+    }
+
     if (!isValidBody(body)) {
       throw new InvalidArgumentError('body must be a string, a Buffer or a Readable stream')
     }
 
+    this.signal = signal
+
     this.method = method
 
     this.path = path
@@ -88,6 +101,27 @@ class Request extends AsyncResource {
 
     this.rawHeaders = ''
 
+    if (this.signal) {
+      const onAbort = () => {
+        const { callback } = this
+
+        if (!callback) {
+          return
+        }
+
+        this.callback = null
+        this.opaque = null
+
+        callback(new RequestAbortedError(), null)
+      }
+
+      if ('addEventListener' in this.signal) {
+        this.signal.addEventListener('abort', onAbort)
+      } else {
+        this.signal.once('abort', onAbort)
+      }
+    }
+
     if (headers) {
       const headerNames = Object.keys(headers)
       for (let i = 0; i < headerNames.length; i++) {

package.json

@@ -17,6 +17,7 @@
   },
   "homepage": "https://github.com/mcollina/undici#readme",
   "devDependencies": {
+    "abort-controller": "^3.0.0",
     "benchmark": "^2.1.4",
     "https-pem": "^2.0.0",
     "pre-commit": "^1.2.2",