add http resource, files core upgrades, cat cli upgrades, begin interface-core integration

package.json

@@ -78,6 +78,7 @@
     "lodash.get": "^4.3.0",
     "lodash.set": "^4.2.0",
     "multiaddr": "^2.0.0",
+    "multihashes": "^0.2.2",
     "path-exists": "^3.0.0",
     "peer-book": "^0.1.1",
     "peer-id": "^0.6.6",
@@ -122,4 +123,4 @@
     "kumavis <[email protected]>",
     "nginnever <[email protected]>"
   ]
-}
+}

src/cli/commands/files/add.js

@@ -8,6 +8,7 @@ log.error = debug('cli:version:error')
 const bs58 = require('bs58')
 const fs = require('fs')
 const parallelLimit = require('run-parallel-limit')
+const async = require('async')
 const path = require('path')
 const glob = require('glob')
 
@@ -35,6 +36,50 @@ function checkPath (inPath, recursive) {
   return inPath
 }
 
+function daemonOn (res, inPath, ipfs) {
+  const files = []
+  if (res.length !== 0) {
+    const index = inPath.lastIndexOf('/')
+    async.eachLimit(res, 10, (element, callback) => {
+      if (fs.statSync(element).isDirectory()) {
+        callback()
+      } else {
+        const filePair = {
+          path: element.substring(index + 1, element.length),
+          content: fs.createReadStream(element)
+        }
+        files.push(filePair)
+        callback()
+      }
+    }, (err) => {
+      if (err) {
+        throw err
+      }
+      ipfs.add(files, (err, res) => {
+        if (err) {
+          throw err
+        }
+        res.forEach((goRes) => {
+          console.log('added', goRes.Hash, goRes.Name)
+        })
+      })
+    })
+  } else {
+    const filePair = {
+      path: inPath.substring(inPath.lastIndexOf('/') + 1, inPath.length),
+      content: fs.createReadStream(inPath)
+    }
+    files.push(filePair)
+    ipfs.add(files, (err, res) => {
+      if (err) {
+        throw err
+      }
+      console.log('added', res[0].Hash, res[0].Name)
+    })
+  }
+  return
+}
+
 module.exports = Command.extend({
   desc: 'Add a file to IPFS using the UnixFS data format',
 
@@ -59,36 +104,44 @@ module.exports = Command.extend({
         if (err) {
           throw err
         }
-        const i = ipfs.files.add()
-        var filePair
-        i.on('data', (file) => {
-          console.log('added', bs58.encode(file.multihash).toString(), file.path)
-        })
-        i.once('end', () => {
-          return
-        })
-        if (res.length !== 0) {
-          const index = inPath.lastIndexOf('/')
-          parallelLimit(res.map((element) => (callback) => {
-            if (!fs.statSync(element).isDirectory()) {
-              i.write({
-                path: element.substring(index + 1, element.length),
-                stream: fs.createReadStream(element)
-              })
-            }
-            callback()
-          }), 10, (err) => {
+        if (utils.isDaemonOn()) {
+          daemonOn(res, inPath, ipfs)
+        } else {
+          ipfs.files.add((err, i) => {
             if (err) {
               throw err
             }
-            i.end()
+            var filePair
+            i.on('data', (file) => {
+              console.log('added', bs58.encode(file.multihash).toString(), file.path)
+            })
+            i.once('end', () => {
+              return
+            })
+            if (res.length !== 0) {
+              const index = inPath.lastIndexOf('/')
+              parallelLimit(res.map((element) => (callback) => {
+                if (!fs.statSync(element).isDirectory()) {
+                  i.write({
+                    path: element.substring(index + 1, element.length),
+                    stream: fs.createReadStream(element)
+                  })
+                }
+                callback()
+              }), 10, (err) => {
+                if (err) {
+                  throw err
+                }
+                i.end()
+              })
+            } else {
+              rs = fs.createReadStream(inPath)
+              inPath = inPath.substring(inPath.lastIndexOf('/') + 1, inPath.length)
+              filePair = {path: inPath, stream: rs}
+              i.write(filePair)
+              i.end()
+            }
           })
-        } else {
-          rs = fs.createReadStream(inPath)
-          inPath = inPath.substring(inPath.lastIndexOf('/') + 1, inPath.length)
-          filePair = {path: inPath, stream: rs}
-          i.write(filePair)
-          i.end()
         }
       })
     })

src/cli/commands/files/cat.js

@@ -31,13 +31,11 @@ module.exports = Command.extend({
         })
         return
       }
-      ipfs.files.cat(path, (err, res) => {
+      ipfs.files.cat(path, (err, file) => {
         if (err) {
           throw (err)
         }
-        res.on('data', (data) => {
-          data.stream.pipe(process.stdout)
-        })
+        file.pipe(process.stdout)
       })
     })
   }

src/core/index.js

@@ -57,5 +57,7 @@ function IPFS (repoInstance) {
   this.object = object(this)
   this.libp2p = libp2p(this)
   this.files = files(this)
+  this.cat = files(this).cat // Alias for js-ipfs-api cat
+  this.add = files(this).add // Alias for js-ipfs-api add
   this.bitswap = bitswap(this)
 }

src/core/ipfs/files.js

@@ -3,10 +3,11 @@
 const Importer = require('ipfs-unixfs-engine').importer
 const Exporter = require('ipfs-unixfs-engine').exporter
 const UnixFS = require('ipfs-unixfs')
+const promisify = require('promisify-es6')
 
 module.exports = function files (self) {
   return {
-    add: (arr, callback) => {
+    add: promisify((arr, callback) => {
       if (typeof arr === 'function') {
         callback = arr
         arr = undefined
@@ -15,7 +16,7 @@ module.exports = function files (self) {
         callback = function noop () {}
       }
       if (arr === undefined) {
-        return new Importer(self._dagS)
+        callback(null, new Importer(self._dagS))
       }
 
       const i = new Importer(self._dagS)
@@ -34,8 +35,13 @@ module.exports = function files (self) {
       })
 
       i.end()
-    },
-    cat: (hash, callback) => {
+    }),
+
+    cat: promisify((hash, callback) => {
+      if (typeof hash === 'function') {
+        callback = hash
+        return callback('You must supply a multihash', null)
+      }
       self._dagS.get(hash, (err, fetchedNode) => {
         if (err) {
           return callback(err, null)
@@ -45,13 +51,16 @@ module.exports = function files (self) {
           callback('This dag node is a directory', null)
         } else {
           const exportStream = Exporter(hash, self._dagS)
-          callback(null, exportStream)
+          exportStream.once('data', (object) => {
+            callback(null, object.stream)
+          })
         }
       })
-    },
-    get: (hash, callback) => {
+    }),
+
+    get: promisify((hash, callback) => {
       var exportFile = Exporter(hash, self._dagS)
       callback(null, exportFile)
-    }
+    })
   }
 }

src/http-api/index.js

@@ -19,7 +19,6 @@ exports = module.exports = function HttpApi (repo) {
 
   this.start = (callback) => {
     if (typeof repo === 'string') {
-      console.log('here')
       repo = new IPFSRepo(repo, {stores: fsbs})
     }
 

src/http-api/resources/files.js

@@ -1,6 +1,9 @@
 'use strict'
 
 const bs58 = require('bs58')
+const multihash = require('multihashes')
+const ndjson = require('ndjson')
+const multipart = require('ipfs-multipart')
 const debug = require('debug')
 const log = debug('http-api:files')
 log.error = debug('http-api:files:error')
@@ -42,8 +45,70 @@ exports.cat = {
           Code: 0
         }).code(500)
       }
-      stream.on('data', (data) => {
-        return reply(data.stream)
+      return reply(stream)
+    })
+  }
+}
+
+exports.add = {
+  handler: (request, reply) => {
+    if (!request.payload) {
+      return reply('Array, Buffer, or String is required.').code(400).takeover()
+    }
+
+    const parser = multipart.reqParser(request.payload)
+
+    var filesParsed = false
+    var filesAdded = 0
+
+    var serialize = ndjson.serialize()
+    // hapi doesn't permit object streams: http://hapijs.com/api#replyerr-result
+    serialize._readableState.objectMode = false
+
+    request.server.app.ipfs.files.add((err, fileAdder) => {
+      if (err) {
+        return reply({
+          Message: err,
+          Code: 0
+        }).code(500)
+      }
+
+      fileAdder.on('data', (file) => {
+        serialize.write({
+          Name: file.path,
+          Hash: multihash.toB58String(file.multihash)
+        })
+        filesAdded++
+      })
+
+      fileAdder.on('end', () => {
+        if (filesAdded === 0 && filesParsed) {
+          return reply({
+            Message: 'Failed to add files.',
+            Code: 0
+          }).code(500)
+        } else {
+          serialize.end()
+          return reply(serialize)
+            .header('x-chunked-output', '1')
+            .header('content-type', 'application/json')
+        }
+      })
+
+      parser.on('file', (fileName, fileStream) => {
+        var filePair = {
+          path: fileName,
+          stream: fileStream
+        }
+        filesParsed = true
+        fileAdder.write(filePair)
+      })
+
+      parser.on('end', () => {
+        if (!filesParsed) {
+          return reply("File argument 'data' is required.").code(400).takeover()
+        }
+        fileAdder.end()
       })
     })
   }

src/http-api/routes/files.js

@@ -15,4 +15,16 @@ module.exports = (server) => {
       handler: resources.files.cat.handler
     }
   })
