feat: support adding async iterators

.aegir.js

@@ -3,9 +3,11 @@
 const IPFSFactory = require('ipfsd-ctl')
 const parallel = require('async/parallel')
 const MockPreloadNode = require('./test/utils/mock-preload-node')
+const EchoServer = require('interface-ipfs-core/src/utils/echo-http-server')
 
 const ipfsdServer = IPFSFactory.createServer()
 const preloadNode = MockPreloadNode.createNode()
+const echoServer = EchoServer.createServer()
 
 module.exports = {
   bundlesize: { maxSize: '689kB' },
@@ -26,8 +28,18 @@ module.exports = {
   },
   hooks: {
     node: {
-      pre: (cb) => preloadNode.start(cb),
-      post: (cb) => preloadNode.stop(cb)
+      pre: (cb) => {
+        parallel([
+          (cb) => preloadNode.start(cb),
+          (cb) => echoServer.start(cb)
+        ], cb)
+      },
+      post: (cb) => {
+        parallel([
+          (cb) => preloadNode.stop(cb),
+          (cb) => echoServer.stop(cb)
+        ], cb)
+      }
     },
     browser: {
       pre: (cb) => {
@@ -36,7 +48,8 @@ module.exports = {
             ipfsdServer.start()
             cb()
           },
-          (cb) => preloadNode.start(cb)
+          (cb) => preloadNode.start(cb),
+          (cb) => echoServer.start(cb)
         ], cb)
       },
       post: (cb) => {
@@ -45,7 +58,8 @@ module.exports = {
             ipfsdServer.stop()
             cb()
           },
-          (cb) => preloadNode.stop(cb)
+          (cb) => preloadNode.stop(cb),
+          (cb) => echoServer.stop(cb)
         ], cb)
       }
     }

package.json

@@ -96,10 +96,10 @@
     "ipfs-bitswap": "~0.25.1",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.15.2",
-    "ipfs-http-client": "^34.0.0",
+    "ipfs-http-client": "^35.1.0",
     "ipfs-http-response": "~0.3.1",
-    "ipfs-mfs": "~0.12.0",
-    "ipfs-multipart": "~0.1.1",
+    "ipfs-mfs": "^0.12.2",
+    "ipfs-multipart": "^0.2.0",
     "ipfs-repo": "~0.26.6",
     "ipfs-unixfs": "~0.1.16",
     "ipfs-unixfs-exporter": "~0.37.7",
@@ -119,6 +119,8 @@
     "is-pull-stream": "~0.0.0",
     "is-stream": "^2.0.0",
     "iso-url": "~0.4.6",
+    "it-pipe": "^1.0.1",
+    "it-to-stream": "^0.1.1",
     "just-safe-set": "^2.1.0",
     "kind-of": "^6.0.2",
     "libp2p": "~0.26.1",
@@ -142,7 +144,7 @@
     "merge-options": "^1.0.1",
     "mime-types": "^2.1.21",
     "mkdirp": "~0.5.1",
-    "mortice": "^1.2.2",
+    "mortice": "^2.0.0",
     "multiaddr": "^6.1.0",
     "multiaddr-to-uri": "^5.0.0",
     "multibase": "~0.6.0",
@@ -194,7 +196,7 @@
     "execa": "^2.0.4",
     "form-data": "^2.5.1",
     "hat": "0.0.3",
-    "interface-ipfs-core": "^0.111.1",
+    "interface-ipfs-core": "^0.113.0",
     "ipfsd-ctl": "~0.46.0",
     "libp2p-websocket-star": "~0.10.2",
     "ncp": "^2.0.0",

src/cli/commands/add.js

@@ -1,65 +1,19 @@
 'use strict'
 
-const pull = require('pull-stream/pull')
-const through = require('pull-stream/throughs/through')
-const end = require('pull-stream/sinks/on-end')
 const promisify = require('promisify-es6')
 const getFolderSize = promisify(require('get-folder-size'))
 const byteman = require('byteman')
 const mh = require('multihashes')
 const multibase = require('multibase')
-const toPull = require('stream-to-pull-stream')
 const { createProgressBar } = require('../utils')
 const { cidToString } = require('../../utils/cid')
-const globSource = require('../../utils/files/glob-source')
+const globSource = require('ipfs-utils/src/files/glob-source')
 
 async function getTotalBytes (paths) {
   const sizes = await Promise.all(paths.map(p => getFolderSize(p)))
   return sizes.reduce((total, size) => total + size, 0)
 }
 
