feat: implementation of the new put() function

BREAKING CHANGE: The API of `put()` changes.

The API docs for it:

> Stores the given IPLD Nodes of a recognized IPLD Format.

 - `nodes` (`Iterable<Object>`): deserialized IPLD nodes that should be inserted.
 - `format` (`multicodec`, required): the multicodec of the format that IPLD Node should be encoded in.
 - `options` is applied to any of the `nodes` and is an object with the following properties:
   - `hashAlg` (`multicodec`, default: hash algorithm of the given multicodec): the hashing algorithm that is used to calculate the CID.
   - `cidVersion` (`boolean`, default: 1): the CID version to use.
   - `onlyHash` (`boolean`, default: false): if true the serialized form of the IPLD Node will not be passed to the underlying block store.

Returns an async iterator with the CIDs of the serialized IPLD Nodes.

Author: vmx

package.json

@@ -64,7 +64,8 @@
     "multicodec": "~0.5.0",
     "pull-defer": "~0.2.3",
     "pull-stream": "^3.6.9",
-    "pull-traverse": "^1.0.3"
+    "pull-traverse": "^1.0.3",
+    "typical": "^3.0.0"
   },
   "contributors": [
     "Alan Shaw <[email protected]>",

src/index.js

@@ -12,6 +12,7 @@ const ipldDagCbor = require('ipld-dag-cbor')
 const ipldDagPb = require('ipld-dag-pb')
 const ipldRaw = require('ipld-raw')
 const multicodec = require('multicodec')
+const typical = require('typical')
 const { fancyIterator } = require('./util')
 
 function noop () {}
@@ -170,40 +171,82 @@ class IPLDResolver {
     })
   }
 
