fix: use streaming methods where possible

package.json

@@ -47,9 +47,12 @@
   "dependencies": {
     "debug": "^4.1.1",
     "interface-datastore": "^5.1.1",
+    "it-drain": "^1.0.4",
     "it-filter": "^1.0.2",
     "it-map": "^1.0.5",
     "it-merge": "^1.0.1",
+    "it-pipe": "^1.1.0",
+    "it-pushable": "^1.4.2",
     "it-take": "^1.0.1",
     "uint8arrays": "^2.1.5"
   },

src/keytransform.js

@@ -2,6 +2,7 @@
 
 const { Adapter } = require('interface-datastore')
 const map = require('it-map')
+const { pipe } = require('it-pipe')
 
 /**
  * @typedef {import('interface-datastore').Datastore} Datastore
@@ -10,9 +11,15 @@ const map = require('it-map')
  * @typedef {import('interface-datastore').Query} Query
  * @typedef {import('interface-datastore').KeyQuery} KeyQuery
  * @typedef {import('interface-datastore').Key} Key
+ * @typedef {import('interface-datastore').Pair} Pair
  * @typedef {import('./types').KeyTransform} KeyTransform
  */
 
+/**
+ * @template TEntry
+ * @typedef {import('interface-store').AwaitIterable<TEntry>} AwaitIterable
+ */
+
 /**
  * A datastore shim, that wraps around a given datastore, changing
  * the way keys look to the user, for example namespacing
@@ -69,6 +76,78 @@ class KeyTransformDatastore extends Adapter {
     return this.child.delete(this.transform.convert(key), options)
   }
 
+  /**
+   * @param {AwaitIterable<Pair>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Pair>}
+   */
+  async * putMany (source, options = {}) {
+    const transform = this.transform
+    const child = this.child
+
+    yield * pipe(
+      source,
+      async function * (source) {
+        yield * map(source, ({ key, value }) => ({
+          key: transform.convert(key),
+          value
+        }))
+      },
+      async function * (source) {
+        yield * child.putMany(source, options)
+      },
+      async function * (source) {
+        yield * map(source, ({ key, value }) => ({
+          key: transform.invert(key),
+          value
+        }))
+      }
+    )
+  }
+
+  /**
+   * @param {AwaitIterable<Key>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Uint8Array>}
+   */
+  async * getMany (source, options = {}) {
+    const transform = this.transform
+    const child = this.child
+
+    yield * pipe(
+      source,
+      async function * (source) {
+        yield * map(source, key => transform.convert(key))
+      },
+      async function * (source) {
+        yield * child.getMany(source, options)
+      }
+    )
+  }
+
+  /**
+   * @param {AwaitIterable<Key>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Key>}
+   */
+  async * deleteMany (source, options = {}) {
+    const transform = this.transform
+    const child = this.child
+
+    yield * pipe(
+      source,
+      async function * (source) {
+        yield * map(source, key => transform.convert(key))
+      },
+      async function * (source) {
+        yield * child.deleteMany(source, options)
+      },
+      async function * (source) {
+        yield * map(source, key => transform.invert(key))
+      }
+    )
+  }
+
   /**
    * @returns {Batch}
    */

src/mount.js

@@ -22,11 +22,6 @@ const Keytransform = require('./keytransform')
  * @typedef {import('./types').KeyTransform} KeyTransform
  */
 
-/**
- * @template TEntry
- * @typedef {import('./types').AwaitIterable<TEntry>} AwaitIterable
- */
-
 /**
  * A datastore that can combine multiple stores inside various
  * key prefixes

src/sharding.js

@@ -22,7 +22,12 @@ const shardReadmeKey = new Key(sh.README_FN)
  */
 /**
  * @template TValue
- * @typedef {import('./types').Await<TValue> } Await
+ * @typedef {import('interface-store').Await<TValue> } Await
+ */
+
+/**
+ * @template TEntry
+ * @typedef {import('interface-store').AwaitIterable<TEntry>} AwaitIterable
  */
 
 /**
@@ -162,6 +167,33 @@ class ShardingDatastore extends Adapter {
     return this.child.delete(key, options)
   }
 
+  /**
+   * @param {AwaitIterable<Pair>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Pair>}
+   */
+  async * putMany (source, options = {}) {
+    yield * this.child.putMany(source, options)
+  }
+
+  /**
+   * @param {AwaitIterable<Key>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Uint8Array>}
+   */
+  async * getMany (source, options = {}) {
+    yield * this.child.getMany(source, options)
+  }
+
+  /**
+   * @param {AwaitIterable<Key>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Key>}
+   */
+  async * deleteMany (source, options = {}) {
+    yield * this.child.deleteMany(source, options)
+  }
+
   batch () {
     return this.child.batch()
   }

src/tiered.js

@@ -2,13 +2,22 @@
 
 const { Adapter, Errors } = require('interface-datastore')
 const log = require('debug')('datastore:core:tiered')
+const pushable = require('it-pushable')
+const drain = require('it-drain')
+
 /**
  * @typedef {import('interface-datastore').Datastore} Datastore
  * @typedef {import('interface-datastore').Options} Options
  * @typedef {import('interface-datastore').Batch} Batch
  * @typedef {import('interface-datastore').Query} Query
  * @typedef {import('interface-datastore').KeyQuery} KeyQuery
  * @typedef {import('interface-datastore').Key} Key
+ * @typedef {import('interface-datastore').Pair} Pair
+ */
+
+/**
+ * @template TEntry
+ * @typedef {import('interface-store').AwaitIterable<TEntry>} AwaitIterable
  */
 
 /**
@@ -90,6 +99,74 @@ class TieredDatastore extends Adapter {
     }
   }
 
+  /**
+   * @param {AwaitIterable<Pair>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Pair>}
+   */
+  async * putMany (source, options = {}) {
+    let error
+    const pushables = this.stores.map(store => {
+      const source = pushable()
+
+      drain(store.putMany(source, options))
+        .catch(err => {
+          // store threw while putting, make sure we bubble the error up
+          error = err
+        })
+
+      return source
+    })
+
+    try {
+      for await (const pair of source) {
+        if (error) {
+          throw error
+        }
+
+        pushables.forEach(p => p.push(pair))
+
+        yield pair
+      }
+    } finally {
+      pushables.forEach(p => p.end())
+    }
+  }
+
+  /**
+   * @param {AwaitIterable<Key>} source
+   * @param {Options} [options]
+   * @returns {AsyncIterable<Key>}
+   */
+  async * deleteMany (source, options = {}) {
+    let error
+    const pushables = this.stores.map(store => {
+      const source = pushable()
+
+      drain(store.deleteMany(source, options))
+        .catch(err => {
+          // store threw while deleting, make sure we bubble the error up
+          error = err
+        })
+
+      return source
+    })
+
+    try {
+      for await (const key of source) {
+        if (error) {
+          throw error
+        }
+
+        pushables.forEach(p => p.push(key))
+
+        yield key
+      }
+    } finally {
+      pushables.forEach(p => p.end())
+    }
+  }
+
   async close () {
     await Promise.all(this.stores.map(store => store.close()))
   }