feat: switch to async await

package.json

@@ -18,7 +18,8 @@
     "release": "aegir release",
     "release-minor": "aegir release --type minor",
     "release-major": "aegir release --type major",
-    "coverage": "aegir coverage"
+    "coverage": "aegir coverage",
+    "dep-check": "aegir dep-check"
   },
   "repository": {
     "type": "git",
@@ -38,35 +39,41 @@
   "homepage": "https://github.com/ipfs/js-ipfs-unixfs-importer#readme",
   "devDependencies": {
     "aegir": "^18.0.2",
+    "async-iterator-all": "0.0.2",
+    "async-iterator-buffer-stream": "~0.0.1",
+    "async-iterator-first": "0.0.2",
+    "async-iterator-last": "0.0.2",
     "chai": "^4.2.0",
     "cids": "~0.5.5",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
     "ipfs-unixfs-exporter": "~0.36.1",
-    "ipld": "~0.21.1",
+    "ipld": "~0.22.0",
     "ipld-in-memory": "^2.0.0",
     "multihashes": "~0.4.14",
-    "pull-buffer-stream": "^1.0.1",
     "pull-generate": "^2.2.0",
     "pull-traverse": "^1.0.3",
     "sinon": "^7.1.0"
   },
   "dependencies": {
     "async": "^2.6.1",
-    "async-iterator-to-pull-stream": "^1.1.0",
+    "async-iterator-to-pull-stream": "^1.3.0",
     "bl": "^3.0.0",
     "deep-extend": "~0.6.0",
+    "err-code": "^1.1.2",
     "hamt-sharding": "~0.0.2",
     "ipfs-unixfs": "~0.1.16",
     "ipld-dag-pb": "~0.15.2",
     "left-pad": "^1.3.0",
+    "multicodec": "~0.5.1",
     "multihashing-async": "~0.5.1",
     "pull-batch": "^1.0.0",
     "pull-pair": "^1.1.0",
     "pull-paramap": "^1.2.2",
     "pull-pause": "0.0.2",
     "pull-pushable": "^2.2.0",
     "pull-stream": "^3.6.9",
+    "pull-stream-to-async-iterator": "^1.0.1",
     "pull-through": "^1.0.18",
     "pull-write": "^1.1.4",
     "stream-to-pull-stream": "^1.7.2"

src/builder/builder.js

@@ -15,6 +15,7 @@ const reduce = require('./reduce')
 const {
   DAGNode
 } = require('ipld-dag-pb')
+const errCode = require('err-code')
 
 const defaultOptions = {
   chunkerOptions: {
@@ -78,7 +79,7 @@ module.exports = function builder (createChunker, ipld, createReducer, _options)
 
       callback(null, {
         path: item.path,
-        multihash: result.cid.buffer,
+        cid: result.cid,
         size: result.node.size
       })
     })
@@ -90,7 +91,7 @@ module.exports = function builder (createChunker, ipld, createReducer, _options)
     }
 
     if (typeof file.content !== 'function') {
-      return callback(new Error('invalid content'))
+      return callback(errCode(new Error('invalid content'), 'EINVALIDCONTENT'))
     }
 
     const reducer = createReducer(reduce(file, ipld, options), options)
@@ -146,7 +147,7 @@ module.exports = function builder (createChunker, ipld, createReducer, _options)
                 size: leaf.size,
                 leafSize: leaf.leafSize,
                 data: results.node,
-                multihash: results.cid.buffer,
+                cid: results.cid,
                 path: leaf.path,
                 name: ''
               })

src/builder/reduce.js

