refactor: async iterables

Author: Alan Shaw

README.md

@@ -302,9 +302,9 @@ a case, you should provide a way to trigger migrations manually.**
 |------|---------|
 | boolean or object | `true` |
 
-Initialize the repo when creating the IPFS node.
+Perform repo initialization steps when creating the IPFS node.
 
-If you have already initialized a repo before creating your IPFS node (e.g. you are loading a repo that was saved to disk from a previous run of your program), you must make sure to set this to `false`. Note that *initializing* a repo is different from creating an instance of [`ipfs.Repo`](https://github.com/ipfs/js-ipfs-repo). The IPFS constructor sets many special properties when initializing a repo, so you should usually not try and call `repoInstance.init()` yourself.
+Note that *initializing* a repo is different from creating an instance of [`ipfs.Repo`](https://github.com/ipfs/js-ipfs-repo). The IPFS constructor sets many special properties when initializing a repo, so you should usually not try and call `repoInstance.init()` yourself.
 
 Instead of a boolean, you may provide an object with custom initialization options. All properties are optional:
 
@@ -314,13 +314,13 @@ Instead of a boolean, you may provide an object with custom initialization optio
     ```js
     // Generating a Peer ID:
     const PeerId = require('peer-id')
-    PeerId.create({ bits: 2048 }, (err, peerId) => {
-      // Generates a new Peer ID, complete with public/private keypair
-      // See https://github.com/libp2p/js-peer-id
-    })
+    // Generates a new Peer ID, complete with public/private keypair
+    // See https://github.com/libp2p/js-peer-id
+    const peerId = await PeerId.create({ bits: 2048 })
     ```
 - `pass` (string) A passphrase to encrypt keys. You should generally use the [top-level `pass` option](#optionspass) instead of the `init.pass` option (this one will take its value from the top-level option if not set).
 - `profiles` (Array) Apply profile settings to config.
+- `allowNew` (boolean, default: `true`) Set to `false` to disallow initialization if the repo does not already exist.
 
 ##### `options.start`
 

package.json

@@ -15,8 +15,7 @@
   ],
   "main": "src/core/index.js",
   "browser": {
-    "./src/core/components/init-assets.js": false,
-    "./src/core/runtime/add-from-fs-nodejs.js": "./src/core/runtime/add-from-fs-browser.js",
+    "./src/core/runtime/init-assets-nodejs.js": "./src/core/runtime/init-assets-browser.js",
     "./src/core/runtime/config-nodejs.js": "./src/core/runtime/config-browser.js",
     "./src/core/runtime/dns-nodejs.js": "./src/core/runtime/dns-browser.js",
     "./src/core/runtime/libp2p-nodejs.js": "./src/core/runtime/libp2p-browser.js",
@@ -25,7 +24,8 @@
     "./src/core/runtime/repo-nodejs.js": "./src/core/runtime/repo-browser.js",
     "./src/core/runtime/ipld-nodejs.js": "./src/core/runtime/ipld-browser.js",
     "./test/utils/create-repo-nodejs.js": "./test/utils/create-repo-browser.js",
-    "stream": "readable-stream"
+    "stream": "readable-stream",
+    "ipfs-utils/src/files/glob-source": false
   },
   "browser-all-ipld-formats": {
     "./src/core/runtime/ipld-browser.js": "./src/core/runtime/ipld-browser-all.js"
@@ -65,6 +65,7 @@
     "@hapi/hapi": "^18.3.2",
     "@hapi/joi": "^15.0.0",
     "abort-controller": "^3.0.0",
+    "any-signal": "^1.1.0",
     "array-shuffle": "^1.0.1",
     "async-iterator-to-pull-stream": "^1.3.0",
     "async-iterator-to-stream": "^1.1.0",
@@ -81,7 +82,8 @@
     "class-is": "^1.1.0",
     "dag-cbor-links": "^1.3.2",
     "datastore-core": "~0.7.0",
-    "datastore-pubsub": "^0.2.1",
+    "datastore-level": "github:ipfs/js-datastore-level#fix/iterator-memory-leak",
+    "datastore-pubsub": "^0.3.0",
     "debug": "^4.1.0",
     "dlv": "^1.1.3",
     "err-code": "^2.0.0",
@@ -93,20 +95,19 @@
     "glob": "^7.1.3",
     "hapi-pino": "^6.1.0",
     "hashlru": "^2.3.0",
-    "human-to-milliseconds": "^2.0.0",
     "interface-datastore": "~0.8.0",
-    "ipfs-bitswap": "^0.26.0",
+    "ipfs-bitswap": "github:ipfs/js-ipfs-bitswap#refactor/libp2p-async",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.16.0",
-    "ipfs-http-client": "^40.1.0",
-    "ipfs-http-response": "~0.4.0",
-    "ipfs-mfs": "^0.13.2",
-    "ipfs-multipart": "^0.2.0",
+    "ipfs-http-client": "github:ipfs/js-ipfs-http-client#refactor/async-iterables2",
+    "ipfs-http-response": "^0.5.0",
+    "ipfs-mfs": "github:ipfs/js-ipfs-mfs#refactor/remove-streams",
+    "ipfs-multipart": "^0.3.0",
     "ipfs-repo": "^0.30.0",
-    "ipfs-unixfs": "~0.1.16",
-    "ipfs-unixfs-exporter": "^0.38.0",
-    "ipfs-unixfs-importer": "^0.42.0",
-    "ipfs-utils": "~0.4.0",
+    "ipfs-unixfs": "^0.3.0",
+    "ipfs-unixfs-exporter": "^0.40.0",
+    "ipfs-unixfs-importer": "^0.43.1",
+    "ipfs-utils": "^0.6.0",
     "ipld": "~0.25.0",
     "ipld-bitcoin": "~0.3.0",
     "ipld-dag-cbor": "~0.15.0",
@@ -115,54 +116,59 @@
     "ipld-git": "~0.5.0",
     "ipld-raw": "^4.0.0",
     "ipld-zcash": "~0.4.0",
-    "ipns": "^0.6.1",
+    "ipns": "^0.7.0",
     "is-domain-name": "^1.0.1",
     "is-ipfs": "~0.6.1",
     "is-pull-stream": "~0.0.0",
     "is-stream": "^2.0.0",
     "iso-url": "~0.4.6",
     "it-all": "^1.0.1",
-    "it-pipe": "^1.0.1",
+    "it-concat": "^1.0.0",
+    "it-last": "^1.0.1",
+    "it-pipe": "^1.1.0",
     "it-to-stream": "^0.1.1",
+    "iterable-ndjson": "^1.1.0",
     "jsondiffpatch": "~0.3.11",
     "just-safe-set": "^2.1.0",
     "kind-of": "^6.0.2",
     "ky": "^0.15.0",
     "ky-universal": "~0.3.0",
-    "libp2p": "^0.26.2",
-    "libp2p-bootstrap": "~0.9.3",
-    "libp2p-crypto": "^0.16.2",
+    "libp2p": "github:libp2p/js-libp2p#refactor/async-await",
+    "libp2p-bootstrap": "^0.10.2",
+    "libp2p-crypto": "^0.17.1",
     "libp2p-delegated-content-routing": "^0.4.1",
-    "libp2p-delegated-peer-routing": "^0.3.1",
-    "libp2p-floodsub": "^0.18.0",
-    "libp2p-gossipsub": "~0.0.5",
-    "libp2p-kad-dht": "~0.16.0",
-    "libp2p-keychain": "^0.5.2",
-    "libp2p-mdns": "~0.12.0",
+    "libp2p-delegated-peer-routing": "^0.4.0",
+    "libp2p-floodsub": "^0.20.0",
+    "libp2p-gossipsub": "^0.2.0",
+    "libp2p-kad-dht": "^0.18.3",
+    "libp2p-keychain": "^0.6.0",
+    "libp2p-mdns": "^0.13.0",
+    "libp2p-mplex": "^0.9.3",
     "libp2p-record": "~0.7.0",
-    "libp2p-secio": "~0.11.0",
-    "libp2p-tcp": "^0.13.0",
-    "libp2p-webrtc-star": "~0.16.0",
+    "libp2p-secio": "^0.12.1",
+    "libp2p-tcp": "^0.14.2",
+    "libp2p-webrtc-star": "^0.17.0",
     "libp2p-websocket-star-multi": "~0.4.3",
-    "libp2p-websockets": "~0.12.3",
+    "libp2p-websockets": "^0.13.0",
     "lodash.flatten": "^4.4.0",
-    "mafmt": "^6.0.10",
+    "mafmt": "^7.0.0",
     "merge-options": "^2.0.0",
     "mime-types": "^2.1.21",
     "mkdirp": "~0.5.1",
     "mortice": "^2.0.0",
-    "multiaddr": "^6.1.1",
-    "multiaddr-to-uri": "^5.0.0",
+    "multiaddr": "^7.2.1",
+    "multiaddr-to-uri": "^5.1.0",
     "multibase": "~0.6.0",
-    "multicodec": "~0.5.5",
+    "multicodec": "^1.0.0",
     "multihashes": "~0.4.14",
     "multihashing-async": "^0.8.0",
     "node-fetch": "^2.3.0",
+    "p-defer": "^3.0.0",
     "p-iteration": "^1.1.8",
     "p-queue": "^6.1.0",
-    "peer-book": "^0.9.1",
-    "peer-id": "~0.12.2",
-    "peer-info": "~0.15.1",
+    "parse-duration": "^0.1.2",
+    "peer-id": "^0.13.5",
+    "peer-info": "^0.17.0",
     "pretty-bytes": "^5.3.0",
     "progress": "^2.0.1",
     "promise-nodeify": "^3.0.1",
@@ -172,7 +178,6 @@
     "pull-cat": "^1.1.11",
     "pull-defer": "~0.2.3",
     "pull-file": "^1.1.0",
-    "pull-mplex": "~0.1.1",
     "pull-ndjson": "^0.2.0",
     "pull-pushable": "^2.2.0",
     "pull-sort": "^1.0.1",
@@ -183,12 +188,15 @@
     "readable-stream": "^3.4.0",
     "receptacle": "^1.3.2",
     "semver": "^6.3.0",
+    "stream-to-it": "^0.2.0",
     "stream-to-pull-stream": "^1.7.3",
+    "streaming-iterables": "^4.1.1",
     "superstruct": "~0.6.2",
     "tar-stream": "^2.0.0",
     "temp": "~0.9.0",
+    "timeout-abort-controller": "^1.1.0",
     "update-notifier": "^4.0.0",
-    "uri-to-multiaddr": "^3.0.1",
+    "uri-to-multiaddr": "^3.0.2",
     "varint": "^5.0.0",
     "yargs": "^15.0.1",
     "yargs-promise": "^1.1.0"
@@ -204,14 +212,16 @@
     "execa": "^3.0.0",
     "form-data": "^3.0.0",
     "hat": "0.0.3",
-    "interface-ipfs-core": "^0.125.0",
+    "interface-ipfs-core": "github:ipfs/interface-js-ipfs-core#refactor/async-iterables",
     "ipfs-interop": "^0.1.1",
     "ipfsd-ctl": "^1.0.2",
     "libp2p-websocket-star": "~0.10.2",
     "lodash": "^4.17.15",
     "ncp": "^2.0.0",
     "p-event": "^4.1.0",
     "p-map": "^3.0.0",
+    "peer-book": "^0.9.2",
+    "pull-mplex": "^0.1.2",
     "qs": "^6.5.2",
     "rimraf": "^3.0.0",
     "sinon": "^7.4.2",

src/cli/bin.js

@@ -37,38 +37,22 @@ updateNotifier({ pkg, updateCheckInterval: oneWeek }).notify()
 
 const cli = new YargsPromise(parser)
 
-let getIpfs = null
-
 // Apply command aliasing (eg `refs local` -> `refs-local`)
 const args = commandAlias(process.argv.slice(2))
 cli
   .parse(args)
   .then(({ data, argv }) => {
-    getIpfs = argv.getIpfs
     if (data) {
       print(data)
     }
   })
   .catch(({ error, argv }) => {
-    getIpfs = argv && argv.getIpfs
-
     if (error.code === InvalidRepoVersionError.code) {
       error.message = 'Incompatible repo version. Migration needed. Pass --migrate for automatic migration'
     }
 
-    if (error.message) {
-      print(error.message)
-      debug(error)
-    } else {
-      print('Unknown error, please re-run the command with DEBUG=ipfs:cli to see debug output')
-      debug(error)
-    }
+    print(error.message || 'Unknown error, please re-run the command with DEBUG=ipfs:cli to see debug output')
+    debug(error)
 
     process.exit(1)
   })
-  .finally(() => {
-    if (getIpfs && getIpfs.instance) {
-      const cleanup = getIpfs.rest[0]
-      return cleanup()
-    }
-  })

src/cli/commands/add.js

@@ -175,20 +175,20 @@ module.exports = {
         ? globSource(argv.file, { recursive: argv.recursive, hidden: argv.hidden })
         : process.stdin // Pipe directly to ipfs.add
 
-      let finalHash
+      let finalCid
 
       try {
-        for await (const file of ipfs._addAsyncIterator(source, options)) {
+        for await (const file of ipfs.add(source, options)) {
           if (argv.silent) {
             continue
           }
 
           if (argv.quieter) {
-            finalHash = file.hash
+            finalCid = file.cid
             continue
           }
 
-          const cid = cidToString(file.hash, { base: argv.cidBase })
+          const cid = cidToString(file.cid, { base: argv.cidBase })
           let message = cid
 
           if (!argv.quiet) {
@@ -203,7 +203,7 @@ module.exports = {
           bar.terminate()
         }
 
-        // Tweak the error message and add more relevant infor for the CLI
+        // Tweak the error message and add more relevant info for the CLI
         if (err.code === 'ERR_DIR_NON_RECURSIVE') {
           err.message = `'${err.path}' is a directory, use the '-r' flag to specify directories`
         }
@@ -216,7 +216,7 @@ module.exports = {
       }
 
       if (argv.quieter) {
-        log(cidToString(finalHash, { base: argv.cidBase }))
+        log(cidToString(finalCid, { base: argv.cidBase }))
       }
     })())
   }

src/cli/commands/bitswap/stat.js

@@ -35,7 +35,7 @@ module.exports = {
         stats.dupDataReceived = prettyBytes(stats.dupDataReceived.toNumber()).toUpperCase()
         stats.wantlist = `[${stats.wantlist.length} keys]`
       } else {
-        const wantlist = stats.wantlist.map((elem) => cidToString(elem['/'], { base: cidBase, upgrade: false }))
+        const wantlist = stats.wantlist.map(cid => cidToString(cid, { base: cidBase, upgrade: false }))
         stats.wantlist = `[${wantlist.length} keys]
             ${wantlist.join('\n            ')}`
       }

src/cli/commands/bitswap/wantlist.js

@@ -25,7 +25,7 @@ module.exports = {
     resolve((async () => {
       const ipfs = await getIpfs()
       const list = await ipfs.bitswap.wantlist(peer)
-      list.Keys.forEach(k => print(cidToString(k['/'], { base: cidBase, upgrade: false })))
+      list.forEach(cid => print(cidToString(cid, { base: cidBase, upgrade: false })))
     })())
   }
 }

src/cli/commands/block/put.js

@@ -1,9 +1,8 @@
 'use strict'
 
-const bl = require('bl')
 const fs = require('fs')
 const multibase = require('multibase')
-const promisify = require('promisify-es6')
+const concat = require('it-concat')
 const { cidToString } = require('../../../utils/cid')
 
 module.exports = {
@@ -41,14 +40,9 @@ module.exports = {
       let data
 
       if (argv.block) {
-        data = await promisify(fs.readFile)(argv.block)
+        data = await fs.readFileSync(argv.block)
       } else {
-        data = await new Promise((resolve, reject) => {
-          process.stdin.pipe(bl((err, input) => {
-            if (err) return reject(err)
-            resolve(input)
-          }))
-        })
+        data = await concat(process.stdin)
       }
 
       const ipfs = await argv.getIpfs()

src/cli/commands/block/rm.js

@@ -25,7 +25,7 @@ module.exports = {
       const ipfs = await getIpfs()
       let errored = false
 
-      for await (const result of ipfs.block._rmAsyncIterator(hash, {
+      for await (const result of ipfs.block.rm(hash, {
         force,
         quiet
       })) {

src/cli/commands/block/stat.js

@@ -20,7 +20,7 @@ module.exports = {
     resolve((async () => {
       const ipfs = await getIpfs()
       const stats = await ipfs.block.stat(key)
-      print('Key: ' + cidToString(stats.key, { base: cidBase }))
+      print('Key: ' + cidToString(stats.cid, { base: cidBase }))
       print('Size: ' + stats.size)
     })())
   }

src/cli/commands/cat.js

@@ -22,7 +22,7 @@ module.exports = {
     resolve((async () => {
       const ipfs = await getIpfs()
 
-      for await (const buf of ipfs._catAsyncIterator(ipfsPath, { offset, length })) {
+      for await (const buf of ipfs.cat(ipfsPath, { offset, length })) {
         process.stdout.write(buf)
       }
     })())

src/cli/commands/dag/resolve.js

@@ -19,10 +19,9 @@ module.exports = {
       const options = {}
 
       try {
-        const result = await ipfs.dag.resolve(ref, options)
         let lastCid
 
-        for (const res of result) {
+        for await (const res of ipfs.dag.resolve(ref, options)) {
           if (CID.isCID(res.value)) {
             lastCid = res.value
           }