feat: support passing AbortSignals to the configured repo (#89)

This is so the user can signal that they are no longer interested
in the results of the operation and system components can stop
trying to fulfil it.

package.json

@@ -30,8 +30,10 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs-block-service#readme",
   "devDependencies": {
+    "abort-controller": "^3.0.0",
     "aegir": "^21.8.1",
     "chai": "^4.2.0",
+    "chai-as-promised": "^7.1.1",
     "cids": "^0.8.0",
     "dirty-chai": "^2.0.1",
     "fs-extra": "^9.0.0",

src/index.js

@@ -54,59 +54,67 @@ class BlockService {
    * Put a block to the underlying datastore.
    *
    * @param {Block} block
+   * @param {Object} [options] -  Options is an object with the following properties
+   * @param {AbortSignal} [options.signal] - A signal that can be used to abort any long-lived operations that are started as a result of this operation
    * @returns {Promise}
    */
-  put (block) {
+  put (block, options) {
     if (this.hasExchange()) {
-      return this._bitswap.put(block)
+      return this._bitswap.put(block, options)
     } else {
-      return this._repo.blocks.put(block)
+      return this._repo.blocks.put(block, options)
     }
   }
 
   /**
    * Put a multiple blocks to the underlying datastore.
    *
    * @param {Array<Block>} blocks
+   * @param {Object} [options] -  Options is an object with the following properties
+   * @param {AbortSignal} [options.signal] - A signal that can be used to abort any long-lived operations that are started as a result of this operation
    * @returns {Promise}
    */
-  putMany (blocks) {
+  putMany (blocks, options) {
     if (this.hasExchange()) {
-      return this._bitswap.putMany(blocks)
+      return this._bitswap.putMany(blocks, options)
     } else {
-      return this._repo.blocks.putMany(blocks)
+      return this._repo.blocks.putMany(blocks, options)
     }
   }
 
   /**
    * Get a block by cid.
    *
    * @param {CID} cid
+   * @param {Object} [options] -  Options is an object with the following properties
+   * @param {AbortSignal} [options.signal] - A signal that can be used to abort any long-lived operations that are started as a result of this operation
    * @returns {Promise<Block>}
    */
-  get (cid) {
+  get (cid, options) {
     if (this.hasExchange()) {
-      return this._bitswap.get(cid)
+      return this._bitswap.get(cid, options)
     } else {
-      return this._repo.blocks.get(cid)
+      return this._repo.blocks.get(cid, options)
     }
   }
 
   /**
    * Get multiple blocks back from an array of cids.
    *
    * @param {Array<CID>} cids
+   * @param {Object} [options] -  Options is an object with the following properties
+   * @param {AbortSignal} [options.signal] - A signal that can be used to abort any long-lived operations that are started as a result of this operation
    * @returns {Iterator<Block>}
    */
-  getMany (cids) {
+  getMany (cids, options) {
     if (!Array.isArray(cids)) {
       throw new Error('first arg must be an array of cids')
     }
 
     if (this.hasExchange()) {
-      return this._bitswap.getMany(cids)
+      return this._bitswap.getMany(cids, options)
     } else {
-      const getRepoBlocks = map((cid) => this._repo.blocks.get(cid))
+      const getRepoBlocks = map((cid) => this._repo.blocks.get(cid, options))
       return getRepoBlocks(cids)
     }
   }
@@ -115,10 +123,12 @@ class BlockService {
    * Delete a block from the blockstore.
    *
    * @param {CID} cid
+   * @param {Object} [options] -  Options is an object with the following properties
+   * @param {AbortSignal} [options.signal] - A signal that can be used to abort any long-lived operations that are started as a result of this operation
    * @returns {Promise}
    */
-  delete (cid) {
-    return this._repo.blocks.delete(cid)
+  delete (cid, options) {
+    return this._repo.blocks.delete(cid, options)
   }
 }
 

test/aborting-requests.spec.js

@@ -0,0 +1,86 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+chai.use(require('dirty-chai'))
+chai.use(require('chai-as-promised'))
+const expect = chai.expect
+
+const { collect } = require('streaming-iterables')
+const AbortController = require('abort-controller')
+
+const BlockService = require('../src')
+
+describe('aborting requests', () => {
+  let abortedErr
+  let r
+
+  beforeEach(() => {
+    abortedErr = new Error('Aborted!')
+    const abortOnSignal = (...args) => {
+      const { signal } = args[args.length - 1]
+
+      return new Promise((resolve, reject) => {
+        signal.addEventListener('abort', () => {
+          reject(abortedErr)
+        })
+      })
+    }
+
+    const repo = {
+      blocks: {
+        put: abortOnSignal,
+        putMany: abortOnSignal,
+        get: abortOnSignal,
+        delete: abortOnSignal,
+        deleteMany: abortOnSignal
+      }
+    }
+    r = new BlockService(repo)
+  })
+
+  it('put - supports abort signals', async () => {
+    const controller = new AbortController()
+    setTimeout(() => controller.abort(), 1)
+
+    await expect(r.put('block', {
+      signal: controller.signal
+    })).to.eventually.rejectedWith(abortedErr)
+  })
+
+  it('putMany - supports abort signals', async () => {
+    const controller = new AbortController()
+    setTimeout(() => controller.abort(), 1)
+
+    await expect(r.putMany(['block'], {
+      signal: controller.signal
+    })).to.eventually.rejectedWith(abortedErr)
+  })
+
+  it('get - supports abort signals', async () => {
+    const controller = new AbortController()
+    setTimeout(() => controller.abort(), 1)
+
+    await expect(r.get('cid', {
+      signal: controller.signal
+    })).to.eventually.rejectedWith(abortedErr)
+  })
+
+  it('getMany - supports abort signals', async () => {
+    const controller = new AbortController()
+    setTimeout(() => controller.abort(), 1)
+
+    await expect(collect(r.getMany(['cid'], {
+      signal: controller.signal
+    }))).to.eventually.rejectedWith(abortedErr)
+  })
+
+  it('remove - supports abort signals', async () => {
+    const controller = new AbortController()
+    setTimeout(() => controller.abort(), 1)
+
+    await expect(r.delete('cid', {
+      signal: controller.signal
+    })).to.eventually.rejectedWith(abortedErr)
+  })
+})