Merge pull request #499 from ipfs/fix/files-get

Fix/files get

Author: daviddias

gulpfile.js

@@ -1,8 +1,8 @@
 'use strict'
 
 const gulp = require('gulp')
-const parallel = require('run-parallel')
-const series = require('run-series')
+const parallel = require('async/parallel')
+const series = require('async/series')
 const createTempNode = require('./test/utils/temp-node')
 const API = require('./src/http-api')
 

package.json

@@ -43,6 +43,7 @@
     "aegir": "^8.0.1",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
+    "execa": "^0.4.0",
     "expose-loader": "^0.7.1",
     "form-data": "^2.0.0",
     "fs-pull-blob-store": "^0.3.0",
@@ -59,6 +60,7 @@
     "transform-loader": "^0.2.3"
   },
   "dependencies": {
+    "async": "^2.0.1",
     "babel-runtime": "^6.11.6",
     "bl": "^1.1.2",
     "boom": "^4.0.0",

src/cli/commands/bitswap/stat.js

@@ -23,8 +23,7 @@ module.exports = {
         stats.Wantlist = stats.Wantlist || []
         stats.Peers = stats.Peers || []
 
-        console.log(`
-bitswap status
+        console.log(`bitswap status
   blocks received: ${stats.BlocksReceived}
   dup blocks received: ${stats.DupBlksReceived}
   dup data received: ${stats.DupDataReceived}B

src/cli/commands/block/stat.js

@@ -1,7 +1,6 @@
 'use strict'
 
 const utils = require('../../utils')
-const bs58 = require('bs58')
 const debug = require('debug')
 const log = debug('cli:block')
 log.error = debug('cli:block:error')
@@ -24,7 +23,7 @@ module.exports = {
           throw err
         }
 
-        console.log('Key:', bs58.encode(stats.key).toString())
+        console.log('Key:', stats.key)
         console.log('Size:', stats.size)
       })
     })

src/cli/commands/bootstrap/add.js

