refactor: async iterables

package.json

@@ -42,27 +42,22 @@
  },
  "dependencies": {
  "abort-controller": "^3.0.0",
- "async-iterator-all": "^1.0.0",
- "async-iterator-to-pull-stream": "^1.3.0",
  "bignumber.js": "^9.0.0",
- "bl": "^4.0.0",
  "bs58": "^4.0.1",
  "buffer": "^5.4.2",
- "callbackify": "^1.1.0",
  "cids": "~0.7.1",
  "debug": "^4.1.0",
  "err-code": "^2.0.0",
  "explain-error": "^1.0.4",
  "form-data": "^3.0.0",
  "ipfs-block": "~0.8.1",
- "ipfs-utils": "^0.4.0",
+ "ipfs-utils": "github:ipfs/js-ipfs-utils#refactor/async-iterators",
  "ipld-dag-cbor": "~0.15.0",
  "ipld-dag-pb": "^0.18.1",
  "ipld-raw": "^4.0.0",
  "is-ipfs": "~0.6.1",
  "it-glob": "0.0.7",
  "it-tar": "^1.1.1",
- "it-to-stream": "^0.1.1",
  "iterable-ndjson": "^1.1.0",
  "ky": "^0.15.0",
  "ky-universal": "^0.3.0",
@@ -72,10 +67,7 @@
  "multibase": "~0.6.0",
  "multicodec": "~0.5.1",
  "multihashes": "~0.4.14",
- "parse-duration": "^0.1.1",
- "peer-id": "~0.12.3",
- "peer-info": "~0.15.1",
- "promise-nodeify": "^3.0.1"
+ "parse-duration": "^0.1.1"
  },
  "devDependencies": {
  "aegir": "^20.4.1",

src/bitswap/index.js

@@ -1,9 +1,7 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
-module.exports = (config) => ({
- wantlist: callbackify.variadic(require('./wantlist')(config)),
- stat: callbackify.variadic(require('./stat')(config)),
- unwant: callbackify.variadic(require('./unwant')(config))
+module.exports = config => ({
+ wantlist: require('./wantlist')(config),
+ stat: require('./stat')(config),
+ unwant: require('./unwant')(config)
 })

src/block/index.js

@@ -1,23 +1,8 @@
 'use strict'
 
-const nodeify = require('promise-nodeify')
-const callbackify = require('callbackify')
-const { collectify } = require('../lib/converters')
-
-module.exports = config => {
- const rm = require('./rm-async-iterator')(config)
-
- return {
- get: callbackify.variadic(require('./get')(config)),
- stat: callbackify.variadic(require('./stat')(config)),
- put: callbackify.variadic(require('./put')(config)),
- rm: (input, options, callback) => {
- if (typeof options === 'function') {
- callback = options
- options = {}
- }
- return nodeify(collectify(rm)(input, options), callback)
- },
- _rmAsyncIterator: rm
- }
-}
+module.exports = config => ({
+ get: require('./get')(config),
+ stat: require('./stat')(config),
+ put: require('./put')(config),
+ rm: require('./rm')(config)
+})

src/bootstrap/index.js

@@ -1,9 +1,7 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
 module.exports = config => ({
- add: callbackify.variadic(require('./add')(config)),
- rm: callbackify.variadic(require('./rm')(config)),
- list: callbackify.variadic(require('./list')(config))
+ add: require('./add')(config),
+ rm: require('./rm')(config),
+ list: require('./list')(config)
 })

src/config/index.js

@@ -1,10 +1,8 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
 module.exports = config => ({
- get: callbackify.variadic(require('./get')(config)),
- set: callbackify.variadic(require('./set')(config)),
- replace: callbackify.variadic(require('./replace')(config)),
+ get: require('./get')(config),
+ set: require('./set')(config),
+ replace: require('./replace')(config),
  profiles: require('./profiles')(config)
 })

src/config/profiles/index.js

@@ -1,8 +1,6 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
 module.exports = config => ({
- apply: callbackify.variadic(require('./apply')(config)),
- list: callbackify.variadic(require('./list')(config))
+ apply: require('./apply')(config),
+ list: require('./list')(config)
 })

src/dag/index.js

@@ -1,9 +1,7 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
 module.exports = config => ({
- get: callbackify.variadic(require('./get')(config)),
- put: callbackify.variadic(require('./put')(config)),
- resolve: callbackify.variadic(require('./resolve')(config))
+ get: require('./get')(config),
+ put: require('./put')(config),
+ resolve: require('./resolve')(config)
 })

src/dht/find-peer.js

@@ -1,7 +1,6 @@
 'use strict'
 
-const PeerId = require('peer-id')
-const PeerInfo = require('peer-info')
+const CID = require('cids')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
@@ -27,9 +26,10 @@ module.exports = configure(({ ky }) => {
  // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
  if (message.Type === 2 && message.Responses) {
  for (const { ID, Addrs } of message.Responses) {
- const peerInfo = new PeerInfo(PeerId.createFromB58String(ID))
- if (Addrs) Addrs.forEach(a => peerInfo.multiaddrs.add(multiaddr(a)))
- yield peerInfo
+ yield {
+ id: new CID(ID),
+ addrs: (Addrs || []).map(a => multiaddr(a))
+ }
  }
  }
  }

src/dht/find-provs.js

@@ -1,7 +1,6 @@
 'use strict'
 
-const PeerId = require('peer-id')
-const PeerInfo = require('peer-info')
+const CID = require('cids')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
@@ -28,9 +27,10 @@ module.exports = configure(({ ky }) => {
  // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L20
  if (message.Type === 4 && message.Responses) {
  for (const { ID, Addrs } of message.Responses) {
- const peerInfo = new PeerInfo(PeerId.createFromB58String(ID))
- if (Addrs) Addrs.forEach(a => peerInfo.multiaddrs.add(multiaddr(a)))
- yield peerInfo
+ yield {
+ id: new CID(ID),
+ addrs: (Addrs || []).map(a => multiaddr(a))
+ }
  }
  }
  }

src/dht/index.js

@@ -1,30 +1,11 @@
 'use strict'
 
-const callbackify = require('callbackify')
-const errCode = require('err-code')
-const { collectify } = require('../lib/converters')
-
-module.exports = config => {
- const get = require('./get')(config)
- const findPeer = require('./find-peer')(config)
-
- return {
- get: callbackify.variadic(async (key, options) => {
- for await (const value of get(key, options)) {
- return value
- }
- throw errCode(new Error('value not found'), 'ERR_TYPE_5_NOT_FOUND')
- }),
- put: callbackify.variadic(collectify(require('./put')(config))),
- findProvs: callbackify.variadic(collectify(require('./find-provs')(config))),
- findPeer: callbackify.variadic(async (peerId, options) => {
- for await (const peerInfo of findPeer(peerId, options)) {
- return peerInfo
- }
- throw errCode(new Error('final peer not found'), 'ERR_TYPE_2_NOT_FOUND')
- }),
- provide: callbackify.variadic(collectify(require('./provide')(config))),
- // find closest peerId to given peerId
- query: callbackify.variadic(collectify(require('./query')(config)))
- }
-}
+module.exports = config => ({
+ get: require('./get')(config),
+ put: require('./put')(config),
+ findProvs: require('./find-provs')(config),
+ findPeer: require('./find-peer')(config),
+ provide: require('./provide')(config),
+ // find closest peerId to given peerId
+ query: require('./query')(config)
+})

src/dht/provide.js

@@ -1,7 +1,6 @@
 'use strict'
 
-const PeerId = require('peer-id')
-const PeerInfo = require('peer-info')
+const CID = require('cids')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
@@ -28,11 +27,10 @@ module.exports = configure(({ ky }) => {
  for await (let message of ndjson(toIterable(res.body))) {
  message = toCamel(message)
  if (message.responses) {
- message.responses = message.responses.map(({ ID, Addrs }) => {
- const peerInfo = new PeerInfo(PeerId.createFromB58String(ID))
- if (Addrs) Addrs.forEach(a => peerInfo.multiaddrs.add(multiaddr(a)))
- return peerInfo
- })
+ message.responses = message.responses.map(({ ID, Addrs }) => ({
+ id: new CID(ID),
+ addrs: (Addrs || []).map(a => multiaddr(a))
+ }))
  }
  yield message
  }

src/dht/put.js

@@ -1,7 +1,6 @@
 'use strict'
 
-const PeerId = require('peer-id')
-const PeerInfo = require('peer-info')
+const CID = require('cids')
 const multiaddr = require('multiaddr')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
@@ -29,11 +28,10 @@ module.exports = configure(({ ky }) => {
  for await (let message of ndjson(toIterable(res.body))) {
  message = toCamel(message)
  if (message.responses) {
- message.responses = message.responses.map(({ ID, Addrs }) => {
- const peerInfo = new PeerInfo(PeerId.createFromB58String(ID))
- if (Addrs) Addrs.forEach(a => peerInfo.multiaddrs.add(multiaddr(a)))
- return peerInfo
- })
+ message.responses = message.responses.map(({ ID, Addrs }) => ({
+ id: new CID(ID),
+ addrs: (Addrs || []).map(a => multiaddr(a))
+ }))
  }
  yield message
  }

src/dht/query.js

@@ -1,7 +1,6 @@
 'use strict'
 
-const PeerId = require('peer-id')
-const PeerInfo = require('peer-info')
+const CID = require('cids')
 const ndjson = require('iterable-ndjson')
 const configure = require('../lib/configure')
 const toIterable = require('../lib/stream-to-iterable')
@@ -22,7 +21,10 @@ module.exports = configure(({ ky }) => {
  })
 
  for await (const message of ndjson(toIterable(res.body))) {
- yield new PeerInfo(PeerId.createFromB58String(message.ID))
+ yield {
+ id: new CID(message.ID),
+ addrs: []
+ }
  }
  }
 })

src/diag/index.js

@@ -1,9 +1,7 @@
 'use strict'
 
-const callbackify = require('callbackify')
-
 module.exports = config => ({
- net: callbackify.variadic(require('./net')(config)),
- sys: callbackify.variadic(require('./sys')(config)),
- cmds: callbackify.variadic(require('./cmds')(config))
+ net: require('./net')(config),
+ sys: require('./sys')(config),
+ cmds: require('./cmds')(config)
 })

src/files/index.js

@@ -1,25 +1,13 @@
 'use strict'
 
-const callbackify = require('callbackify')
-const { collectify, streamify, pullify, concatify } = require('../lib/converters')
-
-module.exports = config => {
- const ls = require('./ls')(config)
- const read = require('./read')(config)
-
- return {
- cp: callbackify.variadic(require('./cp')(config)),
- mkdir: callbackify.variadic(require('./mkdir')(config)),
- flush: callbackify.variadic(require('./flush')(config)),
- stat: callbackify.variadic(require('./stat')(config)),
- rm: callbackify.variadic(require('./rm')(config)),
- ls: callbackify.variadic(collectify(ls)),
- lsReadableStream: streamify.readable(ls),
- lsPullStream: pullify.source(ls),
- read: callbackify.variadic(concatify(read)),
- readReadableStream: streamify.readable(read),
- readPullStream: pullify.source(read),
- write: callbackify.variadic(require('./write')(config)),
- mv: callbackify.variadic(require('./mv')(config))
- }
-}
+module.exports = config => ({
+ cp: require('./cp')(config),
+ mkdir: require('./mkdir')(config),
+ flush: require('./flush')(config),
+ stat: require('./stat')(config),
+ rm: require('./rm')(config),
+ ls: require('./ls')(config),
+ read: require('./read')(config),
+ write: require('./write')(config),
+ mv: require('./mv')(config)
+})