refactor: remove Node.js and pull streams (#66)

* refactor: remove Node.js and pull streams

Also changes `stat` and `ls` to return CID's not strings from core and removes the `long` option from `ls` in core (is applied by CLI/HTTP API).

* fix: cid formatting in files.stat CLI and HTTP API

* fix: revert to older dependencies 😢

* chore: eslint-disable-line require-await

* fix: stream is objectMode

* chore: rebase and fix tests

Co-authored-by: Alex Potsides <alex@achingbrain.net>

Author: Alan Shaw

package.json

@@ -54,7 +54,6 @@
     "ipfs-block-service": "~0.16.0",
     "ipfs-repo": "^0.30.1",
     "ipld": "~0.25.0",
-    "it-all": "^1.0.1",
     "memdown": "^5.1.0",
     "nyc": "^15.0.0",
     "sinon": "^8.0.4",
@@ -75,16 +74,16 @@
     "ipfs-unixfs": "^0.3.0",
     "ipfs-unixfs-exporter": "^0.41.0",
     "ipfs-unixfs-importer": "^0.44.0",
-    "ipfs-utils": "^0.4.2",
+    "ipfs-utils": "^0.7.0",
     "ipld-dag-pb": "^0.18.0",
+    "it-all": "^1.0.1",
     "it-last": "^1.0.1",
+    "it-to-stream": "^0.1.1",
     "it-pipe": "^1.0.1",
     "joi-browser": "^13.4.0",
     "mortice": "^2.0.0",
     "multicodec": "^1.0.0",
-    "multihashes": "^0.4.14",
-    "once": "^1.4.0",
-    "pull-stream": "^3.6.9"
+    "multihashes": "^0.4.14"
   },
   "contributors": [
     "Alan Shaw <alan.shaw@protocol.ai>",

src/cli/flush.js

@@ -11,7 +11,6 @@ module.exports = {
 
   builder: {
     'cid-base': {
-      default: 'base58btc',
       describe: 'CID base to use.'
     }
   },
@@ -28,7 +27,7 @@ module.exports = {
       const ipfs = await getIpfs()
       let cid = await ipfs.files.flush(path || FILE_SEPARATOR, {})
 
-      if (cidBase !== 'base58btc' && cid.version === 0) {
+      if (cidBase && cidBase !== 'base58btc' && cid.version === 0) {
         cid = cid.toV1()
       }
 

src/cli/ls.js

@@ -1,8 +1,6 @@
 'use strict'
 
-const pull = require('pull-stream/pull')
-const onEnd = require('pull-stream/sinks/on-end')
-const through = require('pull-stream/throughs/through')
+const all = require('it-all')
 const {
   asBoolean
 } = require('./utils')
@@ -33,7 +31,6 @@ module.exports = {
       describe: 'Sort entries by name'
     },
     'cid-base': {
-      default: 'base58btc',
       describe: 'CID base to use.'
     }
   },
@@ -50,53 +47,30 @@ module.exports = {
 
     argv.resolve((async () => {
       const ipfs = await getIpfs()
-      return new Promise((resolve, reject) => {
-        if (sort) {
-          ipfs.files.ls(path || FILE_SEPARATOR)
-            .then(files => {
-              // https://github.com/ipfs/go-ipfs/issues/5181
-              if (sort) {
-                files = files.sort((a, b) => {
-                  return a.name.localeCompare(b.name)
-                })
-              }
 
-              if (long) {
-                files.forEach(file => {
-                  print(`${formatMode(file.mode, file.type === 1)}\t${formatMtime(file.mtime)}\t${file.name}\t${file.hash}\t${file.size}`)
-                })
-              } else {
-                files.forEach(link => print(link.name))
-              }
+      const printListing = file => {
+        if (long) {
+          print(`${formatMode(file.mode, file.type === 1)}\t${formatMtime(file.mtime)}\t${file.name}\t${file.cid.toString(cidBase)}\t${file.size}`)
+        } else {
+          print(file.name)
+        }
+      }
 
-              resolve()
-            })
-            .catch(reject)
+      // https://github.com/ipfs/go-ipfs/issues/5181
+      if (sort) {
+        let files = await all(ipfs.files.ls(path || FILE_SEPARATOR))
 
-          return
-        }
+        files = files.sort((a, b) => {
+          return a.name.localeCompare(b.name)
+        })
 
-        pull(
-          ipfs.files.lsPullStream(path, {
-            long,
-            cidBase
-          }),
-          through(file => {
-            if (long) {
-              print(`${formatMode(file.mode, file.type === 1)}\t${formatMtime(file.mtime)}\t${file.name}\t${file.hash}\t${file.size}`)
-            } else {
-              print(file.name)
-            }
-          }),
-          onEnd((error) => {
-            if (error) {
-              return reject(error)
-            }
+        files.forEach(printListing)
+        return
+      }
 
-            resolve()
-          })
-        )
-      })
+      for await (const file of ipfs.files.ls(path || FILE_SEPARATOR)) {
+        printListing(file)
+      }
     })())
   }
 }

src/cli/read.js

@@ -1,9 +1,5 @@
 'use strict'
 
-const pull = require('pull-stream/pull')
-const through = require('pull-stream/throughs/through')
-const onEnd = require('pull-stream/sinks/on-end')
-
 module.exports = {
   command: 'read <path>',
 
@@ -34,24 +30,12 @@ module.exports = {
     argv.resolve((async () => {
       const ipfs = await getIpfs()
 
-      return new Promise((resolve, reject) => {
-        pull(
-          ipfs.files.readPullStream(path, {
-            offset,
-            length
-          }),
-          through(buffer => {
-            print(buffer, false)
-          }),
-          onEnd((error) => {
-            if (error) {
-              return reject(error)
-            }
-
-            resolve()
-          })
-        )
-      })
+      for await (const buffer of ipfs.files.read(path, {
+        offset,
+        length
+      })) {
+        print(buffer, false)
+      }
     })())
   }
 }

src/cli/stat.js

@@ -46,7 +46,6 @@ Mtime: <mtime>`,
       describe: 'Compute the amount of the dag that is local, and if possible the total size'
     },
     'cid-base': {
-      default: 'base58btc',
       describe: 'CID base to use.'
     }
   },
@@ -59,7 +58,8 @@ Mtime: <mtime>`,
       format,
       hash,
       size,
-      withLocal
+      withLocal,
+      cidBase
     } = argv
 
     argv.resolve((async () => {
@@ -70,15 +70,15 @@ Mtime: <mtime>`,
       })
         .then((stats) => {
           if (hash) {
-            return print(stats.hash)
+            return print(stats.cid.toString(cidBase))
           }
 
           if (size) {
             return print(stats.size)
           }
 
           print(format
-            .replace('<hash>', stats.hash)
+            .replace('<hash>', stats.cid.toString(cidBase))
             .replace('<size>', stats.size)
             .replace('<cumulsize>', stats.cumulativeSize)
             .replace('<childs>', stats.blocks)

src/core/write.js

@@ -160,7 +160,7 @@ const write = async (context, source, destination, options) => {
     limitAsyncStreamBytes(source, options.length)
   )
 
-  const content = countBytesStreamed(catAsyncInterators(sources), (bytesWritten) => {
+  const content = countBytesStreamed(catAsyncIterators(sources), (bytesWritten) => {
     if (destination.unixfs && !options.truncate) {
       // if we've done reading from the new source and we are not going
       // to truncate the file, add the end of the existing file to the output
@@ -254,11 +254,9 @@ const asyncZeroes = (count, chunkSize = MAX_CHUNK_SIZE) => {
   return limitAsyncStreamBytes(stream, count)
 }
 
-const catAsyncInterators = async function * (sources) {
+const catAsyncIterators = async function * (sources) { // eslint-disable-line require-await
   for (let i = 0; i < sources.length; i++) {
-    for await (const buf of sources[i]()) {
-      yield buf
-    }
+    yield * sources[i]()
   }
 }
 

src/http/flush.js

@@ -20,7 +20,7 @@ const mfsFlush = {
 
     let cid = await ipfs.files.flush(arg || FILE_SEPARATOR, {})
 
-    if (cidBase !== 'base58btc' && cid.version === 0) {
+    if (cidBase && cidBase !== 'base58btc' && cid.version === 0) {
       cid = cid.toV1()
     }
 
@@ -36,7 +36,7 @@ const mfsFlush = {
       },
       query: Joi.object().keys({
         arg: Joi.string(),
-        cidBase: Joi.string().default('base58btc')
+        cidBase: Joi.string()
       })
     }
   }

src/http/ls.js

@@ -4,13 +4,17 @@ const Joi = require('@hapi/joi')
 const {
   PassThrough
 } = require('stream')
+const toStream = require('it-to-stream')
+const all = require('it-all')
+
+const mapEntry = (entry, options) => {
+  options = options || {}
 
-const mapEntry = (entry) => {
   const output = {
     Name: entry.name,
-    Type: entry.type,
-    Size: entry.size,
-    Hash: entry.hash,
+    Type: options.long ? entry.type : 0,
+    Size: options.long ? entry.size || 0 : 0,
+    Hash: entry.cid.toString(options.cidBase),
     Mode: entry.mode.toString(8).padStart(4, '0')
   }
 
@@ -41,21 +45,18 @@ const mfsLs = {
 
     if (stream) {
       const responseStream = await new Promise((resolve, reject) => {
-        const readableStream = ipfs.files.lsReadableStream(arg, {
-          long,
-          cidBase
-        })
+        const readableStream = toStream.readable(ipfs.files.ls(arg), { objectMode: true })
 
         const passThrough = new PassThrough()
 
         readableStream.on('data', (entry) => {
           resolve(passThrough)
-          passThrough.write(JSON.stringify(mapEntry(entry)) + '\n')
+          passThrough.write(JSON.stringify(mapEntry(entry, { cidBase, long })) + '\n')
         })
 
         readableStream.once('end', (entry) => {
           resolve(passThrough)
-          passThrough.end(entry ? JSON.stringify(mapEntry(entry)) + '\n' : undefined)
+          passThrough.end(entry ? JSON.stringify(mapEntry(entry, { cidBase, long })) + '\n' : undefined)
         })
 
         readableStream.once('error', (err) => {
@@ -67,13 +68,10 @@ const mfsLs = {
       return h.response(responseStream).header('X-Stream-Output', '1')
     }
 
-    const files = await ipfs.files.ls(arg, {
-      long,
-      cidBase
-    })
+    const files = await all(ipfs.files.ls(arg))
 
     return h.response({
-      Entries: files.map(mapEntry)
+      Entries: files.map(entry => mapEntry(entry, { cidBase, long }))
     })
   },
   options: {
@@ -85,7 +83,7 @@ const mfsLs = {
       query: Joi.object().keys({
         arg: Joi.string().default('/'),
         long: Joi.boolean().default(false),
-        cidBase: Joi.string().default('base58btc'),
+        cidBase: Joi.string(),
         stream: Joi.boolean().default(false)
       })
         .rename('l', 'long', {

src/http/read.js

@@ -4,6 +4,7 @@ const Joi = require('@hapi/joi')
 const {
   PassThrough
 } = require('stream')
+const toStream = require('it-to-stream')
 
 const mfsRead = {
   method: 'POST',
@@ -19,10 +20,10 @@ const mfsRead = {
     } = request.query
 
     const responseStream = await new Promise((resolve, reject) => {
-      const stream = ipfs.files.readReadableStream(arg, {
+      const stream = toStream.readable(ipfs.files.read(arg, {
         offset,
         length
-      })
+      }))
 
       stream.once('data', (chunk) => {
         const passThrough = new PassThrough()

src/http/stat.js

@@ -20,15 +20,14 @@ const mfsStat = {
     const stats = await ipfs.files.stat(arg, {
       hash,
       size,
-      withLocal,
-      cidBase
+      withLocal
     })
 
     return h.response({
       Type: stats.type,
       Blocks: stats.blocks,
       Size: stats.size,
-      Hash: stats.hash,
+      Hash: stats.cid.toString(cidBase),
       CumulativeSize: stats.cumulativeSize,
       WithLocality: stats.withLocality,
       Local: stats.local,
@@ -49,7 +48,7 @@ const mfsStat = {
         hash: Joi.boolean().default(false),
         size: Joi.boolean().default(false),
         withLocal: Joi.boolean().default(false),
-        cidBase: Joi.string().default('base58btc')
+        cidBase: Joi.string()
       })
     }
   }