@@ -16,7 +16,7 @@ module.exports = function reduce (file, ipld, options) {
       return callback(null, {
         size: leaf.size,
         leafSize: leaf.leafSize,
-        multihash: leaf.multihash,
+        cid: leaf.cid,
         path: file.path,
         name: leaf.name
       })
@@ -28,7 +28,7 @@ module.exports = function reduce (file, ipld, options) {
     const links = leaves.map((leaf) => {
       f.addBlockSize(leaf.leafSize)
 
-      return new DAGLink(leaf.name, leaf.size, leaf.multihash)
+      return new DAGLink(leaf.name, leaf.size, leaf.cid)
     })
 
     waterfall([
@@ -42,7 +42,7 @@ module.exports = function reduce (file, ipld, options) {
       callback(null, {
         size: result.node.size,
         leafSize: f.fileSize(),
-        multihash: result.cid.buffer,
+        cid: result.cid,
         path: file.path,
         name: ''
       })

src/importer/dir-flat.js

@@ -16,7 +16,7 @@ class DirFlat extends Dir {
   }
 
   put (name, value, callback) {
-    this.multihash = undefined
+    this.cid = undefined
     this.size = undefined
     this._children[name] = value
     process.nextTick(callback)
@@ -52,7 +52,7 @@ class DirFlat extends Dir {
     const links = Object.keys(this._children)
       .map((key) => {
         const child = this._children[key]
-        return new DAGLink(key, child.size, child.multihash)
+        return new DAGLink(key, child.size, child.cid)
       })
 
     const dir = new UnixFS('directory')
@@ -62,12 +62,12 @@ class DirFlat extends Dir {
         (callback) => DAGNode.create(dir.marshal(), links, callback),
         (node, callback) => persist(node, ipld, this._options, callback),
         ({ cid, node }, callback) => {
-          this.multihash = cid.buffer
+          this.cid = cid
           this.size = node.size
           const pushable = {
             path: path,
-            multihash: cid.buffer,
-            size: node.size
+            size: node.size,
+            cid: cid
           }
           source.push(pushable)
           callback(null, node)

src/importer/dir-sharded.js

@@ -52,6 +52,10 @@ class DirSharded extends Dir {
   }
 
   async put (name, value, callback) {
+    if (!callback) {
+      console.info('wut')
+    }
+
     try {
       await this._bucket.put(name, value)
 
@@ -100,7 +104,7 @@ class DirSharded extends Dir {
       if (err) {
         return callback(err)
       } else {
-        this.multihash = results.cid.buffer
+        this.cid = results.cid
         this.size = results.node.size
       }
 
@@ -157,7 +161,7 @@ function flush (options, bucket, path, ipld, source, callback) {
     } else {
       const value = child.value
       const label = labelPrefix + child.key
-      links.push(new DAGLink(label, value.size, value.multihash))
+      links.push(new DAGLink(label, value.size, value.cid))
       callback()
     }
   }
@@ -176,7 +180,7 @@ function flush (options, bucket, path, ipld, source, callback) {
         const pushable = {
           path: path,
           size: node.size,
-          multihash: cid.buffer
+          cid: cid
         }
         if (source) {
           source.push(pushable)

src/importer/index.js

@@ -108,6 +108,7 @@ module.exports = function (ipld, _options) {
           callback(err)
           return
         }
+
         pausable.resume()
         callback(null, hash)
       })

src/importer/tree-builder.js

@@ -10,6 +10,7 @@ const DirFlat = require('./dir-flat')
 const flatToShard = require('./flat-to-shard')
 const Dir = require('./dir')
 const toPathComponents = require('../utils/to-path-components')
+const errCode = require('err-code')
 
 module.exports = createTreeBuilder
 
@@ -104,7 +105,7 @@ function createTreeBuilder (ipld, _options) {
       currentPath += pathElem
       const last = (index === lastIndex)
       parent.dirty = true
-      parent.multihash = null
+      parent.cid = null
       parent.size = null
 
       if (last) {
@@ -151,7 +152,7 @@ function createTreeBuilder (ipld, _options) {
         if (err) {
           callback(err)
         } else {
-          callback(null, node && node.multihash)
+          callback(null, node && node.cid)
         }
       }
     })
@@ -160,7 +161,7 @@ function createTreeBuilder (ipld, _options) {
   function flush (path, tree, callback) {
     if (tree.dir) {
       if (tree.root && tree.childCount() > 1 && !options.wrap) {
-        callback(new Error('detected more than one root'))
+        callback(errCode(new Error('detected more than one root'), 'EMORETHANONEROOT'))
         return // early
       }
       tree.eachChildSeries(
@@ -196,7 +197,7 @@ function createTreeBuilder (ipld, _options) {
     }
 
     if (!tree.dirty) {
-      callback(null, tree.multihash)
+      callback(null, tree.cid)
       return // early
     }
 

src/index.js

@@ -1,3 +1,26 @@
 'use strict'
 
-module.exports = require('./importer')
+const pull = require('pull-stream/pull')
+const map = require('pull-stream/throughs/map')
+const toPull = require('async-iterator-to-pull-stream')
+const toIterator = require('pull-stream-to-async-iterator')
+const importer = require('./importer')
+
+module.exports = function (source, ipld, options = {}) {
+  return toIterator(
+    pull(
+      toPull.source(source),
+      map(({ path, content }) => {
+        if (content && content[Symbol.asyncIterator]) {
+          content = toPull(content)
+        }
+
+        return {
+          path,
+          content
+        }
+      }),
+      importer(ipld, options)
+    )
+  )
+}

src/utils/persist.js

@@ -1,5 +1,8 @@
 'use strict'
 
+const mh = require('multihashes')
+const mc = require('multicodec')
+
 const {
   util: {
     cid
@@ -22,10 +25,14 @@ const persist = (node, ipld, options, callback) => {
     codec = 'raw'
   }
 
-  if (hashAlg !== 'sha2-256') {
+  if (hashAlg !== 'sha2-256' && hashAlg !== mh.names['sha2-256']) {
     cidVersion = 1
   }
 
+  if (isNaN(hashAlg)) {
+    hashAlg = mh.names[hashAlg]
+  }
+
   if (options.onlyHash) {
     return cid(node, {
       version: cidVersion,
@@ -38,16 +45,14 @@ const persist = (node, ipld, options, callback) => {
     })
   }
 
-  ipld.put(node, {
-    version: cidVersion,
-    hashAlg: hashAlg,
-    format: codec
-  }, (error, cid) => {
-    callback(error, {
+  ipld.put(node, mc[codec.toUpperCase().replace(/-/g, '_')], {
+    cidVersion: cidVersion,
+    hashAlg: hashAlg
+  })
+    .then((cid) => callback(null, {
       cid,
       node
-    })
-  })
+    }), callback)
 }
 
 module.exports = persist

test/benchmark.spec.js

@@ -6,12 +6,9 @@ const importer = require('../src')
 const chai = require('chai')
 chai.use(require('dirty-chai'))
 const expect = chai.expect
-const pull = require('pull-stream/pull')
-const values = require('pull-stream/sources/values')
-const onEnd = require('pull-stream/sinks/on-end')
 const IPLD = require('ipld')
 const inMemory = require('ipld-in-memory')
-const bufferStream = require('pull-buffer-stream')
+const bufferStream = require('async-iterator-buffer-stream')
 
 const REPEATS = 10
 const FILE_SIZE = Math.pow(2, 20) * 500 // 500MB
@@ -35,14 +32,14 @@ describe.skip('benchmark', function () {
   const times = []
 
   after(() => {
-    console.info(`Percent\tms`)
+    console.info(`Percent\tms`) // eslint-disable-line no-console
     times.forEach((time, index) => {
-      console.info(`${index}\t${parseInt(time / REPEATS)}`)
+      console.info(`${index}\t${parseInt(time / REPEATS)}`) // eslint-disable-line no-console
     })
   })
 
   for (let i = 0; i < REPEATS; i++) {
-    it(`run ${i}`, (done) => { // eslint-disable-line no-loop-func
+    it(`run ${i}`, async () => { // eslint-disable-line no-loop-func
       this.timeout(0)
 
       const size = FILE_SIZE
@@ -67,22 +64,17 @@ describe.skip('benchmark', function () {
 
       const buf = Buffer.alloc(CHUNK_SIZE).fill(0)
 
-      pull(
-        values([{
-          path: '200Bytes.txt',
-          content: bufferStream(size, {
-            chunkSize: CHUNK_SIZE,
-            generator: (num, cb) => {
-              cb(null, buf)
-            }
-          })
-        }]),
-        importer(ipld, options),
-        onEnd((err) => {
-          expect(err).to.not.exist()
-          done()
+      for await (const file of importer({ // eslint-disable-line no-unused-vars
+        path: '200Bytes.txt',
+        content: bufferStream(size, {
+          chunkSize: CHUNK_SIZE,
+          generator: () => {
+            return buf
+          }
         })
-      )
+      }, ipld, options)) {
+        // do nothing
+      }
     })
   }
 })