feat: stream refs endpoints

Author: dirkmc

src/files-regular/refs-local-pull-stream.js

@@ -0,0 +1,26 @@
+'use strict'
+
+const cleanCID = require('../utils/clean-cid')
+const v = require('is-ipfs')
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
+const deferred = require('pull-defer')
+const moduleConfig = require('../utils/module-config')
+
+module.exports = (send) => {
+  send = moduleConfig(send)
+
+  return (opts) => {
+    opts = opts || {}
+
+    const p = deferred.source()
+
+    send({ path: 'refs/local', qs: opts }, (err, stream) => {
+      if (err) { return p.resolve(pull.error(err)) }
+
+      p.resolve(toPull.source(stream))
+    })
+
+    return p
+  }
+}

src/files-regular/refs-local-readable-stream.js

@@ -0,0 +1,22 @@
+'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 (opts) => {
+    opts = opts || {}
+
+    const pt = new Stream.PassThrough({ objectMode: true })
+
+    send({ path: 'refs/local', qs: opts }, (err, stream) => {
+      if (err) { return pt.destroy(err) }
+
+      pump(stream, pt)
+    })
+
+    return pt
+  }
+}

src/files-regular/refs-local.js

@@ -0,0 +1,25 @@
+'use strict'
+
+const IsIpfs = require('is-ipfs')
+const promisify = require('promisify-es6')
+const streamToValue = require('../utils/stream-to-value')
+const moduleConfig = require('../utils/module-config')
+const cleanCID = require('../utils/clean-cid')
+
+module.exports = (arg) => {
+  const send = moduleConfig(arg)
+
+  return promisify((opts, callback) => {
+    if (typeof (opts) === 'function') {
+      callback = opts
+      opts = {}
+    }
+
+    const request = {
+      path: 'refs/local',
+      qs: opts
+    }
+
+    send.andTransform(request, streamToValue, callback)
+  })
+}

src/files-regular/refs-pull-stream.js

@@ -1,21 +1,32 @@
 'use strict'
 
-const pullError = require('pull-stream/sources/error')
-const pullValues = require('pull-stream/sources/values')
-const pullDefer = require('pull-defer')
-const refs = require('./refs')
+const cleanCID = require('../utils/clean-cid')
+const v = require('is-ipfs')
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
+const deferred = require('pull-defer')
+const moduleConfig = require('../utils/module-config')
 
-module.exports = (arg) => {
-  const refsFn = refs(arg)
+module.exports = (send) => {
+  send = moduleConfig(send)
 
-  return (args, opts) => {
-    const p = pullDefer.source()
+  return (hash, opts) => {
+    opts = opts || {}
 
-    refsFn(args, opts, (err, res) => {
-      if (err) {
-        return p.resolve(pullError(err))
+    const p = deferred.source()
+
+    try {
+      hash = cleanCID(hash)
+    } catch (err) {
+      if (!v.ipfsPath(hash)) {
+        return p.end(err)
       }
-      p.resolve(pullValues(res))
+    }
+
+    send({ path: 'refs', args: hash, qs: opts }, (err, stream) => {
+      if (err) { return p.resolve(pull.error(err)) }
+
+      p.resolve(toPull.source(stream))
     })
 
     return p

src/files-regular/refs-readable-stream.js

@@ -1,20 +1,28 @@
 'use strict'
 
+const cleanCID = require('../utils/clean-cid')
+const v = require('is-ipfs')
 const Stream = require('readable-stream')
-const refs = require('./refs')
+const pump = require('pump')
 
-module.exports = (arg) => {
-  const refsFn = refs(arg)
+module.exports = (send) => {
+  return (hash, opts) => {
+    opts = opts || {}
 
-  return (args, opts) => {
     const pt = new Stream.PassThrough({ objectMode: true })
 
-    refsFn(args, opts, (err, res) => {
-      if (err) {
-        return pt.emit('error', err)
+    try {
+      hash = cleanCID(hash)
+    } catch (err) {
+      if (!v.ipfsPath(hash)) {
+        return pt.destroy(err)
       }
-      res.forEach((item) => pt.write(item))
-      pt.end()
+    }
+
+    send({ path: 'refs', args: hash, qs: opts }, (err, stream) => {
+      if (err) { return pt.destroy(err) }
+
+      pump(stream, pt)
     })
 
     return pt

src/files-regular/refs.js

@@ -40,19 +40,9 @@ module.exports = (arg) => {
     send.andTransform(request, valueOrStreamToValue, callback)
   })
 
-  refs.local = promisify((opts, callback) => {
-    if (typeof (opts) === 'function') {
-      callback = opts
-      opts = {}
-    }
-
-    const request = {
-      path: 'refs/local',
-      qs: opts
-    }
-
-    send.andTransform(request, valueOrStreamToValue, callback)
-  })
+  refs.local = require('./refs-local')(arg)
+  refs.localReadableStream = require('./refs-local-readable-stream')(arg)
+  refs.localPullStream = require('./refs-local-pull-stream')(arg)
 
   return refs
 }

test/sub-modules.spec.js

@@ -184,9 +184,11 @@ describe('submodules', () => {
     expect(filesRegular.lsReadableStream).to.be.a('function')
     expect(filesRegular.lsPullStream).to.be.a('function')
     expect(filesRegular.refs).to.be.a('function')
-    expect(filesRegular.refs.local).to.be.a('function')
     expect(filesRegular.refsReadableStream).to.be.a('function')
     expect(filesRegular.refsPullStream).to.be.a('function')
+    expect(filesRegular.refs.local).to.be.a('function')
+    expect(filesRegular.refs.localReadableStream).to.be.a('function')
+    expect(filesRegular.refs.localPullStream).to.be.a('function')
   })
 
   it('files MFS API', () => {