feat: Implementing the new interfaces (#619)

README.md

@@ -8,10 +8,8 @@
   <a href="http://ipn.io"><img src="https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square" /></a>
   <a href="http://ipfs.io/"><img src="https://img.shields.io/badge/project-IPFS-blue.svg?style=flat-square" /></a>
   <a href="http://webchat.freenode.net/?channels=%23ipfs"><img src="https://img.shields.io/badge/freenode-%23ipfs-blue.svg?style=flat-square" /></a>
-  <br>
   <a href="https://waffle.io/ipfs/js-ipfs"><img src="https://img.shields.io/badge/pm-waffle-blue.svg?style=flat-square" /></a>
   <a href="https://github.com/ipfs/interface-ipfs-core"><img src="https://img.shields.io/badge/interface--ipfs--core-API%20Docs-blue.svg"></a>
-  <a href="https://github.com/ipfs/interface-ipfs-core/issues/55"><img src="https://img.shields.io/badge/interface--ipfs--core-Updates-blue.svg"></a>
 </p>
 
 <p align="center">

package.json

@@ -19,7 +19,7 @@
     "release-minor": "aegir release --type minor ",
     "release-major": "aegir release --type major ",
     "coverage": "aegir coverage --timeout 100000",
-    "coverage-publish": "aegir coverage --provider coveralls k--timeout 100000"
+    "coverage-publish": "aegir coverage --provider coveralls --timeout 100000"
   },
   "dependencies": {
     "async": "^2.6.0",
@@ -45,9 +45,13 @@
     "peer-info": "~0.11.1",
     "promisify-es6": "^1.0.3",
     "pump": "^1.0.3",
+    "pull": "^2.1.1",
+    "pull-defer": "^0.2.2",
+    "pull-pushable": "^2.1.1",
     "qs": "^6.5.1",
     "readable-stream": "^2.3.3",
     "stream-http": "^2.7.2",
+    "stream-to-pull-stream": "^1.7.2",
     "streamifier": "^0.1.1",
     "tar-stream": "^1.5.5"
   },
@@ -65,7 +69,7 @@
     "dirty-chai": "^2.0.1",
     "eslint-plugin-react": "^7.4.0",
     "gulp": "^3.9.1",
-    "interface-ipfs-core": "~0.35.0",
+    "interface-ipfs-core": "~0.36.4",
     "hapi": "^16.6.2",
     "ipfsd-ctl": "~0.24.1",
     "pre-commit": "^1.2.2",

src/files/add-pull-stream.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const addCmd = require('./add.js')
+const pull = require('pull-stream')
+const pushable = require('pull-pushable')
+
+module.exports = (send) => {
+  const add = addCmd(send)
+
+  return (options) => {
+    options = options || {}
+
+    const source = pushable()
+    const sink = pull.collect((err, tuples) => {
+      if (err) { return source.end(err) }
+
+      add(tuples, options, (err, filesAdded) => {
+        if (err) { return source.end(err) }
+
+        filesAdded.forEach((file) => source.push(file))
+        source.end()
+      })
+    })
+
+    return {
+      sink: sink,
+      source: source
+    }
+  }
+}

src/files/add-readable-stream.js

@@ -0,0 +1,31 @@
+'use strict'
+
+const addCmd = require('./add.js')
+const Duplex = require('readable-stream').Duplex
+
+module.exports = (send) => {
+  const add = addCmd(send)
+
+  return (options) => {
+    options = options || {}
+
+    const tuples = []
+
+    const ds = new Duplex({ objectMode: true })
+    ds._read = (n) => {}
+
+    ds._write = (file, enc, next) => {
+      tuples.push(file)
+      next()
+    }
+
+    ds.end = () => add(tuples, options, (err, res) => {
+      if (err) { return ds.emit('error', err) }
+
+      res.forEach((tuple) => ds.push(tuple))
+      ds.push(null)
+    })
+
+    return ds
+  }
+}

src/files/cat-pull-stream.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const cleanCID = require('../utils/clean-cid')
+const v = require('is-ipfs')
+const toPull = require('stream-to-pull-stream')
+const deferred = require('pull-defer')
+
+module.exports = (send) => {
+  return (hash, opts) => {
+    opts = opts || {}
+
+    const p = deferred.source()
+
+    try {
+      hash = cleanCID(hash)
+    } catch (err) {
+      if (!v.ipfsPath(hash)) {
+        return p.end(err)
+      }
+    }
+
+    send({ path: 'cat', args: hash, buffer: opts.buffer }, (err, stream) => {
+      if (err) { return p.end(err) }
+
+      p.resolve(toPull(stream))
+    })
+
+    return p
+  }
+}

src/files/cat-readable-stream.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const cleanCID = require('../utils/clean-cid')
+const v = require('is-ipfs')
+const Stream = require('readable-stream')
+const pump = require('pump')
+
+module.exports = (send) => {
+  return (hash, opts) => {
+    opts = opts || {}
+
+    const pt = new Stream.PassThrough()
+
+    try {
+      hash = cleanCID(hash)
+    } catch (err) {
+      if (!v.ipfsPath(hash)) {
+        return pt.destroy(err)
+      }
+    }
+
+    send({ path: 'cat', args: hash, buffer: opts.buffer }, (err, stream) => {
+      if (err) { return pt.destroy(err) }
+
+      pump(stream, pt)
+    })
+
+    return pt
+  }
+}

src/files/cat.js

