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,12 @@ class Parser extends HTTPParser {
 
     if (body) {
       this.body = body
+
+      if (request.signal) {
+        request.signal.once('abort', () => {
+          body(new RequestAbortedError(), null)
+        })
+      }
     } else {
       this.next()
     }
@@ -702,9 +712,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 +786,26 @@ 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) {
+        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 +831,20 @@ function resume (client) {
       }
     }
 
+    if (signal) {
+      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') {
+      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,21 @@ class Request extends AsyncResource {
 
     this.rawHeaders = ''
 
+    if (this.signal) {
+      this.signal.once('abort', () => {
+        const { callback } = this
+
+        if (!callback) {
+          return
+        }
+
+        this.callback = null
+        this.opaque = null
+
+        callback(new RequestAbortedError(), null)
+      })
+    }
+
     if (headers) {
       const headerNames = Object.keys(headers)
       for (let i = 0; i < headerNames.length; i++) {

test/abort.js

@@ -0,0 +1,106 @@
+'use strict'
+
+const { test } = require('tap')
+const EventEmitter = require('events')
+const { Client, errors } = require('..')
+const { createServer } = require('http')
+
+test('Abort while sending request - event emitter (no body)', { skip: 'never ending' }, (t) => {
+  t.plan(1)
+
+  const server = createServer((req, res) => {
+    t.fail('The requets should be aborted')
+  })
+  t.teardown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const ee = new EventEmitter()
+    t.teardown(client.close.bind(client))
+
+    client.request({ path: '/', method: 'GET', signal: ee }, (err, response) => {
+      t.ok(err instanceof errors.RequestAbortedError)
+    })
+
+    ee.emit('abort')
+  })
+})
+
+test('Abort while waiting response - event emitter (no body)', (t) => {
+  t.plan(1)
+
+  const server = createServer((req, res) => {
+    setTimeout(() => {
+      res.setHeader('content-type', 'text/plain')
+      res.end('hello world')
+    }, 1000)
+  })
+  t.teardown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const ee = new EventEmitter()
+    t.teardown(client.close.bind(client))
+
+    client.request({ path: '/', method: 'GET', signal: ee }, (err, response) => {
+      t.ok(err instanceof errors.RequestAbortedError)
+    })
+
+    setTimeout(() => {
+      ee.emit('abort')
+    }, 500)
+  })
+})
+
+test('Abort while waiting response - event emitter (write headers started) (no body)', (t) => {
+  t.plan(1)
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain' })
+    setTimeout(() => {
+      res.end('hello world')
+    }, 1000)
+  })
+  t.teardown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const ee = new EventEmitter()
+    t.teardown(client.close.bind(client))
+
+    client.request({ path: '/', method: 'GET', signal: ee }, (err, response) => {
+      t.ok(err instanceof errors.RequestAbortedError)
+    })
+
+    setTimeout(() => {
+      ee.emit('abort')
+    }, 500)
+  })
+})
+
+test('Abort while waiting response - event emitter (write headers (write body started) (no body)', (t) => {
+  t.plan(1)
+
+  const server = createServer((req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain' })
+    res.write('hello')
+    setTimeout(() => {
+      res.end('world')
+    }, 1000)
+  })
+  t.teardown(server.close.bind(server))
+
+  server.listen(0, () => {
+    const client = new Client(`http://localhost:${server.address().port}`)
+    const ee = new EventEmitter()
+    t.teardown(client.close.bind(client))
+
+    client.request({ path: '/', method: 'GET', signal: ee }, (err, response) => {
+      t.ok(err instanceof errors.RequestAbortedError)
+    })
+
+    setTimeout(() => {
+      ee.emit('abort')
+    }, 500)
+  })
+})