Merge pull request ipfs#1101 from ipfs/refactor/fetch-cat

refactor: use fetch for ipfs.cat

package.json

@@ -75,7 +75,7 @@
     "just-kebab-case": "^1.1.0",
     "just-map-keys": "^1.1.0",
     "kind-of": "^6.0.2",
-    "ky": "^0.13.0",
+    "ky": "^0.14.0",
     "ky-universal": "^0.3.0",
     "lru-cache": "^5.1.1",
     "merge-options": "^1.0.1",

src/cat.js

@@ -0,0 +1,34 @@
+'use strict'
+
+const CID = require('cids')
+const { Buffer } = require('buffer')
+const configure = require('./lib/configure')
+const toIterable = require('./lib/stream-to-iterable')
+
+module.exports = configure(({ ky }) => {
+  return (path, options) => (async function * () {
+    options = options || {}
+
+    const searchParams = new URLSearchParams(options.searchParams)
+
+    if (typeof path === 'string') {
+      searchParams.set('arg', path)
+    } else {
+      searchParams.set('arg', new CID(path).toString())
+    }
+
+    if (options.offset) searchParams.set('offset', options.offset)
+    if (options.length) searchParams.set('length', options.length)
+
+    const res = await ky.get('cat', {
+      timeout: options.timeout,
+      signal: options.signal,
+      headers: options.headers,
+      searchParams
+    })
+
+    for await (const chunk of toIterable(res.body)) {
+      yield Buffer.from(chunk)
+    }
+  })()
+})

src/files-regular/index.js

@@ -2,13 +2,14 @@
 
 const nodeify = require('promise-nodeify')
 const moduleConfig = require('../utils/module-config')
-const { collectify, pullify, streamify } = require('../lib/converters')
+const { concatify, collectify, pullify, streamify } = require('../lib/converters')
 
 module.exports = (arg) => {
   const send = moduleConfig(arg)
   const add = require('../add')(arg)
   const addFromFs = require('../add-from-fs')(arg)
   const addFromURL = require('../add-from-url')(arg)
+  const cat = require('../cat')(arg)
 
   return {
     add: (input, options, callback) => {
@@ -42,9 +43,15 @@ module.exports = (arg) => {
       return nodeify(collectify(add)(input, options), callback)
     },
     _addAsyncIterator: add,
-    cat: require('../files-regular/cat')(send),
-    catReadableStream: require('../files-regular/cat-readable-stream')(send),
-    catPullStream: require('../files-regular/cat-pull-stream')(send),
+    cat: (path, options, callback) => {
+      if (typeof options === 'function') {
+        callback = options
+        options = {}
+      }
+      return nodeify(concatify(cat)(path, options), callback)
+    },
+    catReadableStream: streamify.readable(cat),
+    catPullStream: pullify.source(cat),
     get: require('../files-regular/get')(send),
     getReadableStream: require('../files-regular/get-readable-stream')(send),
     getPullStream: require('../files-regular/get-pull-stream')(send),

src/lib/converters.js

@@ -3,14 +3,18 @@
 const toPull = require('async-iterator-to-pull-stream')
 const all = require('async-iterator-all')
 const toStream = require('it-to-stream')
+const { Buffer } = require('buffer')
 
 exports.collectify = fn => (...args) => all(fn(...args))
 
+exports.concatify = fn => async (...args) => Buffer.concat(await all(fn(...args)))
+
 exports.pullify = {
   source: fn => (...args) => toPull(fn(...args)),
   transform: fn => (...args) => toPull.transform(source => fn(source, ...args))
 }
 
 exports.streamify = {
+  readable: fn => (...args) => toStream(fn(...args), { objectMode: true }),
   transform: fn => (...args) => toStream.transform(source => fn(source, ...args), { objectMode: true })
 }

src/lib/error-handler.js

@@ -2,13 +2,25 @@
 
 const { HTTPError } = require('ky-universal')
 const log = require('debug')('ipfs-http-client:lib:error-handler')
+const { isNode, isElectron } = require('ipfs-utils/src/env')
 
 function isJsonResponse (res) {
   return (res.headers.get('Content-Type') || '').startsWith('application/json')
 }
 
 module.exports = async function errorHandler (input, options, response) {
-  if (response.ok) return
+  if (response.ok) {
+    // FIXME: remove when fixed https://github.com/sindresorhus/ky-universal/issues/8
+    //
+    // ky clones the response for each handler. In Node.js the response body is
+    // piped to 2 PassThroughs, one becomes the real body and the other is used
+    // in the clone.
+    //
+    // If the body in the clone is not consumed or destroyed the highwater mark
+    // will be reached (for large payloads) and stop the real body from flowing.
+    if (isNode || isElectron) response.body.destroy()
+    return
+  }
 
   let msg