fix: return blocks from putmany as blocks are passed in

Author: achingbrain

package.json

@@ -73,6 +73,8 @@
     "ipfs-repo-migrations": "^0.2.1",
     "ipfs-utils": "^2.2.0",
     "ipld-block": "^0.9.1",
+    "it-map": "^1.0.2",
+    "it-pipe": "^1.1.0",
     "just-safe-get": "^2.0.0",
     "just-safe-set": "^2.1.0",
     "multibase": "^0.7.0",

src/blockstore.js

@@ -3,7 +3,9 @@
 const core = require('datastore-core')
 const ShardingStore = core.ShardingDatastore
 const Block = require('ipld-block')
-const { cidToKey } = require('./blockstore-utils')
+const { cidToKey, keyToCid } = require('./blockstore-utils')
+const map = require('it-map')
+const pipe = require('it-pipe')
 
 module.exports = async (filestore, options) => {
   const store = await maybeWithSharding(filestore, options)
@@ -77,7 +79,7 @@ function createBaseStore (store) {
      *
      * @param {Block} block
      * @param {Object} options
-     * @returns {Promise<void>}
+     * @returns {Promise<Block>}
      */
     async put (block, options) {
       if (!Block.isBlock(block)) {
@@ -100,20 +102,28 @@ function createBaseStore (store) {
      *
      * @param {AsyncIterable<Block>|Iterable<Block>} blocks
      * @param {Object} options
-     * @returns {Promise<void>}
+     * @returns {AsyncIterable<Block>}
      */
     async * putMany (blocks, options) { // eslint-disable-line require-await
-      yield * store.putMany((async function * () {
-        for await (const block of blocks) {
-          const key = cidToKey(block.cid)
-
-          if (await store.has(key, options)) {
-            continue
-          }
-
-          yield { key, value: block.data }
+      yield * pipe(
+        blocks,
+        (source) => {
+          // turn them into a key/value pair
+          return map(source, (block) => {
+            return { key: cidToKey(block.cid), value: block.data }
+          })
+        },
+        (source) => {
+          // put them into the datastore
+          return store.putMany(source, options)
+        },
+        (source) => {
+          // map the returned key/value back into a block
+          return map(source, ({ key, value }) => {
+            return new Block(value, keyToCid(key))
+          })
         }
-      }()), options)
+      )
     },
     /**
      * Does the store contain block with this cid?

test/blockstore-test.js

@@ -73,52 +73,6 @@ module.exports = (repo) => {
         }
       })
 
-      it('should not .putMany when block is already present', async () => {
-        const data = Buffer.from(`TEST${Date.now()}`)
-        const hash = await multihashing(data, 'sha2-256')
-        const cid = new CID(hash)
-        const sent = []
-        otherRepo = new IPFSRepo(tempDir(), {
-          storageBackends: {
-            blocks: class ExplodingBlockStore {
-              open () {
-              }
-
-              close () {
-
-              }
-
-              has () {
-                return true
-              }
-
-              async * putMany (source) {
-                for await (const thing of source) {
-                  sent.push(thing)
-
-                  yield thing
-                }
-              }
-            }
-          },
-          storageBackendOptions: {
-            blocks: {
-              sharding: false
-            }
-          }
-        })
-
-        await otherRepo.init({})
-        await otherRepo.open()
-
-        await drain(otherRepo.blocks.putMany([{
-          cid,
-          data
-        }]))
-
-        expect(sent).to.have.lengthOf(0)
-      })
-
       it('returns an error on invalid block', () => {
         return expect(repo.blocks.put('hello')).to.eventually.be.rejected()
       })