feat: support adding async iterators

Adds a `ipfs._addAsyncIterator` method for adding async iterators and
refactors all add methods to call this, as when the Great Async Iteator
Migration is complete this will become the one, true method to add
files to IPFS.

Author: achingbrain

package.json

@@ -98,13 +98,13 @@
     "ipfs-block-service": "~0.15.2",
     "ipfs-http-client": "^34.0.0",
     "ipfs-http-response": "~0.3.1",
-    "ipfs-mfs": "~0.12.0",
-    "ipfs-multipart": "~0.1.1",
+    "ipfs-mfs": "ipfs/js-ipfs-mfs#v0.12.x-update-ipfs-multipart",
+    "ipfs-multipart": "^0.2.0",
     "ipfs-repo": "~0.26.6",
     "ipfs-unixfs": "~0.1.16",
     "ipfs-unixfs-exporter": "~0.37.7",
     "ipfs-unixfs-importer": "~0.39.11",
-    "ipfs-utils": "~0.0.4",
+    "ipfs-utils": "achingbrain/js-ipfs-utils#support-async-iterators",
     "ipld": "~0.24.1",
     "ipld-bitcoin": "~0.3.0",
     "ipld-dag-cbor": "~0.15.0",
@@ -119,6 +119,9 @@
     "is-pull-stream": "~0.0.0",
     "is-stream": "^2.0.0",
     "iso-url": "~0.4.6",
+    "it-glob": "^0.0.1",
+    "it-pipe": "^1.0.1",
+    "it-to-stream": "^0.1.1",
     "just-safe-set": "^2.1.0",
     "kind-of": "^6.0.2",
     "libp2p": "~0.25.4",

src/cli/commands/add.js

@@ -1,14 +1,10 @@
 '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')
@@ -18,48 +14,6 @@ async function getTotalBytes (paths) {
   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...]',
 
@@ -200,16 +154,51 @@ module.exports = {
 
       const source = argv.file
         ? globSource(...argv.file, { recursive: argv.recursive })
-        : toPull.source(process.stdin) // Pipe directly to ipfs.add
+        : 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,206 @@
+'use strict'
+
+const importer = require('ipfs-unixfs-importer')
+const isSource = require('is-pull-stream').isSource
+const validateAddInput = require('ipfs-utils/src/files/add-input-validation')
+const { parseChunkerString } = require('./utils')
+const pipe = require('it-pipe')
+const { supportsFileReader } = require('ipfs-utils/src/supports')
+const toAsyncIterator = require('pull-stream-to-async-iterator')
+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 || {}
+
+    let 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
+
+    if (Buffer.isBuffer(source) || typeof source === 'string') {
+      source = [
+        source
+      ]
+    }
+
+    const iterator = pipe(
+      source,
+      validateInput(),
+      normalizeInput(opts),
+      doImport(self, opts),
+      prepareFile(self, opts),
+      preloadFile(self, opts),
+      pinFile(self, opts)
+    )
+
+    const releaseLock = await self._gcLock.readLock()
+
+    try {
+      yield * iterator
+    } finally {
+      releaseLock()
+    }
+  }
+}
+
+function validateInput () {
+  return async function * (source) {
+    for await (const data of source) {
+      validateAddInput(data)
+
+      yield data
+    }
+  }
+}
+
+function normalizeContent (content) {
+  if (supportsFileReader && kindOf(content) === 'file') {
+    return streamFromFileReader(content)
+  }
+
+  // pull stream source
+  if (isSource(content)) {
+    return toAsyncIterator(content)
+  }
+
+  if (typeof content === 'string') {
+    return Buffer.from(content)
+  }
+
+  if (Array.isArray(content) && content.length && !Array.isArray(content[0])) {
+    return [content]
+  }
+
+  return content
+}
+
+function normalizeInput (opts) {
+  return async function * (source) {
+    for await (let data of source) {
+      if (data.content) {
+        data.content = normalizeContent(data.content)
+      } else {
+        data = {
+          path: '',
+          content: normalizeContent(data)
+        }
+      }
+
+      if (opts.wrapWithDirectory && !data.path) {
+        throw new Error('Must provide a path when wrapping with a directory')
+      }
+
+      yield data
+    }
+  }
+}
+
+function doImport (ipfs, opts) {
+  return function (source) {
+    return importer(source, ipfs._ipld, opts)
+  }
+}
+
+function prepareFile (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) {
+        await ipfs.pin.add(file.hash, { preload: false })
+      }
+
+      yield file
+    }
+  }
+}