more fixes for merkledag changes

Author: dignifiedquire

package.json

@@ -2,7 +2,7 @@
   "name": "ipfs-unixfs-engine",
   "version": "0.11.4",
   "description": "JavaScript implementation of the unixfs Engine used by IPFS",
-  "main": "lib/index.js",
+  "main": "src/index.js",
   "jsnext:main": "src/index.js",
   "scripts": {
     "lint": "aegir-lint",
@@ -45,10 +45,10 @@
     "pre-commit": "^1.1.3",
     "pull-zip": "^2.0.0",
     "raw-loader": "^0.5.1",
-    "rimraf": "^2.5.4",
-    "run-series": "^1.1.4"
+    "rimraf": "^2.5.4"
   },
   "dependencies": {
+    "async": "^2.0.1",
     "ipfs-merkle-dag": "^0.7.0",
     "ipfs-unixfs": "^0.1.4",
     "is-ipfs": "^0.2.0",
@@ -58,8 +58,7 @@
     "pull-pushable": "^2.0.1",
     "pull-stream": "^3.4.5",
     "pull-traverse": "^1.0.3",
-    "pull-write": "^1.1.0",
-    "run-parallel": "^1.1.6"
+    "pull-write": "^1.1.0"
   },
   "contributors": [
     "David Dias <daviddias.p@gmail.com>",

src/exporter.js

@@ -11,7 +11,11 @@ const dirExporter = require('./exporters/dir')
 const fileExporter = require('./exporters/file')
 
 module.exports = (hash, dagService, options) => {
-  hash = cleanMultihash(hash)
+  try {
+    hash = cleanMultihash(hash)
+  } catch (err) {
+    return pull.error(err)
+  }
   options = options || {}
 
   function visitor (item) {

src/importer.js

@@ -6,10 +6,12 @@ const assert = require('assert')
 const pull = require('pull-stream')
 const pushable = require('pull-pushable')
 const write = require('pull-write')
-const parallel = require('run-parallel')
+const parallel = require('async/parallel')
+const waterfall = require('async/waterfall')
 
 const fsc = require('./chunker-fixed-size')
 const createAndStoreTree = require('./tree')
+const getSizeAndHash = require('./util').getSizeAndHash
 
 const DAGNode = merkleDAG.DAGNode
 
@@ -67,15 +69,18 @@ function createAndStoreDir (item, ds, cb) {
   const n = new DAGNode()
   n.data = d.marshal()
 
-  ds.put(n, (err) => {
-    if (err) return cb(err)
-    cb(null, {
-      path: item.path,
-      multihash: n.multihash(),
-      size: n.size()
-      // dataSize: d.fileSize()
-    })
-  })
+  waterfall([
+    (cb) => ds.put(n, cb),
+    (cb) => getSizeAndHash(n, cb),
+    (res, cb) => {
+      cb(null, {
+        path: item.path,
+        multihash: res.multihash,
+        size: res.size
+        // dataSize: d.fileSize()
+      })
+    }
+  ], cb)
 }
 
 function createAndStoreFile (file, ds, cb) {
@@ -102,14 +107,17 @@ function createAndStoreFile (file, ds, cb) {
       const l = new UnixFS('file', Buffer(chunk))
       const n = new DAGNode(l.marshal())
 
-      ds.put(n, (err) => {
+      waterfall([
+        (cb) => ds.put(n, cb),
+        (cb) => getSizeAndHash(n, cb)
+      ], (err, stats) => {
         if (err) {
-          return cb(new Error('Failed to store chunk'))
+          return cb(err)
         }
 
         cb(null, {
-          Hash: n.multihash(),
-          Size: n.size(),
+          Hash: stats.multihash,
+          Size: stats.size,
           leafSize: l.fileSize(),
           Name: ''
         })
@@ -140,13 +148,18 @@ function createAndStoreFile (file, ds, cb) {
       }
 
       n.data = f.marshal()
-      ds.put(n, (err) => {
-        if (err) return cb(err)
+      parallel([
+        (cb) => ds.put(n, cb),
+        (cb) => getSizeAndHash(n, cb)
+      ], (err, res) => {
+        if (err) {
+          return cb(err)
+        }
 
         cb(null, {
           path: file.path,
-          multihash: n.multihash(),
-          size: n.size()
+          multihash: res[1].multihash,
+          size: res[1].size
           // dataSize: f.fileSize()
         })
       })

src/tree.js

@@ -3,6 +3,9 @@
 const mh = require('multihashes')
 const UnixFS = require('ipfs-unixfs')
 const merkleDAG = require('ipfs-merkle-dag')
+const mapValues = require('async/mapValues')
+
+const getSizeAndHash = require('./util').getSizeAndHash
 
 const DAGLink = merkleDAG.DAGLink
 const DAGNode = merkleDAG.DAGNode
@@ -75,55 +78,75 @@ module.exports = (files, dagService, source, cb) => {
 
   function traverse (tree, path, done) {
     const keys = Object.keys(tree)
-    let tmpTree = tree
-    keys.map((key) => {
-      if (typeof tmpTree[key] === 'object' &&
-          !Buffer.isBuffer(tmpTree[key])) {
-        tmpTree[key] = traverse.call(this, tmpTree[key], path ? path + '/' + key : key, done)
-      }
-    })
-
-    // at this stage, all keys are multihashes
-    // create a dir node
-    // add all the multihashes as links
-    // return this new node multihash
-
-    const d = new UnixFS('directory')
-    const n = new DAGNode()
-
-    keys.forEach((key) => {
-      const b58mh = mh.toB58String(tmpTree[key])
-      const l = new DAGLink(
-        key, mhIndex[b58mh].size, tmpTree[key])
-      n.addRawLink(l)
-    })
-
-    n.data = d.marshal()
 
-    pendingWrites++
-    dagService.put(n, (err) => {
-      pendingWrites--
+    mapValues(tree, (node, key, cb) => {
+      if (typeof node === 'object' && !Buffer.isBuffer(node)) {
+        traverse.call(this, node, path ? `${path}/${key}` : key, cb)
+      } else {
+        cb(null, node)
+      }
+    }, (err, tmpTree) => {
       if (err) {
-        source.push(new Error('failed to store dirNode'))
-      } else if (path) {
-        source.push({
-          path: path,
-          multihash: n.multihash(),
-          size: n.size()
-        })
+        return done(err)
       }
 
-      if (pendingWrites <= 0) {
-        done()
+      // at this stage, all keys are multihashes
+      // create a dir node
+      // add all the multihashes as links
+      // return this new node multihash
+
+      const d = new UnixFS('directory')
+      const n = new DAGNode()
+
+      keys.forEach((key) => {
+        const b58mh = mh.toB58String(tmpTree[key])
+        const l = new DAGLink(
+          key, mhIndex[b58mh].size, tmpTree[key])
+        n.addRawLink(l)
+      })
+
+      n.data = d.marshal()
+
+      pendingWrites++
+      dagService.put(n, (err) => {
+        pendingWrites--
+        if (err) {
+          source.push(new Error('failed to store dirNode'))
+        } else if (path) {
+          getSizeAndHash(n, (err, stats) => {
+            if (err) {
+              return source.push(err)
+            }
+            source.push({
+              path: path,
+              multihash: stats.multihash,
+              size: stats.size
+            })
+          })
+        }
+
+        if (pendingWrites <= 0) {
+          finish()
+        }
+      })
+
+      function finish () {
+        if (!path) {
+          return done()
+        }
+
+        getSizeAndHash(n, (err, stats) => {
+          if (err) {
+            return done(err)
+          }
+
+          mhIndex[mh.toB58String(stats.multihash)] = {
+            size: stats.size
+          }
+          done(null, stats.multihash)
+        })
       }
     })
-
-    if (!path) {
-      return
-    }
-
-    mhIndex[mh.toB58String(n.multihash())] = { size: n.size() }
-    return n.multihash()
   }
 
   traverse(fileTree, null, cb)

src/util.js

@@ -4,6 +4,7 @@ const UnixFS = require('ipfs-unixfs')
 const pull = require('pull-stream')
 const mh = require('multihashes')
 const isIPFS = require('is-ipfs')
+const parallel = require('async/parallel')
 
 exports.switchType = (node, dirHandler, fileHandler) => {
   const data = UnixFS.unmarshal(node.data)
@@ -26,3 +27,10 @@ exports.cleanMultihash = (multihash) => {
 
   return multihash
 }
+
+exports.getSizeAndHash = (n, cb) => {
+  parallel({
+    multihash: (cb) => n.multihash(cb),
+    size: (cb) => n.size(cb)
+  }, cb)
+}

test/node.js

@@ -7,7 +7,7 @@ const path = require('path')
 const IPFSRepo = require('ipfs-repo')
 const Store = require('fs-pull-blob-store')
 const mkdirp = require('mkdirp')
-const series = require('run-series')
+const series = require('async/series')
 
 describe('core', () => {
   const repoExample = path.join(process.cwd(), '/test/repo-example')