feat: add streaming/cancellable API

package.json

@@ -38,15 +38,15 @@
   },
   "homepage": "https://github.com/ipfs/js-datastore-core#readme",
   "devDependencies": {
-    "aegir": "^21.4.5",
+    "aegir": "^22.0.0",
     "async-iterator-all": "^1.0.0",
     "chai": "^4.2.0",
     "dirty-chai": "^2.0.1"
   },
   "dependencies": {
     "buffer": "^5.5.0",
     "debug": "^4.1.1",
-    "interface-datastore": "^0.8.2"
+    "interface-datastore": "^1.0.2"
   },
   "engines": {
     "node": ">=6.0.0",

src/keytransform.js

@@ -1,15 +1,17 @@
 'use strict'
 
-const utils = require('interface-datastore').utils
+const { Adapter, utils } = require('interface-datastore')
 const map = utils.map
 
 /**
  * A datastore shim, that wraps around a given datastore, changing
  * the way keys look to the user, for example namespacing
  * keys, reversing them, etc.
  */
-class KeyTransformDatastore {
+class KeyTransformDatastore extends Adapter {
   constructor (child, transform) {
+    super()
+
     this.child = child
     this.transform = transform
   }
@@ -18,20 +20,20 @@ class KeyTransformDatastore {
     return this.child.open()
   }
 
-  put (key, val) {
-    return this.child.put(this.transform.convert(key), val)
+  put (key, val, options) {
+    return this.child.put(this.transform.convert(key), val, options)
   }
 
-  get (key) {
-    return this.child.get(this.transform.convert(key))
+  get (key, options) {
+    return this.child.get(this.transform.convert(key), options)
   }
 
-  has (key) {
-    return this.child.has(this.transform.convert(key))
+  has (key, options) {
+    return this.child.has(this.transform.convert(key), options)
   }
 
-  delete (key) {
-    return this.child.delete(this.transform.convert(key))
+  delete (key, options) {
+    return this.child.delete(this.transform.convert(key), options)
   }
 
   batch () {
@@ -43,14 +45,14 @@ class KeyTransformDatastore {
       delete: (key) => {
         b.delete(this.transform.convert(key))
       },
-      commit: () => {
-        return b.commit()
+      commit: (options) => {
+        return b.commit(options)
       }
     }
   }
 
-  query (q) {
-    return map(this.child.query(q), e => {
+  query (q, options) {
+    return map(this.child.query(q, options), e => {
       e.key = this.transform.invert(e.key)
       return e
     })

src/mount.js

@@ -1,22 +1,25 @@
 /* @flow */
 'use strict'
 
-const Key = require('interface-datastore').Key
-const Errors = require('interface-datastore').Errors
-const utils = require('interface-datastore').utils
-const filter = utils.filter
-const take = utils.take
-const sortAll = utils.sortAll
-const replaceStartWith = utils.replaceStartWith
+const {
+  Adapter, Key, Errors, utils: {
+    filter,
+    take,
+    sortAll,
+    replaceStartWith
+  }
+} = require('interface-datastore')
 
 const Keytransform = require('./keytransform')
 
 /**
  * A datastore that can combine multiple stores inside various
  * key prefixs.
  */
-class MountDatastore {
+class MountDatastore extends Adapter {
   constructor (mounts) {
+    super()
+
     this.mounts = mounts.slice()
   }
 
@@ -44,38 +47,38 @@ class MountDatastore {
     }
   }
 
-  put (key, value) {
+  put (key, value, options) {
     const match = this._lookup(key)
     if (match == null) {
       throw Errors.dbWriteFailedError(new Error('No datastore mounted for this key'))
     }
 
-    return match.datastore.put(match.rest, value)
+    return match.datastore.put(match.rest, value, options)
   }
 
-  get (key) {
+  get (key, options) {
     const match = this._lookup(key)
     if (match == null) {
       throw Errors.notFoundError(new Error('No datastore mounted for this key'))
     }
-    return match.datastore.get(match.rest)
+    return match.datastore.get(match.rest, options)
   }
 
-  has (key) {
+  has (key, options) {
     const match = this._lookup(key)
     if (match == null) {
       return false
     }
-    return match.datastore.has(match.rest)
+    return match.datastore.has(match.rest, options)
   }
 
-  delete (key) {
+  delete (key, options) {
     const match = this._lookup(key)
     if (match == null) {
       throw Errors.dbDeleteFailedError(new Error('No datastore mounted for this key'))
     }
 
-    return match.datastore.delete(match.rest)
+    return match.datastore.delete(match.rest, options)
   }
 
   close () {
@@ -112,13 +115,13 @@ class MountDatastore {
         const match = lookup(key)
         match.batch.delete(match.rest)
       },
-      commit: () => {
-        return Promise.all(Object.keys(batchMounts).map(p => batchMounts[p].commit()))
+      commit: (options) => {
+        return Promise.all(Object.keys(batchMounts).map(p => batchMounts[p].commit(options)))
       }
     }
   }
 
-  query (q) {
+  query (q, options) {
     const qs = this.mounts.map(m => {
       const ks = new Keytransform(m.datastore, {
         convert: (key) => {
@@ -138,7 +141,7 @@ class MountDatastore {
         prefix: prefix,
         filters: q.filters,
         keysOnly: q.keysOnly
-      })
+      }, options)
     })
 
     let it = _many(qs)

src/sharding.js

@@ -1,8 +1,7 @@
 'use strict'
 
 const { Buffer } = require('buffer')
-const Key = require('interface-datastore').Key
-
+const { Adapter, Key } = require('interface-datastore')
 const sh = require('./shard')
 const KeytransformStore = require('./keytransform')
 
@@ -15,8 +14,10 @@ const shardReadmeKey = new Key(sh.README_FN)
  * Wraps another datastore such that all values are stored
  * sharded according to the given sharding function.
  */
-class ShardingDatastore {
+class ShardingDatastore extends Adapter {
   constructor (store, shard) {
+    super()
+
     this.child = new KeytransformStore(store, {
       convert: this._convertKey.bind(this),
       invert: this._invertKey.bind(this)
@@ -75,27 +76,27 @@ class ShardingDatastore {
     throw new Error('datastore exists')
   }
 
-  put (key, val) {
-    return this.child.put(key, val)
+  put (key, val, options) {
+    return this.child.put(key, val, options)
   }
 
-  get (key) {
-    return this.child.get(key)
+  get (key, options) {
+    return this.child.get(key, options)
   }
 
-  has (key) {
-    return this.child.has(key)
+  has (key, options) {
+    return this.child.has(key, options)
   }
 
-  delete (key) {
-    return this.child.delete(key)
+  delete (key, options) {
+    return this.child.delete(key, options)
   }
 
   batch () {
     return this.child.batch()
   }
 
-  query (q) {
+  query (q, options) {
     const tq = {
       keysOnly: q.keysOnly,
       offset: q.offset,
@@ -130,7 +131,7 @@ class ShardingDatastore {
       })
     }
 
-    return this.child.query(tq)
+    return this.child.query(tq, options)
   }
 
   close () {

src/tiered.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Errors = require('interface-datastore').Errors
+const { Adapter, Errors } = require('interface-datastore')
 const log = require('debug')('datastore:core:tiered')
 
 /**
@@ -10,8 +10,10 @@ const log = require('debug')('datastore:core:tiered')
  * last one first.
  *
  */
-class TieredDatastore {
+class TieredDatastore extends Adapter {
   constructor (stores) {
+    super()
+
     this.stores = stores.slice()
   }
 
@@ -31,10 +33,10 @@ class TieredDatastore {
     }
   }
 
-  async get (key) {
+  async get (key, options) {
     for (const store of this.stores) {
       try {
-        const res = await store.get(key)
+        const res = await store.get(key, options)
         if (res) return res
       } catch (err) {
         log(err)
@@ -43,19 +45,19 @@ class TieredDatastore {
     throw Errors.notFoundError()
   }
 
-  async has (key) {
+  async has (key, options) {
     for (const s of this.stores) {
-      if (await s.has(key)) {
+      if (await s.has(key, options)) {
         return true
       }
     }
 
     return false
   }
 
-  async delete (key) {
+  async delete (key, options) {
     try {
-      await Promise.all(this.stores.map(store => store.delete(key)))
+      await Promise.all(this.stores.map(store => store.delete(key, options)))
     } catch (err) {
       throw Errors.dbDeleteFailedError()
     }
@@ -75,16 +77,16 @@ class TieredDatastore {
       delete: (key) => {
         batches.forEach(b => b.delete(key))
       },
-      commit: async () => {
+      commit: async (options) => {
         for (const batch of batches) {
-          await batch.commit()
+          await batch.commit(options)
         }
       }
     }
   }
 
-  query (q) {
-    return this.stores[this.stores.length - 1].query(q)
+  query (q, options) {
+    return this.stores[this.stores.length - 1].query(q, options)
   }
 }