@@ -17,6 +17,7 @@ module.exports = {
       if (err) {
         throw err
       }
+
       ipfs.bootstrap.add(argv.peer, (err, list) => {
         if (err) {
           throw err

src/cli/commands/config/edit.js

@@ -3,7 +3,7 @@
 const spawn = require('child_process').spawn
 const fs = require('fs')
 const temp = require('temp')
-const waterfall = require('run-waterfall')
+const waterfall = require('async/waterfall')
 const debug = require('debug')
 const log = debug('cli:config')
 log.error = debug('cli:config:error')

src/cli/commands/files/cat.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const waterfall = require('async/waterfall')
 const debug = require('debug')
 const utils = require('../../utils')
 const log = debug('cli:files')
@@ -14,19 +15,16 @@ module.exports = {
 
   handler (argv) {
     const path = argv['ipfs-path']
-    utils.getIPFS((err, ipfs) => {
+
+    waterfall([
+      (cb) => utils.getIPFS(cb),
+      (ipfs, cb) => ipfs.files.cat(path, cb)
+    ], (err, file) => {
       if (err) {
         throw err
       }
 
-      ipfs.files.cat(path, onFile)
+      file.pipe(process.stdout)
     })
   }
 }
-
-function onFile (err, file) {
-  if (err) {
-    throw (err)
-  }
-  file.pipe(process.stdout)
-}

src/cli/commands/files/get.js

@@ -7,6 +7,8 @@ log.error = debug('cli:files:error')
 var fs = require('fs')
 const path = require('path')
 const pathExists = require('path-exists')
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
 
 function checkArgs (hash, outPath) {
   // format the output directory
@@ -33,30 +35,39 @@ function ensureDir (dir, cb) {
     .catch(cb)
 }
 
-function fileHandler (result, dir) {
-  return function onFile (file) {
+function fileHandler (dir) {
+  return function onFile (file, cb) {
+    const lastSlash = file.path.lastIndexOf('/')
     // Check to see if the result is in a directory
-    if (file.path.lastIndexOf('/') === -1) {
+    if (lastSlash === -1) {
       const dirPath = path.join(dir, file.path)
       // Check to see if the result is a directory
-      if (file.dir === false) {
+      if (file.content) {
         file.content.pipe(fs.createWriteStream(dirPath))
+          .once('error', cb)
+          .once('end', cb)
       } else {
-        ensureDir(dirPath, (err) => {
-          if (err) {
-            throw err
-          }
-        })
+        ensureDir(dirPath, cb)
       }
     } else {
-      const filePath = file.path.substring(0, file.path.lastIndexOf('/') + 1)
+      const filePath = file.path.substring(0, lastSlash + 1)
       const dirPath = path.join(dir, filePath)
+
       ensureDir(dirPath, (err) => {
         if (err) {
-          throw err
+          return cb(err)
         }
 
-        file.content.pipe(fs.createWriteStream(dirPath))
+        if (file.content) {
+          const filename = file.path.substring(lastSlash)
+          const target = path.join(dirPath, filename)
+
+          file.content.pipe(fs.createWriteStream(target))
+            .once('error', cb)
+            .once('end', cb)
+          return
+        }
+        cb()
       })
     }
   }
@@ -76,17 +87,28 @@ module.exports = {
   },
 
   handler (argv) {
-    const dir = checkArgs(argv.ipfsPath, argv.output)
+    const ipfsPath = argv['ipfs-path']
+    const dir = checkArgs(ipfsPath, argv.output)
 
     utils.getIPFS((err, ipfs) => {
       if (err) {
         throw err
       }
-      ipfs.files.get(argv.ipfsPath, (err, result) => {
+
+      ipfs.files.get(ipfsPath, (err, stream) => {
         if (err) {
           throw err
         }
-        result.on('data', fileHandler(result, dir))
+        console.log(`Saving file(s) to ${ipfsPath}`)
+        pull(
+          toPull.source(stream),
+          pull.asyncMap(fileHandler(dir)),
+          pull.onEnd((err) => {
+            if (err) {
+              throw err
+            }
+          })
+        )
       })
     })
   }

src/core/components/block.js

@@ -33,7 +33,7 @@ module.exports = function block (self) {
           return callback(err)
         }
         callback(null, {
-          key: hash,
+          key: multihash.toB58String(hash),
           size: block.data.length
         })
       })

src/core/components/go-online.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const series = require('run-series')
+const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
 
 module.exports = function goOnline (self) {

src/core/components/object.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const mDAG = require('ipfs-merkle-dag')
-const waterfall = require('run-waterfall')
+const waterfall = require('async/waterfall')
 const promisify = require('promisify-es6')
 const bs58 = require('bs58')
 const DAGNode = mDAG.DAGNode
@@ -69,12 +69,7 @@ module.exports = function object (self) {
             cb(err, node)
           })
         }
-      ], (err, node) => {
-        if (err) {
-          return cb(err)
-        }
-        cb(null, node)
-      })
+      ], cb)
     }
   }
 

src/http-api/index.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const parallel = require('run-parallel')
+const parallel = require('async/parallel')
 const Hapi = require('hapi')
 const debug = require('debug')
 const fs = require('fs')

src/http-api/resources/block.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const bs58 = require('bs58')
+const mh = require('multihashes')
 const multipart = require('ipfs-multipart')
 const Block = require('ipfs-block')
 const debug = require('debug')
@@ -17,7 +17,7 @@ exports.parseKey = (request, reply) => {
 
   try {
     return reply({
-      key: new Buffer(bs58.decode(request.query.arg))
+      key: mh.fromB58String(request.query.arg)
     })
   } catch (err) {
     log.error(err)
@@ -93,7 +93,7 @@ exports.put = {
       }
 
       return reply({
-        Key: bs58.encode(block.key).toString(),
+        Key: mh.toB58String(block.key),
         Size: block.data.length
       })
     })
@@ -112,7 +112,7 @@ exports.del = {
       if (err) {
         log.error(err)
         return reply({
-          Message: 'Failed to get block stats: ' + err,
+          Message: 'Failed to delete block: ' + err,
           Code: 0
         }).code(500)
       }
@@ -129,7 +129,7 @@ exports.stat = {
   // main route handler which is called after the above `parseArgs`, but only if the args were valid
   handler: (request, reply) => {
     const key = request.pre.args.key
-
+    console.log('fetching', key)
     request.server.app.ipfs.block.stat(key, (err, block) => {
       if (err) {
         log.error(err)
@@ -140,7 +140,7 @@ exports.stat = {
       }
 
       return reply({
-        Key: bs58.encode(block.key).toString(),
+        Key: block.key,
         Size: block.size
       })
     })

src/http-api/resources/bootstrap.js

@@ -1,28 +1,64 @@
 'use strict'
 
 const boom = require('boom')
+const multiaddr = require('multiaddr')
 
 exports = module.exports
 
+// common pre request handler that parses the args and returns `key` which is assigned to `request.pre.args`
+exports.parseKey = (request, reply) => {
+  if (!request.query.arg) {
+    return reply("Argument 'multiaddr' is required").code(400).takeover()
+  }
+
+  try {
+    return reply({
+      addr: multiaddr(request.query.arg)
+    })
+  } catch (err) {
+    return reply({
+      Message: 'Not a valid multiaddr',
+      Code: 0
+    }).code(500).takeover()
+  }
+}
+
 exports.list = (request, reply) => {
-  request.server.app.ipfs.bootstrap.list((err, list) => {
+  const ipfs = request.server.app.ipfs
+  ipfs.bootstrap.list((err, list) => {
     if (err) {
       return reply(boom.badRequest(err))
     }
     return reply(list)
   })
 }
 
-exports.add = (request, reply) => {
-//  request.server.app.ipfs.id((err, id) => {
-//    if (err) { return reply(boom.badRequest(err)) }
-//    return reply(id)
-//   })
+exports.add = {
+  parseArgs: exports.parseKey,
+  handler (request, reply) {
+    const ipfs = request.server.app.ipfs
+    const addr = request.pre.args.addr
+
+    ipfs.bootstrap.add(addr.toString(), (err, list) => {
+      if (err) {
+        return reply(boom.badRequest(err))
+      }
+      return reply()
+    })
+  }
 }
 
-exports.rm = (request, reply) => {
-//  request.server.app.ipfs.id((err, id) => {
-//    if (err) { return reply(boom.badRequest(err)) }
-//    return reply(id)
-//   })
+exports.rm = {
+  parseArgs: exports.parseKey,
+  handler (request, reply) {
+    const ipfs = request.server.app.ipfs
+    const addr = request.pre.args.addr
+
+    ipfs.bootstrap.rm(addr.toString(), (err, list) => {
+      if (err) {
+        return reply(boom.badRequest(err))
+      }
+      return reply()
+    })
+  }
 }