-  put (node, options, callback) {
-    if (typeof options === 'function') {
-      callback = options
-      return setImmediate(() => callback(
-        new Error('IPLDResolver.put requires options')
-      ))
+  /**
+   * Stores the given IPLD Nodes of a recognized IPLD Format.
+   *
+   * @param {Iterable.<Object>} nodes - Deserialized IPLD nodes that should be inserted.
+   * @param {number} format - The multicodec of the format that IPLD Node should be encoded in.
+   * @param {Object} [userOptions] -  Options are applied to any of the `nodes` and is an object with the following properties.
+   * @param {number} [userOtions.hashAlg=hash algorithm of the given multicodec] - The hashing algorithm that is used to calculate the CID.
+   * @param {number} [userOptions.cidVersion=1]`- The CID version to use.
+   * @param {boolean} [userOptions.onlyHash=false] - If true the serialized form of the IPLD Node will not be passed to the underlying block store.
+   * @returns {Iterable.<Promise.<CID>>} - Returns an async iterator with the CIDs of the serialized IPLD Nodes.
+   */
+  put (nodes, format, userOptions) {
+    if (!typical.isIterable(nodes) || typical.isString(nodes) ||
+        Buffer.isBuffer(nodes)) {
+      throw new Error('`nodes` must be an iterable')
+    }
+    if (format === undefined) {
+      throw new Error('`put` requires a format')
+    }
+    if (typeof format !== 'number') {
+      throw new Error('`format` parameter must be number (multicodec)')
     }
-    callback = callback || noop
 
-    if (options.cid && CID.isCID(options.cid)) {
-      if (options.onlyHash) {
-        return setImmediate(() => callback(null, options.cid))
-      }
+    let options
+    let formatImpl
 
-      return this._put(options.cid, node, callback)
-    }
+    const next = () => {
+      // End iteration if there are no more nodes to put
+      if (nodes.length === 0) {
+        return Promise.resolve({ done: true })
+      }
 
-    // TODO vmx 2018-12-07: Make this async/await once `put()` returns a
-    // Promise
-    this._getFormat(options.format).then((format) => {
-      format.util.cid(node, options, (err, cid) => {
-        if (err) {
-          return callback(err)
+      return new Promise(async (resolve, reject) => {
+        // Lazy load the options not when the iterator is initialized, but
+        // when we hit the first iteration. This way the constructor can be
+        // a synchronous function.
+        if (options === undefined) {
+          try {
+            formatImpl = await this._getFormat(format)
+          } catch (err) {
+            return reject(err)
+          }
+          const defaultOptions = {
+            hashAlg: formatImpl.defaultHashAlg,
+            cidVersion: 1,
+            onlyHash: false
+          }
+          options = mergeOptions(defaultOptions, userOptions)
         }
 
-        if (options.onlyHash) {
-          return callback(null, cid)
+        const node = nodes.shift()
+        const cidOptions = {
+          version: options.cidVersion,
+          hashAlg: options.hashAlg,
+          onlyHash: options.onlyHash
         }
+        formatImpl.util.cid(node, cidOptions, (err, cid) => {
+          if (err) {
+            return reject(err)
+          }
 
-        this._put(cid, node, callback)
+          if (options.onlyHash) {
+            return resolve({ done: false, value: cid })
+          }
+
+          this._put(cid, node, (err, cid) => {
+            if (err) {
+              return reject(err)
+            }
+            return resolve({ done: false, value: cid })
+          })
+        })
       })
-    }).catch((err) => {
-      callback(err)
-    })
+    }
+
+    return fancyIterator(next)
   }
 
   treeStream (cid, path, options) {

test/basics.js

@@ -10,6 +10,7 @@ chai.use(chaiAsProised)
 const BlockService = require('ipfs-block-service')
 const CID = require('cids')
 const multihash = require('multihashes')
+const multicodec = require('multicodec')
 const pull = require('pull-stream')
 const inMemory = require('ipld-in-memory')
 
@@ -62,25 +63,19 @@ module.exports = (repo) => {
     //   })
     // }
 
-    it('put - errors on unknown resolver', (done) => {
+    it('put - errors on unknown resolver', async () => {
       const bs = new BlockService(repo)
       const r = new IPLDResolver({ blockService: bs })
       // choosing a format that is not supported
-      r.put(null, { format: 'blake2b-8' }, (err, result) => {
-        expect(err).to.exist()
-        expect(err.message).to.eql('No resolver found for codec "blake2b-8"')
-        done()
-      })
+      const result = r.put([null], multicodec.BLAKE2B_8)
+      await expect(result.next()).to.be.rejectedWith(
+        'No resolver found for codec "blake2b-8"')
     })
 
-    it('put - errors if no options', (done) => {
+    it('put - errors if no format is provided', () => {
       const bs = new BlockService(repo)
       const r = new IPLDResolver({ blockService: bs })
-      r.put(null, (err, result) => {
-        expect(err).to.exist()
-        expect(err.message).to.eql('IPLDResolver.put requires options')
-        done()
-      })
+      expect(() => r.put([null])).to.be.throw('`put` requires a format')
     })
 
     it('_put - errors on unknown resolver', (done) => {

test/format-support.js

@@ -17,17 +17,14 @@ module.exports = (repo) => {
   describe('IPLD format support', () => {
     let data, cid
 
-    before((done) => {
+    before(async () => {
       const bs = new BlockService(repo)
       const resolver = new IPLDResolver({ blockService: bs })
 
       data = { now: Date.now() }
 
-      dagCBOR.util.cid(data, (err, c) => {
-        expect(err).to.not.exist()
-        cid = c
-        resolver.put(data, { cid }, done)
-      })
+      const result = resolver.put([data], multicodec.DAG_CBOR)
+      cid = await result.last()
     })
 
     describe('Dynamic format loading', () => {

test/ipld-all.js

@@ -17,6 +17,7 @@ const each = require('async/each')
 const waterfall = require('async/waterfall')
 const CID = require('cids')
 const inMemory = require('ipld-in-memory')
+const multicodec = require('multicodec')
 
 const IPLDResolver = require('../src')
 
@@ -52,10 +53,15 @@ describe('IPLD Resolver for dag-cbor + dag-pb', () => {
         cidCbor = cid
 
         each([
-          { node: nodePb, cid: cidPb },
-          { node: nodeCbor, cid: cidCbor }
+          { node: nodePb, format: multicodec.DAG_PB, cidVersion: 0 },
+          { node: nodeCbor, format: multicodec.DAG_CBOR, cidVersion: 1 }
         ], (nac, cb) => {
-          resolver.put(nac.node, { cid: nac.cid }, cb)
+          resolver.put([nac.node], nac.format, {
+            cidVersion: nac.cidVersion
+          }).first().then(
+            () => cb(null),
+            (error) => cb(error)
+          )
         }, cb)
       }
     ], done)
@@ -76,12 +82,17 @@ describe('IPLD Resolver for dag-cbor + dag-pb', () => {
   it('does not store nodes when onlyHash is passed', (done) => {
     waterfall([
       (cb) => dagPB.DAGNode.create(Buffer.from('Some data here'), cb),
-      (node, cb) => resolver.put(node, {
-        onlyHash: true,
-        version: 1,
-        hashAlg: 'sha2-256',
-        format: 'dag-pb'
-      }, cb),
+      (node, cb) => {
+        const result = resolver.put([node], multicodec.DAG_PB, {
+          onlyHash: true,
+          cidVersion: 1,
+          hashAlg: multicodec.SHA2_256
+        })
+        result.first().then(
+          (cid) => cb(null, cid),
+          (error) => cb(error)
+        )
+      },
       (cid, cb) => resolver.bs._repo.blocks.has(cid, cb)
     ], (error, result) => {
       if (error) {
@@ -93,30 +104,6 @@ describe('IPLD Resolver for dag-cbor + dag-pb', () => {
     })
   })
 
-  it('does not store nodes when onlyHash is passed and a CID is passed', (done) => {
-    const cid = new CID('QmTmxQfEHbQzntsXPTU4ae2ZgBGwseBmS12AkZnKCkuf2G')
-
-    waterfall([
-      (cb) => dagPB.DAGNode.create(Buffer.from('Some data here'), cb),
-      (node, cb) => resolver.put(node, {
-        onlyHash: true,
-        cid
-      }, cb),
-      (cid2, cb) => {
-        expect(cid2).to.equal(cid)
-
-        resolver.bs._repo.blocks.has(cid2, cb)
-      }
-    ], (error, result) => {
-      if (error) {
-        return done(error)
-      }
-
-      expect(result).to.be.false()
-      done()
-    })
-  })
-
   describe('getMany', () => {
     it('should return nodes correctly', (done) => {
       resolver.getMany([cidCbor, cidPb], (err, result) => {

test/ipld-bitcoin.js

@@ -11,7 +11,7 @@ const BitcoinBlock = require('bitcoinjs-lib').Block
 const multihash = require('multihashes')
 const series = require('async/series')
 const each = require('async/each')
-const pull = require('pull-stream')
+const multicodec = require('multicodec')
 
 const IPLDResolver = require('../src')
 
@@ -83,16 +83,12 @@ module.exports = (repo) => {
         }
       ], store)
 
-      function store () {
-        pull(
-          pull.values([
-            { node: node1, cid: cid1 },
-            { node: node2, cid: cid2 },
-            { node: node3, cid: cid3 }
-          ]),
-          pull.asyncMap((nac, cb) => resolver.put(nac.node, { cid: nac.cid }, cb)),
-          pull.onEnd(done)
-        )
+      async function store () {
+        const nodes = [node1, node2, node3]
+        const result = resolver.put(nodes, multicodec.BITCOIN_BLOCK)
+        ;[cid1, cid2, cid3] = await result.all()
+
+        done()
       }
     })
 
@@ -121,34 +117,26 @@ module.exports = (repo) => {
     })
 
     describe('public api', () => {
-      it('resolver.put', (done) => {
-        resolver.put(node1, { cid: cid1 }, done)
-      })
-
-      it('resolver.put with format', (done) => {
-        resolver.put(node1, { format: 'bitcoin-block' }, (err, cid) => {
-          expect(err).to.not.exist()
-          expect(cid).to.exist()
-          expect(cid.version).to.equal(1)
-          expect(cid.codec).to.equal('bitcoin-block')
-          expect(cid.multihash).to.exist()
-          const mh = multihash.decode(cid.multihash)
-          expect(mh.name).to.equal('dbl-sha2-256')
-          done()
-        })
+      it('resolver.put with format', async () => {
+        const result = resolver.put([node1], multicodec.BITCOIN_BLOCK)
+        const cid = await result.first()
+        expect(cid.version).to.equal(1)
+        expect(cid.codec).to.equal('bitcoin-block')
+        expect(cid.multihash).to.exist()
+        const mh = multihash.decode(cid.multihash)
+        expect(mh.name).to.equal('dbl-sha2-256')
       })
 
-      it('resolver.put with format + hashAlg', (done) => {
-        resolver.put(node1, { format: 'bitcoin-block', hashAlg: 'sha3-512' }, (err, cid) => {
-          expect(err).to.not.exist()
-          expect(cid).to.exist()
-          expect(cid.version).to.equal(1)
-          expect(cid.codec).to.equal('bitcoin-block')
-          expect(cid.multihash).to.exist()
-          const mh = multihash.decode(cid.multihash)
-          expect(mh.name).to.equal('sha3-512')
-          done()
+      it('resolver.put with format + hashAlg', async () => {
+        const result = resolver.put([node1], multicodec.BITCOIN_BLOCK, {
+          hashAlg: multicodec.SHA3_512
         })
+        const cid = await result.first()
+        expect(cid.version).to.equal(1)
+        expect(cid.codec).to.equal('bitcoin-block')
+        expect(cid.multihash).to.exist()
+        const mh = multihash.decode(cid.multihash)
+        expect(mh.name).to.equal('sha3-512')
       })
 
       // TODO vmx 2018-11-30: Implement getting the whole object properly