chore: wip

Author: achingbrain

src/core/components/files-regular/ls.js

@@ -4,7 +4,7 @@ const callbackify = require('callbackify')
 const all = require('async-iterator-all')
 
 module.exports = function (self) {
-  return callbackify(async function ls (ipfsPath, options) { // eslint-disable-line require-await
+  return callbackify.variadic(async function ls (ipfsPath, options) { // eslint-disable-line require-await
     return all(self._lsAsyncIterator(ipfsPath, options))
   })
 }

src/core/components/files-regular/refs-async-iterator.js

@@ -41,6 +41,7 @@ function getFullPath (ipfs, ipfsPath, options) {
   const path = normalizePath(ipfsPath)
   const pathComponents = path.split('/')
   const cid = pathComponents[0]
+
   if (!isIpfs.cid(cid)) {
     throw new Error(`Error resolving path '${path}': '${cid}' is not a valid CID`)
   }
@@ -89,43 +90,33 @@ function formatLink (srcCid, dstCid, linkName, format) {
 }
 
 // Do a depth first search of the DAG, starting from the given root cid
-async function * objectStream (ipfs, rootCid, maxDepth, isUnique) {
-  const uniques = new Set()
+async function * objectStream (ipfs, rootCid, maxDepth, uniqueOnly) {
+  const seen = new Set()
 
-  async function * traverseLevel (obj) {
-    const { node, depth } = obj
+  async function * traverseLevel (parent, depth) {
+    const nextLevelDepth = depth + 1
 
     // Check the depth
-    const nextLevelDepth = depth + 1
     if (nextLevelDepth > maxDepth) {
       return
     }
 
-    // If unique option is enabled, check if the CID has been seen before.
-    // Note we need to do this here rather than before adding to the stream
-    // so that the unique check happens in the order that items are examined
-    // in the DAG.
-    if (isUnique) {
-      if (uniques.has(node.cid.toString())) {
-        // Mark this object as a duplicate so we can filter it out later
-        obj.isDuplicate = true
-        return
-      }
-      uniques.add(node.cid.toString())
-    }
-
     // Get this object's links
     try {
-      // Add to the stream each link, parent and the new depth
-      for (const link of await getLinks(ipfs, node.cid)) {
-        const child = {
-          parent: node,
+      // Look at each link, parent and the new depth
+      for (const link of await getLinks(ipfs, parent.cid)) {
+
+        yield {
+          parent: parent,
           node: link,
-          depth: nextLevelDepth
+          isDuplicate: uniqueOnly && seen.has(link.cid.toString())
+        }
+
+        if (uniqueOnly) {
+          seen.add(link.cid.toString())
         }
 
-        yield child
-        yield * await traverseLevel(child)
+        yield * await traverseLevel(link, nextLevelDepth)
       }
     } catch (err) {
       if (err.code === ERR_NOT_FOUND) {
@@ -136,7 +127,7 @@ async function * objectStream (ipfs, rootCid, maxDepth, isUnique) {
     }
   }
 
-  yield * await traverseLevel({ node: { cid: rootCid }, depth: 0 })
+  yield * await traverseLevel({ cid: rootCid }, 0)
 }
 
 // Fetch a node from IPLD then get all its links

src/core/components/files-regular/refs-local.js

@@ -4,7 +4,7 @@ const callbackify = require('callbackify')
 const all = require('async-iterator-all')
 
 module.exports = function (self) {
-  return callbackify(async function refsLocal (ipfsPath, options) { // eslint-disable-line require-await
+  return callbackify.variadic(async function refsLocal (ipfsPath, options) { // eslint-disable-line require-await
     return all(self.refs._localAsyncIterator(ipfsPath, options))
   })
 }

src/core/components/files-regular/refs.js

@@ -1,26 +1,11 @@
 'use strict'
 
-const promisify = require('promisify-es6')
-const pull = require('pull-stream')
+const callbackify = require('callbackify')
+const all = require('async-iterator-all')
 
 module.exports = function (self) {
-  return promisify((ipfsPath, options, callback) => {
-    if (typeof options === 'function') {
-      callback = options
-      options = {}
-    }
-
-    options = options || {}
-
-    pull(
-      self.refsPullStream(ipfsPath, options),
-      pull.collect((err, values) => {
-        if (err) {
-          return callback(err)
-        }
-        callback(null, values)
-      })
-    )
+  return callbackify.variadic(async function refs (ipfsPath, options) {
+    return all(self._refsAsyncIterator(ipfsPath, options))
   })
 }
 

src/http/api/resources/files-regular.js

@@ -330,69 +330,69 @@ exports.refs = {
     const { ipfs } = request.server.app
     const { key } = request.pre.args
 
-    const recursive = request.query.recursive
-    const format = request.query.format
-    const edges = request.query.edges
-    const unique = request.query.unique
-    const maxDepth = request.query['max-depth']
-
-    const source = ipfs.refsPullStream(key, { recursive, format, edges, unique, maxDepth })
-    return sendRefsReplyStream(request, h, `refs for ${key}`, source)
+    const options = {
+      recursive: request.query.recursive,
+      format: request.query.format,
+      edges: request.query.edges,
+      unique: request.query.unique,
+      maxDepth: request.query['max-depth']
+    }
+
+    if (options.edges && options.format && options.format !== Format.default) {
+      throw new Error('Cannot set edges to true and also specify format')
+    }
+
+    return streamResponse(request, h, async (output) => {
+      for await (const ref of ipfs._refsAsyncIterator(key, options)) {
+        output.write(
+          JSON.stringify({
+            Ref: ref.ref,
+            Err: ref.err
+          }) + '\n'
+        )
+      }
+    })
   }
 }
 
 exports.refs.local = {
   // main route handler
   handler (request, h) {
     const { ipfs } = request.server.app
-    const source = ipfs.refs.localPullStream()
-    return sendRefsReplyStream(request, h, 'local refs', source)
+
+    return streamResponse(request, h, async (output) => {
+      for await (const ref of ipfs.refs._localAsyncIterator()) {
+        output.write(
+          JSON.stringify({
+            Ref: ref.ref,
+            Err: ref.err
+          }) + '\n'
+        )
+      }
+    })
   }
 }
 
-function sendRefsReplyStream (request, h, desc, source) {
-  const replyStream = pushable()
-  const aborter = abortable()
-
-  const stream = toStream.source(pull(
-    replyStream,
-    aborter,
-    ndjson.serialize()
-  ))
-
-  // const stream = toStream.source(replyStream.source)
-  // hapi is not very clever and throws if no
-  // - _read method
-  // - _readableState object
-  // are there :(
-  if (!stream._read) {
-    stream._read = () => {}
-    stream._readableState = {}
-    stream.unpipe = () => {}
-  }
+function streamResponse (request, h, fn) {
+  const output = new PassThrough()
+  const errorTrailer = 'X-Stream-Error'
 
-  pull(
-    source,
-    pull.drain(
-      (ref) => replyStream.push({ Ref: ref.ref, Err: ref.err }),
-      (err) => {
-        if (err) {
-          request.raw.res.addTrailers({
-            'X-Stream-Error': JSON.stringify({
-              Message: `Failed to get ${desc}: ${err.message || ''}`,
-              Code: 0
-            })
+  Promise.resolve()
+      .then(() => fn(output))
+      .catch(err => {
+        request.raw.res.addTrailers({
+          [errorTrailer]: JSON.stringify({
+            Message: err.message,
+            Code: 0
           })
-          return aborter.abort()
-        }
-
-        replyStream.end()
-      }
-    )
-  )
+        })
+      })
+      .finally(() => {
+        output.end()
+      })
 
-  return h.response(stream)
+  return h.response(output)
     .header('x-chunked-output', '1')
     .header('content-type', 'application/json')
-    .header('Trailer', 'X-Stream-Error')
+    .header('Trailer', errorTrailer)
 }