-function addPipeline (source, addStream, options, log) {
-  let finalHash
-
-  return new Promise((resolve, reject) => {
-    pull(
-      source,
-      addStream,
-      through((file) => {
-        const cid = finalHash = cidToString(file.hash, { base: options.cidBase })
-
-        if (options.silent || options.quieter) {
-          return
-        }
-
-        let message = cid
-
-        if (!options.quiet) {
-          // print the hash twice if we are piping from stdin
-          message = `added ${cid} ${options.file ? file.path || '' : cid}`.trim()
-        }
-
-        log(message)
-      }),
-      end((err) => {
-        if (err) {
-          // Tweak the error message and add more relevant infor for the CLI
-          if (err.code === 'ERR_DIR_NON_RECURSIVE') {
-            err.message = `'${err.path}' is a directory, use the '-r' flag to specify directories`
-          }
-          return reject(err)
-        }
-
-        if (options.quieter) {
-          log(finalHash)
-        }
-
-        resolve()
-      })
-    )
-  })
-}
-
 module.exports = {
   command: 'add [file...]',
 
@@ -199,17 +153,51 @@ module.exports = {
       }
 
       const source = argv.file
-        ? globSource(...argv.file, { recursive: argv.recursive })
-        : toPull.source(process.stdin) // Pipe directly to ipfs.add
+        ? globSource(argv.file, { recursive: argv.recursive })
+        : process.stdin // Pipe directly to ipfs.add
 
-      const adder = ipfs.addPullStream(options)
+      let finalHash
 
       try {
-        await addPipeline(source, adder, argv, log)
-      } finally {
+        for await (const file of ipfs._addAsyncIterator(source, options)) {
+          if (argv.silent) {
+            continue
+          }
+
+          if (argv.quieter) {
+            finalHash = file.hash
+            continue
+          }
+
+          const cid = cidToString(file.hash, { base: argv.cidBase })
+          let message = cid
+
+          if (!argv.quiet) {
+            // print the hash twice if we are piping from stdin
+            message = `added ${cid} ${argv.file ? file.path || '' : cid}`.trim()
+          }
+
+          log(message)
+        }
+      } catch (err) {
         if (bar) {
           bar.terminate()
         }
+
+        // Tweak the error message and add more relevant infor for the CLI
+        if (err.code === 'ERR_DIR_NON_RECURSIVE') {
+          err.message = `'${err.path}' is a directory, use the '-r' flag to specify directories`
+        }
+
+        throw err
+      }
+
+      if (bar) {
+        bar.terminate()
+      }
+
+      if (argv.quieter) {
+        log(cidToString(finalHash, { base: argv.cidBase }))
       }
     })())
   }

src/core/components/files-regular/add-async-iterator.js

@@ -0,0 +1,148 @@
+'use strict'
+
+const importer = require('ipfs-unixfs-importer')
+const normaliseAddInput = require('ipfs-utils/src/files/normalise-input')
+const { parseChunkerString } = require('./utils')
+const pipe = require('it-pipe')
+const log = require('debug')('ipfs:add')
+log.error = require('debug')('ipfs:add:error')
+
+function noop () {}
+
+module.exports = function (self) {
+  // Internal add func that gets used by all add funcs
+  return async function * addAsyncIterator (source, options) {
+    options = options || {}
+
+    const chunkerOptions = parseChunkerString(options.chunker)
+
+    const opts = Object.assign({}, {
+      shardSplitThreshold: self._options.EXPERIMENTAL.sharding
+        ? 1000
+        : Infinity
+    }, options, {
+      chunker: chunkerOptions.chunker,
+      chunkerOptions: chunkerOptions.chunkerOptions
+    })
+
+    // CID v0 is for multihashes encoded with sha2-256
+    if (opts.hashAlg && opts.cidVersion !== 1) {
+      opts.cidVersion = 1
+    }
+
+    let total = 0
+
+    const prog = opts.progress || noop
+    const progress = (bytes) => {
+      total += bytes
+      prog(total)
+    }
+
+    opts.progress = progress
+
+    const iterator = pipe(
+      normaliseAddInput(source),
+      doImport(self, opts),
+      transformFile(self, opts),
+      preloadFile(self, opts),
+      pinFile(self, opts)
+    )
+
+    const releaseLock = await self._gcLock.readLock()
+
+    try {
+      yield * iterator
+    } finally {
+      releaseLock()
+    }
+  }
+}
+
+function doImport (ipfs, opts) {
+  return async function * (source) { // eslint-disable-line require-await
+    yield * importer(source, ipfs._ipld, opts)
+  }
+}
+
+function transformFile (ipfs, opts) {
+  return async function * (source) {
+    for await (const file of source) {
+      let cid = file.cid
+      const hash = cid.toBaseEncodedString()
+      let path = file.path ? file.path : hash
+
+      if (opts.wrapWithDirectory && !file.path) {
+        path = ''
+      }
+
+      if (opts.onlyHash) {
+        yield {
+          path,
+          hash,
+          size: file.unixfs.fileSize()
+        }
+
+        return
+      }
+
+      const node = await ipfs.object.get(file.cid, Object.assign({}, opts, { preload: false }))
+
+      if (opts.cidVersion === 1) {
+        cid = cid.toV1()
+      }
+
+      let size = node.size
+
+      if (Buffer.isBuffer(node)) {
+        size = node.length
+      }
+
+      yield {
+        path,
+        hash,
+        size
+      }
+    }
+  }
+}
+
+function preloadFile (ipfs, opts) {
+  return async function * (source) {
+    for await (const file of source) {
+      const isRootFile = !file.path || opts.wrapWithDirectory
+        ? file.path === ''
+        : !file.path.includes('/')
+
+      const shouldPreload = isRootFile && !opts.onlyHash && opts.preload !== false
+
+      if (shouldPreload) {
+        ipfs._preload(file.hash)
+      }
+
+      yield file
+    }
+  }
+}
+
+function pinFile (ipfs, opts) {
+  return async function * (source) {
+    for await (const file of source) {
+      // Pin a file if it is the root dir of a recursive add or the single file
+      // of a direct add.
+      const pin = 'pin' in opts ? opts.pin : true
+      const isRootDir = !file.path.includes('/')
+      const shouldPin = pin && isRootDir && !opts.onlyHash && !opts.hashAlg
+
+      if (shouldPin) {
+        // Note: addAsyncIterator() has already taken a GC lock, so tell
+        // pin.add() not to take a (second) GC lock
+        await ipfs.pin.add(file.hash, {
+          preload: false,
+          lock: false
+        })
+      }
+
+      yield file
+    }
+  }
+}