feat: add addFromFs method

This PR adds a new method [`addFromFs`](https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/FILES.md#addfromfs) allowing users to more easily add files from their file system without having to specify every single file to add.

In the browser the user will receive a "not available" error.

I've pulled out a module `glob-source.js` - call it with some file paths and it returns a pull stream source that can be piped to `ipfs.addPullStream`.

This PR comes with the following added benefits:

* `ipfs add` on the CLI uses `glob-source.js` - **nice and DRY**
* `glob-source.js` uses the events that the `glob` module provides allowing the globbing to be a `pull-pushable`, which means that matched paths can begin to be added before all the globbing is done - **faster**
* `ipfs add` now supports adding multiple paths, fixes #1625 - **better**
* `ipfs add --progress=false` doesn't calculate the total size of the files to be added anymore! It didn't need to do that as the total was completely discarded when progress was disabled. It means we can add BIGGER directories without running into memory issues - **stronger**

License: MIT
Signed-off-by: Alan Shaw <alan.shaw@protocol.ai>

package.json

@@ -94,7 +94,6 @@
     "datastore-core": "~0.6.0",
     "datastore-pubsub": "~0.1.1",
     "debug": "^4.1.0",
-    "deep-extend": "~0.6.0",
     "err-code": "^1.1.2",
     "file-type": "^10.2.0",
     "fnv1a": "^1.0.1",
@@ -160,16 +159,14 @@
     "promisify-es6": "^1.0.3",
     "protons": "^1.0.1",
     "pull-abortable": "^4.1.1",
-    "pull-catch": "^1.0.0",
+    "pull-cat": "^1.1.11",
     "pull-defer": "~0.2.3",
     "pull-file": "^1.1.0",
     "pull-ndjson": "~0.1.1",
-    "pull-paramap": "^1.2.2",
     "pull-pushable": "^2.2.0",
     "pull-sort": "^1.0.1",
     "pull-stream": "^3.6.9",
     "pull-stream-to-stream": "^1.3.4",
-    "pull-zip": "^2.0.1",
     "pump": "^3.0.0",
     "read-pkg-up": "^4.0.0",
     "readable-stream": "3.0.6",

src/cli/commands/add.js

@@ -1,86 +1,41 @@
 'use strict'
 
-const fs = require('fs')
-const path = require('path')
-const glob = require('glob')
 const sortBy = require('lodash/sortBy')
 const pull = require('pull-stream')
-const paramap = require('pull-paramap')
-const zip = require('pull-zip')
 const getFolderSize = require('get-folder-size')
 const byteman = require('byteman')
-const waterfall = require('async/waterfall')
+const reduce = require('async/reduce')
 const mh = require('multihashes')
 const utils = require('../utils')
 const print = require('../utils').print
 const createProgressBar = require('../utils').createProgressBar
+const globSource = require('../../utils/files/glob-source')
 
-function checkPath (inPath, recursive) {
-  // This function is to check for the following possible inputs
-  // 1) "." add the cwd but throw error for no recursion flag
-  // 2) "." -r return the cwd
-  // 3) "/some/path" but throw error for no recursion
-  // 4) "/some/path" -r
-  // 5) No path, throw err
-  // 6) filename.type return the cwd + filename
-
-  if (!inPath) {
-    throw new Error('Error: Argument \'path\' is required')
-  }
-
-  if (inPath === '.') {
-    inPath = process.cwd()
-  }
-
-  // Convert to POSIX format
-  inPath = inPath
-    .split(path.sep)
-    .join('/')
-
-  // Strips trailing slash from path.
-  inPath = inPath.replace(/\/$/, '')
-
-  if (fs.statSync(inPath).isDirectory() && recursive === false) {
-    throw new Error(`Error: ${inPath} is a directory, use the '-r' flag to specify directories`)
-  }
-
-  return inPath
-}
-
-function getTotalBytes (path, recursive, cb) {
-  if (recursive) {
-    getFolderSize(path, cb)
-  } else {
-    fs.stat(path, (err, stat) => cb(err, stat.size))
-  }
+function getTotalBytes (paths, cb) {
+  reduce(paths, 0, (total, path, cb) => {
+    getFolderSize(path, (err, size) => {
+      if (err) return cb(err)
+      cb(null, total + size)
+    })
+  }, cb)
 }
 
-function addPipeline (index, addStream, list, argv) {
+function addPipeline (paths, addStream, options) {
   const {
+    recursive,
     quiet,
     quieter,
     silent
-  } = argv
+  } = options
   pull(
-    zip(
-      pull.values(list),
-      pull(
-        pull.values(list),
-        paramap(fs.stat.bind(fs), 50)
-      )
-    ),
-    pull.map((pair) => ({
-      path: pair[0],
-      isDirectory: pair[1].isDirectory()
-    })),
-    pull.filter((file) => !file.isDirectory),
-    pull.map((file) => ({
-      path: file.path.substring(index, file.path.length),
-      content: fs.createReadStream(file.path)
-    })),
+    globSource(...paths, { recursive }),
     addStream,
     pull.collect((err, added) => {
       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`
+        }
         throw err
       }
 
@@ -102,7 +57,7 @@ function addPipeline (index, addStream, list, argv) {
 }
 
 module.exports = {
-  command: 'add <file>',
+  command: 'add <file...>',
 
   describe: 'Add a file to IPFS using the UnixFS data format',
 
@@ -186,8 +141,7 @@ module.exports = {
   },
 
   handler (argv) {
-    const inPath = checkPath(argv.file, argv.recursive)
-    const index = inPath.lastIndexOf('/') + 1
+    const { ipfs } = argv
     const options = {
       strategy: argv.trickle ? 'trickle' : 'balanced',
       shardSplitThreshold: argv.enableShardingExperiment
@@ -205,38 +159,21 @@ module.exports = {
     if (options.enableShardingExperiment && utils.isDaemonOn()) {
       throw new Error('Error: Enabling the sharding experiment should be done on the daemon')
     }
-    const ipfs = argv.ipfs
 
-    let list
-    waterfall([
-      (next) => {
-        if (fs.statSync(inPath).isDirectory()) {
-          return glob('**/*', { cwd: inPath }, next)
-        }
-        next(null, [])
-      },
-      (globResult, next) => {
-        if (globResult.length === 0) {
-          list = [inPath]
-        } else {
-          list = globResult.map((f) => inPath + '/' + f)
-        }
-        getTotalBytes(inPath, argv.recursive, next)
-      },
-      (totalBytes, next) => {
-        if (argv.progress) {
-          const bar = createProgressBar(totalBytes)
-          options.progress = function (byteLength) {
-            bar.update(byteLength / totalBytes, { progress: byteman(byteLength, 2, 'MB') })
-          }
-        }
+    if (!argv.progress) {
+      return addPipeline(argv.file, ipfs.addPullStream(options), argv)
+    }
 
-        next(null, ipfs.addPullStream(options))
-      }
-    ], (err, addStream) => {
+    getTotalBytes(argv.file, (err, totalBytes) => {
       if (err) throw err
 
-      addPipeline(index, addStream, list, argv)
+      const bar = createProgressBar(totalBytes)
+
+      options.progress = byteLength => {
+        bar.update(byteLength / totalBytes, { progress: byteman(byteLength, 2, 'MB') })
+      }
+
+      addPipeline(argv.file, ipfs.addPullStream(options), argv)
     })
   }
 }

src/core/runtime/add-from-fs-browser.js

@@ -1,3 +1,10 @@
 'use strict'
 
-module.exports = () => null
+const promisify = require('promisify-es6')
+
+module.exports = self => {
+  return promisify((...args) => {
+    const callback = args.pop()
+    callback(new Error('not available in the browser'))
+  })
+}

src/core/runtime/add-from-fs-nodejs.js

@@ -1 +1,20 @@
 'use strict'
+
+const promisify = require('promisify-es6')
+const pull = require('pull-stream')
+const globSource = require('../../utils/files/glob-source')
+const isString = require('lodash/isString')
+
+module.exports = self => {
+  return promisify((...args) => {
+    const callback = args.pop()
+    const options = isString(args[args.length - 1]) ? {} : args.pop()
+    const paths = args
+
+    pull(
+      globSource(...paths, options),
+      self.addPullStream(options),
+      pull.collect(callback)
+    )
+  })
+}

src/utils/files/glob-source.js

@@ -0,0 +1,108 @@
+'use strict'
+
+const fs = require('fs')
+const Path = require('path')
+const isString = require('lodash/isString')
+const pull = require('pull-stream')
+const glob = require('glob')
+const cat = require('pull-cat')
+const defer = require('pull-defer')
+const pushable = require('pull-pushable')
+const map = require('async/map')
+const parallel = require('async/parallel')
+const errCode = require('err-code')
+
+/**
+* Create a pull stream source that can be piped to ipfs.addPullStream for the
+* provided file paths.
+*
+* @param ...paths {String} File system path(s) to glob from
+* @param [options] {Object} Optional options
+* @param [options.recursive] Recursively glob all paths in directories
+*/
+module.exports = (...args) => {
+  const options = isString(args[args.length - 1]) ? {} : args.pop()
+  const paths = args
+  const deferred = defer.source()
+
+  const globSourceOptions = {
+    recursive: options.recursive,
+    glob: {
+      dot: Boolean(options.hidden),
+      ignore: Array.isArray(options.ignore) ? options.ignore : [],
+      follow: options.followSymlinks != null ? options.followSymlinks : true
+    }
+  }
+
+  // Check the input paths comply with options.recursive and convert to glob sources
+  map(paths, normalizePathWithType, (err, results) => {
+    if (err) return deferred.abort(err)
+
+    try {
+      const sources = results.map(res => toGlobSource(res, globSourceOptions))
+      return deferred.resolve(cat(sources))
+    } catch (err) {
+      return deferred.abort(err)
+    }
+  })
+
+  return pull(
+    deferred,
+    pull.map(({ path, contentPath }) => ({
+      path,
+      content: fs.createReadStream(contentPath)
+    }))
+  )
+}
+
+function toGlobSource ({ path, type }, options) {
+  options = options || {}
+
+  const baseName = Path.basename(path)
+
+  if (type === 'file') {
+    return pull.values([{ path: baseName, contentPath: path }])
+  }
+
+  if (type === 'dir' && !options.recursive) {
+    throw errCode(
+      new Error(`'${path}' is a directory and recursive option not set`),
+      'ERR_DIR_NON_RECURSIVE',
+      { path }
+    )
+  }
+
+  const globOptions = Object.assign({}, options.glob, {
+    cwd: path,
+    nodir: true,
+    realpath: false,
+    absolute: false
+  })
+
+  // TODO: want to use pull-glob but it doesn't have the features...
+  const pusher = pushable()
+
+  glob('**/*', globOptions)
+    .on('match', m => pusher.push(m))
+    .on('end', () => pusher.end())
+    .on('abort', () => pusher.end())
+    .on('error', err => pusher.end(err))
+
+  return pull(
+    pusher,
+    pull.map(p => ({
+      path: Path.join(baseName, p),
+      contentPath: Path.join(path, p)
+    }))
+  )
+}
+
+function normalizePathWithType (path, cb) {
+  parallel({
+    stat: cb => fs.stat(path, cb),
+    realpath: cb => fs.realpath(path, cb)
+  }, (err, res) => {
+    if (err) return cb(err)
+    cb(null, { path: res.realpath, type: res.stat.isDirectory() ? 'dir' : 'file' })
+  })
+}

test/cli/files.js

@@ -143,6 +143,18 @@ describe('files', () => runOnAndOff((thing) => {
       })
   })
 
+  it('add multiple', function () {
+    this.timeout(30 * 1000)
+
+    return ipfs('add', 'src/init-files/init-docs/readme', 'test/fixtures/odd-name-[v0]/odd name [v1]/hello', '--wrap-with-directory')
+      .then((out) => {
+        expect(out)
+          .to.include('added QmPZ9gcCEpqKTo6aq61g2nXGUhM4iCL3ewB6LDXZCtioEB readme\n')
+        expect(out)
+          .to.include('added QmT78zSuBmuS4z925WZfrqQ1qHaJ56DQaTfyMUF7F8ff5o hello\n')
+      })
+  })
+
   it('add alias', function () {
     this.timeout(30 * 1000)
 
@@ -278,7 +290,7 @@ describe('files', () => runOnAndOff((thing) => {
   it('add --quieter', function () {
     this.timeout(30 * 1000)
 
-    return ipfs('add -Q -w test/fixtures/test-data/hello test/test-data/node.json')
+    return ipfs('add -Q -w test/fixtures/test-data/hello')
       .then((out) => {
         expect(out)
           .to.eql('QmYRMUVULBfj7WrdPESnwnyZmtayN6Sdrwh1nKcQ9QgQeZ\n')

test/core/interface.spec.js

@@ -79,10 +79,7 @@ describe('interface-ipfs-core tests', () => {
   })
 
   tests.filesRegular(defaultCommonFactory, {
-    skip: isNode ? [{
-      name: 'addFromFs',
-      reason: 'TODO: not implemented yet'
-    }] : [{
+    skip: isNode ? null : [{
       name: 'addFromStream',
       reason: 'Not designed to run in the browser'
     }, {