@@ -3,6 +3,7 @@
 const promisify = require('promisify-es6')
 const cleanCID = require('../utils/clean-cid')
 const v = require('is-ipfs')
+const bl = require('bl')
 
 module.exports = (send) => {
   return promisify((hash, opts, callback) => {
@@ -19,10 +20,14 @@ module.exports = (send) => {
       }
     }
 
-    send({
-      path: 'cat',
-      args: hash,
-      buffer: opts.buffer
-    }, callback)
+    send({ path: 'cat', args: hash, buffer: opts.buffer }, (err, stream) => {
+      if (err) { return callback(err) }
+
+      stream.pipe(bl((err, data) => {
+        if (err) { return callback(err) }
+
+        callback(null, data)
+      }))
+    })
   })
 }

src/files/get-pull-stream.js

@@ -0,0 +1,44 @@
+'use strict'
+
+const cleanCID = require('../utils/clean-cid')
+const TarStreamToObjects = require('../utils/tar-stream-to-objects')
+const v = require('is-ipfs')
+const through = require('through2')
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
+const deferred = require('pull-defer')
+
+module.exports = (send) => {
+  return (path, opts) => {
+    opts = opts || {}
+
+    const p = deferred.source()
+
+    try {
+      path = cleanCID(path)
+    } catch (err) {
+      if (!v.ipfsPath(path)) {
+        return p.end(err)
+      }
+    }
+
+    const request = { path: 'get', args: path, qs: opts }
+
+    // Convert the response stream to TarStream objects
+    send.andTransform(request, TarStreamToObjects, (err, stream) => {
+      if (err) { return p.end(err) }
+
+      const files = []
+      stream.pipe(through.obj((file, enc, next) => {
+        if (file.content) {
+          files.push({ path: file.path, content: toPull(file.content) })
+        } else {
+          files.push(file)
+        }
+        next()
+      }, () => p.resolve(pull.values(files))))
+    })
+
+    return p
+  }
+}

src/files/get-readable-stream.js

@@ -0,0 +1,34 @@
+'use strict'
+
+const cleanCID = require('../utils/clean-cid')
+const TarStreamToObjects = require('../utils/tar-stream-to-objects')
+const v = require('is-ipfs')
+const Stream = require('readable-stream')
+const pump = require('pump')
+
+module.exports = (send) => {
+  return (path, opts) => {
+    opts = opts || {}
+
+    const pt = new Stream.PassThrough({objectMode: true})
+
+    try {
+      path = cleanCID(path)
+    } catch (err) {
+      if (!v.ipfsPath(path)) {
+        return pt.destroy(err)
+      }
+    }
+
+    const request = { path: 'get', args: path, qs: opts }
+
+    // Convert the response stream to TarStream objects
+    send.andTransform(request, TarStreamToObjects, (err, stream) => {
+      if (err) { return pt.destroy(err) }
+
+      pump(stream, pt)
+    })
+
+    return pt
+  }
+}

src/files/get.js

@@ -3,6 +3,8 @@
 const promisify = require('promisify-es6')
 const cleanCID = require('../utils/clean-cid')
 const TarStreamToObjects = require('../utils/tar-stream-to-objects')
+const concat = require('concat-stream')
+const through = require('through2')
 const v = require('is-ipfs')
 
 module.exports = (send) => {
@@ -14,8 +16,7 @@ module.exports = (send) => {
 
     // opts is the real callback --
     // 'callback' is being injected by promisify
-    if (typeof opts === 'function' &&
-        typeof callback === 'function') {
+    if (typeof opts === 'function' && typeof callback === 'function') {
       callback = opts
       opts = {}
     }
@@ -28,13 +29,23 @@ module.exports = (send) => {
       }
     }
 
-    const request = {
-      path: 'get',
-      args: path,
-      qs: opts
-    }
+    const request = { path: 'get', args: path, qs: opts }
 
     // Convert the response stream to TarStream objects
-    send.andTransform(request, TarStreamToObjects, callback)
+    send.andTransform(request, TarStreamToObjects, (err, stream) => {
+      if (err) { return callback(err) }
+
+      const files = []
+      stream.pipe(through.obj((file, enc, next) => {
+        if (file.content) {
+          file.content.pipe(concat((content) => {
+            files.push({ path: file.path, content: content })
+          }))
+        } else {
+          files.push(file)
+        }
+        next()
+      }, () => callback(null, files)))
+    })
   })
 }

src/files/index.js

@@ -7,14 +7,21 @@ module.exports = (arg) => {
 
   return {
     add: require('./add')(send),
-    createAddStream: require('./create-add-stream')(send),
-    get: require('./get')(send),
+    addReadableStream: require('./add-readable-stream')(send),
+    addPullStream: require('./add-pull-stream')(send),
     cat: require('./cat')(send),
+    catReadableStream: require('./cat-readable-stream')(send),
+    catPullStream: require('./cat-pull-stream')(send),
+    get: require('./get')(send),
+    getReadableStream: require('./get-readable-stream')(send),
+    getPullStream: require('./get-pull-stream')(send),
+
+    // Specific to MFS (for now)
     cp: require('./cp')(send),
-    ls: require('./ls')(send),
     mkdir: require('./mkdir')(send),
     stat: require('./stat')(send),
     rm: require('./rm')(send),
+    ls: require('./ls')(send),
     read: require('./read')(send),
     write: require('./write')(send),
     mv: require('./mv')(send)