+
+  api.route({
+    method: '*',
+    path: '/api/v0/add',
+    config: {
+      payload: {
+        parse: false,
+        output: 'stream'
+      },
+      handler: resources.files.add.handler
+    }
+  })
 }

src/http-api/routes/index.js

@@ -8,7 +8,7 @@ module.exports = (server) => {
   require('./object')(server)
   // require('./repo')(server)
   require('./config')(server)
+  require('./files')(server)
   require('./swarm')(server)
   require('./bitswap')(server)
-  require('./files')(server)
 }

test/cli-tests/test-bitswap.js

@@ -12,7 +12,7 @@ const createTempNode = require('../utils/temp-node')
 const repoPath = require('./index').repoPath
 
 describe('bitswap', function () {
-  this.timeout(20000)
+  this.timeout(80000)
   const env = _.clone(process.env)
   env.IPFS_PATH = repoPath
 

test/cli-tests/test-files.js

@@ -12,6 +12,17 @@ describe('files', () => {
   env.IPFS_PATH = repoPath
 
   describe('api offline', () => {
+    it('add', (done) => {
+      nexpect.spawn('node', [process.cwd() + '/src/cli/bin.js', 'files', 'add', process.cwd() + '/test/test-data/node.json'], {env})
+        .run((err, stdout, exitcode) => {
+          expect(err).to.not.exist
+          expect(exitcode).to.equal(0)
+          expect(stdout[0])
+             .to.equal('added QmRRdjTN2PjyEPrW73GBxJNAZrstH5tCZzwHYFJpSTKkhe node.json')
+          done()
+        })
+    })
+
     it('cat', (done) => {
       nexpect.spawn('node', [process.cwd() + '/src/cli/bin.js', 'files', 'cat', 'QmT78zSuBmuS4z925WZfrqQ1qHaJ56DQaTfyMUF7F8ff5o'], {env})
         .run((err, stdout, exitcode) => {
@@ -40,6 +51,17 @@ describe('files', () => {
       })
     })
 
+    it('add', (done) => {
+      nexpect.spawn('node', [process.cwd() + '/src/cli/bin.js', 'files', 'add', process.cwd() + '/test/test-data/node.json'], {env})
+        .run((err, stdout, exitcode) => {
+          expect(err).to.not.exist
+          expect(exitcode).to.equal(0)
+          expect(stdout[0])
+             .to.equal('added QmRRdjTN2PjyEPrW73GBxJNAZrstH5tCZzwHYFJpSTKkhe node.json')
+          done()
+        })
+    })
+
     it('cat', (done) => {
       nexpect.spawn('node', [process.cwd() + '/src/cli/bin.js', 'files', 'cat', 'QmT78zSuBmuS4z925WZfrqQ1qHaJ56DQaTfyMUF7F8ff5o'], {env})
         .run((err, stdout, exitcode) => {

test/cli-tests/test-swarm.js

@@ -9,7 +9,7 @@ const repoPath = require('./index').repoPath
 const _ = require('lodash')
 
 describe('swarm', function () {
-  this.timeout(20000)
+  this.timeout(80000)
   const env = _.clone(process.env)
   env.IPFS_PATH = repoPath