Implementing the new interfaces

src/core/components/files.js

@@ -9,16 +9,104 @@ const sort = require('pull-sort')
 const pushable = require('pull-pushable')
 const toStream = require('pull-stream-to-stream')
 const toPull = require('stream-to-pull-stream')
+const deferred = require('pull-defer')
 const waterfall = require('async/waterfall')
 const isStream = require('is-stream')
-const Duplex = require('stream').Duplex
+const Duplex = require('readable-stream').Duplex
 const CID = require('cids')
 const toB58String = require('multihashes').toB58String
 
+function noop () {}
+
+function prepareFile (self, opts, file, callback) {
+  opts = opts || {}
+
+  waterfall([
+    (cb) => self.object.get(file.multihash, cb),
+    (node, cb) => {
+      let cid = new CID(node.multihash)
+
+      if (opts['cid-version'] === 1) {
+        cid = cid.toV1()
+      }
+
+      const b58Hash = cid.toBaseEncodedString()
+
+      cb(null, {
+        path: file.path || b58Hash,
+        hash: b58Hash,
+        size: node.size
+      })
+    }
+  ], callback)
+}
+
+function normalizeContent (content) {
+  if (!Array.isArray(content)) {
+    content = [content]
+  }
+
+  return content.map((data) => {
+    // Buffer input
+    if (Buffer.isBuffer(data)) {
+      data = { path: '', content: pull.values([data]) }
+    }
+
+    // Readable stream input
+    if (isStream.readable(data)) {
+      data = { path: '', content: toPull.source(data) }
+    }
+
+    if (data && data.content && typeof data.content !== 'function') {
+      if (Buffer.isBuffer(data.content)) {
+        data.content = pull.values([data.content])
+      }
+
+      if (isStream.readable(data.content)) {
+        data.content = toPull.source(data.content)
+      }
+    }
+
+    return data
+  })
+}
+
+class AddHelper extends Duplex {
+  constructor (pullStream, push, options) {
+    super(Object.assign({ objectMode: true }, options))
+    this._pullStream = pullStream
+    this._pushable = push
+    this._waitingPullFlush = []
+  }
+
+  _read () {
+    this._pullStream(null, (end, data) => {
+      while (this._waitingPullFlush.length) {
+        const cb = this._waitingPullFlush.shift()
+        cb()
+      }
+      if (end) {
+        if (end instanceof Error) {
+          this.emit('error', end)
+        }
+      } else {
+        this.push(data)
+      }
+    })
+  }
+
+  _write (chunk, encoding, callback) {
+    this._waitingPullFlush.push(callback)
+    this._pushable.push(chunk)
+  }
+}
+
 module.exports = function files (self) {
-  const createAddPullStream = (options) => {
+  function _addPullStream (options) {
     const opts = Object.assign({}, {
-      shardSplitThreshold: self._options.EXPERIMENTAL.sharding ? 1000 : Infinity
+      shardSplitThreshold: self._options.EXPERIMENTAL.sharding
+        ? 1000
+        : Infinity
     }, options)
 
     let total = 0
@@ -37,29 +125,32 @@ module.exports = function files (self) {
     )
   }
 
-  return {
-    createAddStream: (options, callback) => {
-      if (typeof options === 'function') {
-        callback = options
-        options = undefined
-      }
-
-      const addPullStream = createAddPullStream(options)
-      const p = pushable()
-      const s = pull(
-        p,
-        addPullStream
-      )
+  function _catPullStream (ipfsPath) {
+    if (typeof ipfsPath === 'function') {
+      throw new Error('You must supply an ipfsPath')
+    }
 
-      const retStream = new AddStreamDuplex(s, p)
+    const d = deferred.source()
 
-      retStream.once('finish', () => p.end())
+    pull(
+      exporter(ipfsPath, self._ipldResolver),
+      pull.collect((err, files) => {
+        if (err) { d.end(err) }
+        if (!files || !files.length) {
+          return d.end(new Error('No such file'))
+        }
 
-      callback(null, retStream)
-    },
+        const content = files[files.length - 1].content
+        console.log('do I resolve?', content)
+        d.resolve(content)
+        // toStream.source(content).pipe(concat((data) => callback(null, data)))
+      })
+    )
 
-    createAddPullStream: createAddPullStream,
+    return d
+  }
 
+  return {
     add: promisify((data, options, callback) => {
       if (typeof options === 'function') {
         callback = options
@@ -76,7 +167,7 @@ module.exports = function files (self) {
 
       pull(
         pull.values([data]),
-        createAddPullStream(options),
+        _addPullStream(options),
         sort((a, b) => {
           if (a.path < b.path) return 1
           if (a.path > b.path) return -1
@@ -86,23 +177,40 @@ module.exports = function files (self) {
       )
     }),
 
-    cat: promisify((ipfsPath, callback) => {
-      if (typeof ipfsPath === 'function') {
-        return callback(new Error('You must supply an ipfsPath'))
-      }
+    addReadableStream: (options) => {
+      options = options || {}
+
+      const p = pushable()
+      const s = pull(
+        p,
+        _addPullStream(options)
+      )
 
+      const retStream = new AddHelper(s, p)
+
+      retStream.once('finish', () => p.end())
+
+      return retStream
+    },
+
+    addPullStream: _addPullStream,
+
+    cat: promisify((ipfsPath, callback) => {
+      const p = _catPullStream(ipfsPath)
       pull(
-        exporter(ipfsPath, self._ipldResolver),
-        pull.collect((err, files) => {
-          if (err) {
-            return callback(err)
-          }
-          if (!files || !files.length) return callback(new Error('No such file'))
-          callback(null, toStream.source(files[files.length - 1].content))
-        })
+        p,
+        pull.concat(callback)
       )
     }),
 
+    catReadableStream: (ipfsPath) => {
+      const p = _catPullStream(ipfsPath)
+
+      return toStream.source(p)
+    },
+
+    catPullStream: _catPullStream,
+
     get: promisify((ipfsPath, callback) => {
       callback(null, toStream.source(pull(
         exporter(ipfsPath, self._ipldResolver),
@@ -117,112 +225,30 @@ module.exports = function files (self) {
       )))
     }),
 
-    getPull: promisify((ipfsPath, callback) => {
-      callback(null, exporter(ipfsPath, self._ipldResolver))
-    }),
+    getReadableStream: (ipfsPath) => {
+      // TODO
+      return exporter(ipfsPath, self._ipldResolver)
+    },
 
-    immutableLs: promisify((ipfsPath, callback) => {
+    getPullStream: (ipfsPath) => {
+      return exporter(ipfsPath, self._ipldResolver)
+    },
+
+    lsImmutable: promisify((ipfsPath, callback) => {
       pull(
         self.files.immutableLsPullStream(ipfsPath),
         pull.collect(callback))
     }),
 
-    immutableLsPullStream: (ipfsPath) => {
+    lsReadableStreamImmutable: (ipfsPath) => {
+      // TODO
+    },
+
+    lsPullStreamImmutable: (ipfsPath) => {
       return pull(
         exporter(ipfsPath, self._ipldResolver, { maxDepth: 1 }),
         pull.filter((node) => node.depth === 1),
         pull.map((node) => Object.assign({}, node, { hash: toB58String(node.hash) })))
     }
   }
 }
-
-function prepareFile (self, opts, file, callback) {
-  opts = opts || {}
-
-  waterfall([
-    (cb) => self.object.get(file.multihash, cb),
-    (node, cb) => {
-      let cid = new CID(node.multihash)
-
-      if (opts['cid-version'] === 1) {
-        cid = cid.toV1()
-      }
-
-      const b58Hash = cid.toBaseEncodedString()
-
-      cb(null, {
-        path: file.path || b58Hash,
-        hash: b58Hash,
-        size: node.size
-      })
-    }
-  ], callback)
-}
-
-function normalizeContent (content) {
-  if (!Array.isArray(content)) {
-    content = [content]
-  }
-
-  return content.map((data) => {
-    // Buffer input
-    if (Buffer.isBuffer(data)) {
-      data = {
-        path: '',
-        content: pull.values([data])
-      }
-    }
-
-    // Readable stream input
-    if (isStream.readable(data)) {
-      data = {
-        path: '',
-        content: toPull.source(data)
-      }
-    }
-
-    if (data && data.content && typeof data.content !== 'function') {
-      if (Buffer.isBuffer(data.content)) {
-        data.content = pull.values([data.content])
-      }
-
-      if (isStream.readable(data.content)) {
-        data.content = toPull.source(data.content)
-      }
-    }
-
-    return data
-  })
-}
-
-function noop () {}
-
-class AddStreamDuplex extends Duplex {
-  constructor (pullStream, push, options) {
-    super(Object.assign({ objectMode: true }, options))
-    this._pullStream = pullStream
-    this._pushable = push
-    this._waitingPullFlush = []
-  }
-
-  _read () {
-    this._pullStream(null, (end, data) => {
-      while (this._waitingPullFlush.length) {
-        const cb = this._waitingPullFlush.shift()
-        cb()
-      }
-      if (end) {
-        if (end instanceof Error) {
-          this.emit('error', end)
-        }
-      } else {
-        this.push(data)
-      }
-    })
-  }
-
-  _write (chunk, encoding, callback) {
-    this._waitingPullFlush.push(callback)
-    this._pushable.push(chunk)
-  }
-}

src/core/index.js

@@ -113,7 +113,9 @@ class IPFS extends EventEmitter {
     this.state = require('./state')(this)
 
     // ipfs.ls
-    this.ls = this.files.immutableLs
+    this.ls = this.files.lsImmutable
+    this.lsReadableStream = this.files.lsReadableStreamImmutable
+    this.lsPullStream = this.files.lsPullStreamImmutable
 
     boot(this)
   }