feat: use streaming API for blockstore access (#232)

Uses the streaming APIs from interface-datastore to work with data being sent/fetched from the blockstore.

The previous behaviour was to use batches for putting streams of blocks - this was bad because we'd end up buffering the whole stream before adding blocks to the blockstore.  Instead now we stream the blocks as we get them.

* Adds `repo.blockstore.getMany()` function
* Passes `options` to underlying datastores to allow passing AbortSignals

BREAKING CHANGES:

* `repo.blockstore.putMany()` used to return a promise that resolved when all the deletes were done, now it returns an AsyncIterator that yields `{key, value}` objects as the put progresses
* `repo.blockstore.deleteMany()` used to return a promise that resolved when all the deletes were done, now it returns an AsyncIterator that yields CID objects as the delete progresses

Author: achingbrain

README.md

@@ -30,19 +30,22 @@ This is the implementation of the [IPFS repo spec](https://github.com/ipfs/specs
 - [API](#api)
   - [Setup](#setup)
     - [`new Repo(path[, options])`](#new-repopath-options)
-    - [`Promise repo.init ()`](#promise-repoinit-)
-    - [`Promise repo.open ()`](#promise-repoopen-)
-    - [`Promise repo.close ()`](#promise-repoclose-)
-    - [`Promise<boolean> repo.exists ()`](#promiseboolean-repoexists-)
+    - [`Promise repo.init()`](#promise-repoinit)
+    - [`Promise repo.open()`](#promise-repoopen)
+    - [`Promise repo.close()`](#promise-repoclose)
+    - [`Promise<boolean> repo.exists()`](#promiseboolean-repoexists)
+    - [`Promise<Boolean> repo.isInitialized()`](#promiseboolean-repoisinitialized)
   - [Repos](#repos)
-    - [`Promise repo.put (key, value:Buffer)`](#promise-repoput-key-valuebuffer)
-    - [`Promise<Buffer> repo.get (key)`](#promisebuffer-repoget-key)
-    - [`Promise<Boolean> repo.isInitialized ()`](#promiseboolean-repoisinitialized-)
-    - [`Promise repo.blocks.put (block:Block)`](#promise-repoblocksput-blockblock)
-    - [`Promise repo.blocks.putMany (blocks)`](#promise-repoblocksputmany-blocks)
-    - [`Promise<Buffer> repo.blocks.get (cid)`](#promisebuffer-repoblocksget-cid)
-    - [`Promise repo.blocks.delete (cid:CID)`](#promise-repoblocksdelete-cidcid)
-    - [`Promise repo.blocks.deleteMany (cids)`](#promise-repoblocksdeletemany-cids)
+    - [`Promise repo.put(key, value:Buffer)`](#promise-repoputkey-valuebuffer)
+    - [`Promise<Buffer> repo.get(key)`](#promisebuffer-repogetkey)
+  - [Blocks](#blocks)
+    - [`Promise<Block> repo.blocks.put(block:Block)`](#promiseblock-repoblocksputblockblock)
+    - [`AsyncIterator<Block> repo.blocks.putMany(source)`](#asynciteratorblock-repoblocksputmanysource)
+    - [`Promise<Buffer> repo.blocks.get(cid)`](#promisebuffer-repoblocksgetcid)
+    - [`AsyncIterable<Buffer> repo.blocks.getMany(source)`](#asynciterablebuffer-repoblocksgetmanysource)
+    - [`Promise<CID> repo.blocks.delete(cid:CID)`](#promisecid-repoblocksdeletecidcid)
+    - [`AsyncIterator<CID> repo.blocks.deleteMany(source)`](#asynciteratorcid-repoblocksdeletemanysource)
+  - [Datastore](#datastore)
     - [`repo.datastore`](#repodatastore)
   - [Config](#config)
     - [`Promise repo.config.set(key:string, value)`](#promise-repoconfigsetkeystring-value)
@@ -51,17 +54,17 @@ This is the implementation of the [IPFS repo spec](https://github.com/ipfs/specs
     - [`Promise<Object> repo.config.getAll()`](#promiseobject-repoconfiggetall)
     - [`Promise<boolean> repo.config.exists()`](#promiseboolean-repoconfigexists)
   - [Version](#version)
-    - [`Promise<Number> repo.version.get ()`](#promisenumber-repoversionget-)
+    - [`Promise<Number> repo.version.get()`](#promisenumber-repoversionget)
     - [`Promise repo.version.set (version:Number)`](#promise-repoversionset-versionnumber)
   - [API Addr](#api-addr)
-    - [`Promise<String> repo.apiAddr.get ()`](#promisestring-repoapiaddrget-)
-    - [`Promise repo.apiAddr.set (value)`](#promise-repoapiaddrset-value)
+    - [`Promise<String> repo.apiAddr.get()`](#promisestring-repoapiaddrget)
+    - [`Promise repo.apiAddr.set(value)`](#promise-repoapiaddrsetvalue)
   - [Status](#status)
-    - [`Promise<Object> repo.stat ()`](#promiseobject-repostat-)
+    - [`Promise<Object> repo.stat()`](#promiseobject-repostat)
   - [Lock](#lock)
-    - [`Promise lock.lock (dir)`](#promise-locklock-dir)
-    - [`Promise closer.close ()`](#promise-closerclose-)
-    - [`Promise<boolean> lock.locked (dir)`](#promiseboolean-locklocked-dir)
+    - [`Promise lock.lock(dir)`](#promise-locklockdir)
+    - [`Promise closer.close()`](#promise-closerclose)
+    - [`Promise<boolean> lock.locked(dir)`](#promiseboolean-locklockeddir)
 - [Notes](#notes)
   - [Migrations](#migrations)
 - [Contribute](#contribute)
@@ -184,79 +187,84 @@ Arguments:
 const repo = new Repo('path/to/repo')
 ```
 
-#### `Promise repo.init ()`
+#### `Promise repo.init()`
 
-Creates the necessary folder structure inside the repo.
+Creates the necessary folder structure inside the repo
 
-#### `Promise repo.open ()`
+#### `Promise repo.open()`
 
-[Locks](https://en.wikipedia.org/wiki/Record_locking) the repo to prevent conflicts arising from simultaneous access.
+[Locks](https://en.wikipedia.org/wiki/Record_locking) the repo to prevent conflicts arising from simultaneous access
 
-#### `Promise repo.close ()`
+#### `Promise repo.close()`
 
 Unlocks the repo.
 
-#### `Promise<boolean> repo.exists ()`
+#### `Promise<boolean> repo.exists()`
 
-Tells whether this repo exists or not. Returned promise resolves to a `boolean`.
+Tells whether this repo exists or not. Returned promise resolves to a `boolean`
+
+#### `Promise<Boolean> repo.isInitialized()`
+
+The returned promise resolves to `false` if the repo has not been initialized and `true` if it has
 
 ### Repos
 
 Root repo:
 
-#### `Promise repo.put (key, value:Buffer)`
+#### `Promise repo.put(key, value:Buffer)`
 
-Put a value at the root of the repo.
+Put a value at the root of the repo
 
-* `key` can be a buffer, a string or a [Key](https://github.com/ipfs/interface-datastore#keys).
+* `key` can be a buffer, a string or a [Key][]
 
-#### `Promise<Buffer> repo.get (key)`
+#### `Promise<Buffer> repo.get(key)`
 
-Get a value at the root of the repo.
+Get a value at the root of the repo
 
-* `key` can be a buffer, a string or a [Key](https://github.com/ipfs/interface-datastore#keys).
+* `key` can be a buffer, a string or a [Key][]
 
-[Blocks](https://github.com/ipfs/js-ipld-block#readme):
+### Blocks
 
-#### `Promise<Boolean> repo.isInitialized ()`
+#### `Promise<Block> repo.blocks.put(block:Block)`
 
-The returned promise resolves to `false` if the repo has not been initialized and `true` if it has.
+* `block` should be of type [Block][]
 
-#### `Promise repo.blocks.put (block:Block)`
+#### `AsyncIterator<Block> repo.blocks.putMany(source)`
 
-* `block` should be of type [Block](https://github.com/ipfs/js-ipld-block#readme).
+Put many blocks.
 
-#### `Promise repo.blocks.putMany (blocks)`
+* `source` should be an AsyncIterable that yields entries of type [Block][]
 
-Put many blocks.
+#### `Promise<Buffer> repo.blocks.get(cid)`
 
-* `block` should be an Iterable or AsyncIterable that yields entries of type [Block](https://github.com/ipfs/js-ipld-block#readme).
+Get block.
+
+* `cid` is the content id of type [CID][]
 
-#### `Promise<Buffer> repo.blocks.get (cid)`
+#### `AsyncIterable<Buffer> repo.blocks.getMany(source)`
 
 Get block.
 
-* `cid` is the content id of [type CID](https://github.com/ipld/js-cid#readme).
+* `source` should be an AsyncIterable that yields entries of type [CID][]
 
-#### `Promise repo.blocks.delete (cid:CID)`
+#### `Promise<CID> repo.blocks.delete(cid:CID)`
 
-* `cid` should be of the [type CID](https://github.com/ipld/js-cid#readme).
+* `cid` should be of the type [CID][]
 
 Delete a block
 
-#### `Promise repo.blocks.deleteMany (cids)`
+#### `AsyncIterator<CID> repo.blocks.deleteMany(source)`
 
-* `cids` should be an Iterable or AsyncIterable that yields entries of the [type CID](https://github.com/ipld/js-cid#readme).
+* `source` should be an Iterable or AsyncIterable that yields entries of the type [CID][]
 
 Delete many blocks
 
-Datastore:
+### Datastore
 
 #### `repo.datastore`
 
 This contains a full implementation of [the `interface-datastore` API](https://github.com/ipfs/interface-datastore#api).
 
-
 ### Config
 
 Instead of using `repo.set('config')` this exposes an API that allows you to set and get a decoded config object, as well as, in a safe manner, change any of the config values individually.
@@ -298,7 +306,7 @@ Whether the config sub-repo exists.
 
 ### Version
 
-#### `Promise<Number> repo.version.get ()`
+#### `Promise<Number> repo.version.get()`
 
 Gets the repo version (an integer).
 
@@ -308,19 +316,19 @@ Sets the repo version
 
 ### API Addr
 
-#### `Promise<String> repo.apiAddr.get ()`
+#### `Promise<String> repo.apiAddr.get()`
 
 Gets the API address.
 
-#### `Promise repo.apiAddr.set (value)`
+#### `Promise repo.apiAddr.set(value)`
 
 Sets the API address.
 
-* `value` should be a [Multiaddr](https://github.com/multiformats/js-multiaddr) or a String representing a valid one.
+* `value` should be a [Multiaddr][] or a String representing a valid one.
 
 ### Status
 
-#### `Promise<Object> repo.stat ()`
+#### `Promise<Object> repo.stat()`
 
 Gets the repo status.
 
@@ -343,21 +351,21 @@ const memoryLock = require('ipfs-repo/src/lock-memory')  // Default in browser
 
 You can also provide your own custom Lock. It must be an object with the following interface:
 
-#### `Promise lock.lock (dir)`
+#### `Promise lock.lock(dir)`
 
 Sets the lock if one does not already exist. If a lock already exists, should throw an error.
 
 `dir` is a string to the directory the lock should be created at. The repo typically creates the lock at its root.
 
 Returns `closer`, where `closer` has a `close` method for removing the lock.
 
-#### `Promise closer.close ()`
+#### `Promise closer.close()`
 
 Closes the lock created by `lock.open`
 
 If no error was thrown, the lock was successfully removed.
 
-#### `Promise<boolean> lock.locked (dir)`
+#### `Promise<boolean> lock.locked(dir)`
 
 Checks the existence of the lock.
 
@@ -392,3 +400,8 @@ This repository falls under the IPFS [Code of Conduct](https://github.com/ipfs/c
 ## License
 
 [MIT](LICENSE)
+
+[CID]: https://github.com/multiformats/js-cid
+[Key]: https://github.com/ipfs/interface-datastore#keys
+[Block]: https://github.com/ipld/js-ipld-block
+[Multiaddr]: https://github.com/multiformats/js-multiaddr

package.json

@@ -43,10 +43,13 @@
     "npm": ">=3.0.0"
   },
   "devDependencies": {
-    "aegir": "^21.10.1",
+    "aegir": "^22.0.0",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
     "dirty-chai": "^2.0.1",
+    "it-all": "^1.0.2",
+    "it-drain": "^1.0.1",
+    "it-first": "^1.0.2",
     "just-range": "^2.1.0",
     "memdown": "^5.1.0",
     "multihashes": "~0.4.15",
@@ -60,13 +63,13 @@
     "buffer": "^5.6.0",
     "bytes": "^3.1.0",
     "cids": "^0.8.0",
-    "datastore-core": "^1.0.0",
-    "datastore-fs": "^1.0.0",
-    "datastore-idb": "^1.0.2",
-    "datastore-level": "^1.0.0",
+    "datastore-core": "^1.1.0",
+    "datastore-fs": "^1.1.0",
+    "datastore-idb": "^1.1.0",
+    "datastore-level": "^1.1.0",
     "debug": "^4.1.0",
     "err-code": "^2.0.0",
-    "interface-datastore": "^0.8.3",
+    "interface-datastore": "^1.0.2",
     "ipfs-repo-migrations": "^0.2.1",
     "ipfs-utils": "^2.2.0",
     "ipld-block": "^0.9.1",