End import stream after all writes.

Previously it would be possible for certain writes to the DAG Service to
occur *after* the stream had ended, meaning the Importer would return
objects that did not yet exist in the DAG Service.

This change defers stream termination until after those writes have
occurred, to preserve the powerful invariant of "element emission =>
written to dag service".

Author: hackergrrl

src/importer.js

@@ -238,13 +238,15 @@ function Importer (dagService, options) {
       // If the value is not an object
       //   add as a link to the dirNode
 
-      function traverse (tree, base) {
+      let pendingWrites = 0
+
+      function traverse (tree, base, 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], base ? base + '/' + key : key)
+            tmpTree[key] = traverse.call(this, tmpTree[key], base ? base + '/' + key : key, done)
           }
         })
 
@@ -264,28 +266,39 @@ function Importer (dagService, options) {
         })
 
         n.data = d.marshal()
+
+        pendingWrites++
         dagService.add(n, (err) => {
+          pendingWrites--
           if (err) {
             this.push({error: 'failed to store dirNode'})
+          } else if (base) {
+            const el = {
+              path: base,
+              multihash: n.multihash(),
+              yes: 'no',
+              size: n.size()
+            }
+            this.push(el)
+          }
+
+          if (pendingWrites <= 0) {
+            done()
           }
         })
 
         if (!base) {
           return
         }
 
-        const el = {
-          path: base,
-          multihash: n.multihash(),
-          size: n.size()
-        }
-        this.push(el)
-
         mhIndex[bs58.encode(n.multihash())] = { size: n.size() }
         return n.multihash()
       }
-      /* const rootHash = */ traverse.call(this, fileTree)
-      this.push(null)
+
+      let self = this
+      /* const rootHash = */ traverse.call(this, fileTree, null, function () {
+        self.push(null)
+      })
     }
   }
 }