Merge pull request #440 from ipfs/pull

[WIP] Move to pull-streams

package.json

@@ -40,16 +40,18 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs#readme",
   "devDependencies": {
-    "aegir": "^8.0.0",
+    "aegir": "^8.0.1",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "expose-loader": "^0.7.1",
-    "form-data": "^1.0.0-rc4",
+    "form-data": "^1.0.1",
+    "fs-pull-blob-store": "^0.3.0",
     "gulp": "^3.9.1",
     "idb-plus-blob-store": "^1.1.2",
-    "interface-ipfs-core": "^0.14.3",
+    "idb-pull-blob-store": "^0.4.0",
+    "interface-ipfs-core": "^0.14.4",
     "left-pad": "^1.1.1",
-    "lodash": "^4.14.1",
+    "lodash": "^4.15.0",
     "ncp": "^2.0.0",
     "nexpect": "^0.5.0",
     "pre-commit": "^1.1.3",
@@ -60,45 +62,50 @@
   "dependencies": {
     "babel-runtime": "^6.11.6",
     "bl": "^1.1.2",
-    "boom": "^3.2.2",
+    "boom": "^4.0.0",
     "bs58": "^3.0.0",
     "debug": "^2.2.0",
     "detect-node": "^2.0.3",
-    "fs-blob-store": "^5.2.1",
-    "glob": "^7.0.5",
+    "glob": "^7.0.6",
     "hapi": "^15.0.3",
     "ipfs-api": "^8.0.3",
-    "ipfs-bitswap": "^0.6.0",
+    "ipfs-bitswap": "^0.7.0",
     "ipfs-block": "^0.3.0",
-    "ipfs-block-service": "^0.4.0",
-    "ipfs-merkle-dag": "^0.6.2",
+    "ipfs-block-service": "^0.5.0",
+    "ipfs-merkle-dag": "^0.7.0",
     "ipfs-multipart": "^0.1.0",
-    "ipfs-repo": "^0.8.0",
+    "ipfs-repo": "^0.9.0",
     "ipfs-unixfs": "^0.1.4",
-    "ipfs-unixfs-engine": "^0.10.1",
+    "ipfs-unixfs-engine": "^0.11.2",
     "isstream": "^0.1.2",
     "joi": "^9.0.4",
-    "libp2p-ipfs": "^0.12.1",
-    "libp2p-ipfs-browser": "^0.12.1",
-    "lodash.get": "^4.4.1",
+    "libp2p-ipfs": "^0.13.0",
+    "libp2p-ipfs-browser": "^0.14.0",
+    "lodash.get": "^4.4.2",
     "lodash.has": "^4.5.2",
-    "lodash.set": "^4.3.1",
-    "lodash.sortby": "^4.6.1",
-    "mafmt": "^2.1.1",
+    "lodash.set": "^4.3.2",
+    "lodash.sortby": "^4.7.0",
+    "mafmt": "^2.1.2",
     "map-limit": "0.0.1",
-    "multiaddr": "^2.0.2",
+    "multiaddr": "^2.0.3",
     "multihashes": "^0.2.2",
     "path-exists": "^3.0.0",
     "peer-book": "^0.3.0",
     "peer-id": "^0.7.0",
-    "peer-info": "^0.7.0",
+    "peer-info": "^0.7.1",
     "promisify-es6": "^1.0.1",
+    "pull-file": "^1.0.0",
+    "pull-paramap": "^1.1.6",
+    "pull-sort": "^1.0.0",
+    "pull-stream": "^3.4.5",
+    "pull-stream-to-stream": "^1.3.3",
+    "pull-zip": "^2.0.0",
     "read-pkg-up": "^1.0.1",
-    "readable-stream": "1.1.13",
     "run-parallel": "^1.1.6",
     "run-parallel-limit": "^1.0.3",
     "run-series": "^1.1.4",
     "run-waterfall": "^1.1.3",
+    "stream-to-pull-stream": "^1.7.0",
     "temp": "^0.8.3",
     "through2": "^2.0.1",
     "update-notifier": "^1.0.2",

src/cli/commands/files/add.js

@@ -8,7 +8,10 @@ const fs = require('fs')
 const path = require('path')
 const glob = require('glob')
 const sortBy = require('lodash.sortby')
-const mapLimit = require('map-limit')
+const pull = require('pull-stream')
+const pullFile = require('pull-file')
+const paramap = require('pull-paramap')
+const zip = require('pull-zip')
 
 function checkPath (inPath, recursive) {
   // This function is to check for the following possible inputs
@@ -48,76 +51,48 @@ module.exports = {
   },
 
   handler (argv) {
-    let inPath = checkPath(argv.file, argv.recursive)
+    const inPath = checkPath(argv.file, argv.recursive)
+    const index = inPath.lastIndexOf('/') + 1
+
+    utils.getIPFS((err, ipfs) => {
+      if (err) throw err
+
+      glob(path.join(inPath, '/**/*'), (err, list) => {
+        if (err) throw err
+        if (list.length === 0) list = [inPath]
+
+        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: pullFile(file.path)
+          })),
+          ipfs.files.createAddPullStream(),
+          pull.map((file) => ({
+            hash: file.hash,
+            path: file.path
+          })),
+          pull.collect((err, added) => {
+            if (err) throw err
 
-    utils.getIPFS(gotIPFS)
-
-    function gotIPFS (err, ipfs) {
-      if (err) {
-        throw err
-      }
-
-      glob(path.join(inPath, '/**/*'), (err, res) => {
-        if (err) {
-          throw err
-        }
-
-        ipfs.files.createAddStream((err, i) => {
-          if (err) {
-            throw err
-          }
-          const added = []
-
-          i.on('data', (file) => {
-            added.push({
-              hash: file.hash,
-              path: file.path
-            })
-          })
-
-          i.on('end', () => {
             sortBy(added, 'path')
               .reverse()
               .map((file) => `added ${file.hash} ${file.path}`)
               .forEach((msg) => console.log(msg))
           })
-
-          if (res.length === 0) {
-            res = [inPath]
-          }
-
-          const writeToStream = (stream, element) => {
-            const index = inPath.lastIndexOf('/') + 1
-            i.write({
-              path: element.substring(index, element.length),
-              content: fs.createReadStream(element)
-            })
-          }
-
-          mapLimit(res, 50, (file, cb) => {
-            fs.stat(file, (err, stat) => {
-              if (err) {
-                return cb(err)
-              }
-              return cb(null, {
-                path: file,
-                isDirectory: stat.isDirectory()
-              })
-            })
-          }, (err, res) => {
-            if (err) {
-              throw err
-            }
-
-            res
-              .filter((elem) => !elem.isDirectory)
-              .map((elem) => elem.path)
-              .forEach((elem) => writeToStream(i, elem))
-
-            i.end()
-          })
-        })
+        )
       })
-    }
+    })
   }
 }

src/cli/commands/init.js

@@ -2,7 +2,7 @@
 
 const IpfsRepo = require('ipfs-repo')
 const Ipfs = require('../../core')
-const fsBlobStore = require('fs-blob-store')
+const Store = require('fs-pull-blob-store')
 const utils = require('../utils')
 
 module.exports = {
@@ -33,7 +33,7 @@ module.exports = {
     const path = utils.getRepoPath()
 
     const repo = new IpfsRepo(path, {
-      stores: fsBlobStore
+      stores: Store
     })
 
     const ipfs = new Ipfs(repo)

src/core/default-repo/browser.js

@@ -1,9 +1,9 @@
 'use strict'
 
-const idb = require('idb-plus-blob-store')
+const Store = require('idb-pull-blob-store')
 const IPFSRepo = require('ipfs-repo')
 
 module.exports = (dir) => {
   const repoPath = dir || 'ipfs'
-  return new IPFSRepo(repoPath, {stores: idb})
+  return new IPFSRepo(repoPath, {stores: Store})
 }

src/core/default-repo/node.js

@@ -1,11 +1,11 @@
 'use strict'
 
 const os = require('os')
-const fs = require('fs-blob-store')
+const Store = require('fs-pull-blob-store')
 const IPFSRepo = require('ipfs-repo')
 const path = require('path')
 
 module.exports = (dir) => {
   const repoPath = dir || path.join(os.homedir(), '.ipfs')
-  return new IPFSRepo(repoPath, {stores: fs})
+  return new IPFSRepo(repoPath, {stores: Store})
 }

src/core/ipfs/block.js

@@ -7,8 +7,7 @@ module.exports = function block (self) {
   return {
     get: (hash, callback) => {
       hash = cleanHash(hash)
-
-      self._blockS.getBlock(hash, callback)
+      self._blockS.get(hash, callback)
     },
     put: (block, callback) => {
       if (Array.isArray(block)) {
@@ -18,18 +17,18 @@ module.exports = function block (self) {
         block = new Block(block)
       }
 
-      self._blockS.addBlock(block, (err) => {
+      self._blockS.put(block, (err) => {
         callback(err, block)
       })
     },
     del: (hash, callback) => {
       hash = cleanHash(hash)
-      self._blockS.deleteBlock(hash, callback)
+      self._blockS.delete(hash, callback)
     },
     stat: (hash, callback) => {
       hash = cleanHash(hash)
 
-      self._blockS.getBlock(hash, (err, block) => {
+      self._blockS.get(hash, (err, block) => {
         if (err) {
           return callback(err)
         }