From ec3aa81d5200876ea7b9ede826c9489c79bdc5f1 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Mon, 20 Apr 2020 20:42:10 +0200 Subject: [PATCH 1/8] chore: use libp2p 0.28.x --- package.json | 6 +-- src/index.js | 12 ++--- src/network.js | 26 +++++------ test/bitswap-stats.js | 5 +- test/bitswap.js | 26 +++++++++-- test/network/gen-bitswap-network.node.js | 2 +- test/network/network.node.js | 27 ++++++----- test/utils/connect-all.js | 3 +- test/utils/create-libp2p-node.js | 13 ++++-- test/utils/mocks.js | 59 +++++++++++++----------- 10 files changed, 102 insertions(+), 77 deletions(-) diff --git a/package.json b/package.json index 590cb05a..42400ca8 100644 --- a/package.json +++ b/package.json @@ -52,8 +52,8 @@ "iso-random-stream": "^1.1.1", "it-all": "^1.0.2", "it-drain": "^1.0.1", - "libp2p": "^0.27.0", - "libp2p-kad-dht": "^0.18.3", + "libp2p": "libp2p/js-libp2p#0.28.x", + "libp2p-kad-dht": "^0.19.1", "libp2p-mplex": "^0.9.2", "libp2p-secio": "^0.12.1", "libp2p-tcp": "^0.14.2", @@ -65,9 +65,7 @@ "p-defer": "^3.0.0", "p-event": "^4.1.0", "p-wait-for": "^3.1.0", - "peer-book": "~0.9.0", "peer-id": "^0.13.5", - "peer-info": "^0.17.0", "promisify-es6": "^1.0.3", "rimraf": "^3.0.0", "sinon": "^9.0.0", diff --git a/src/index.js b/src/index.js index 0ef91d17..5c39c00a 100644 --- a/src/index.js +++ b/src/index.js @@ -37,7 +37,7 @@ const statsKeys = [ class Bitswap { constructor (libp2p, blockstore, options) { this._libp2p = libp2p - this._log = logger(this.peerInfo.id) + this._log = logger(this.peerId) this._options = Object.assign({}, defaultOptions, options) @@ -54,16 +54,16 @@ class Bitswap { // local database this.blockstore = blockstore - this.engine = new DecisionEngine(this.peerInfo.id, blockstore, this.network, this._stats) + this.engine = new DecisionEngine(this.peerId, blockstore, this.network, this._stats) // handle message sending - this.wm = new WantManager(this.peerInfo.id, this.network, this._stats) + this.wm = new WantManager(this.peerId, this.network, this._stats) - this.notifications = new Notifications(this.peerInfo.id) + this.notifications = new Notifications(this.peerId) } - get peerInfo () { - return this._libp2p.peerInfo + get peerId () { + return this._libp2p.peerId } // handle messages received through the network diff --git a/src/network.js b/src/network.js index 8d34b157..fd79fd16 100644 --- a/src/network.js +++ b/src/network.js @@ -13,7 +13,7 @@ const BITSWAP120 = '/ipfs/bitswap/1.2.0' class Network { constructor (libp2p, bitswap, options, stats) { - this._log = logger(libp2p.peerInfo.id, 'network') + this._log = logger(libp2p.peerId, 'network') options = options || {} this.libp2p = libp2p this.bitswap = bitswap @@ -37,14 +37,14 @@ class Network { this._running = true this.libp2p.handle(this.protocols, this._onConnection) - this.libp2p.on('peer:connect', this._onPeerConnect) - this.libp2p.on('peer:disconnect', this._onPeerDisconnect) + this.libp2p.connectionManager.on('peer:connect', this._onPeerConnect) + this.libp2p.connectionManager.on('peer:disconnect', this._onPeerDisconnect) // All existing connections are like new ones for us for (const peer of this.libp2p.peerStore.peers.values()) { - if (this.libp2p.registrar.getConnection(peer)) { - this._onPeerConnect(peer) - } + const conn = this.libp2p.connectionManager.get(peer.id) + + conn && this._onPeerConnect(conn) } } @@ -54,8 +54,8 @@ class Network { // Unhandle both, libp2p doesn't care if it's not already handled this.libp2p.unhandle(this.protocols) - this.libp2p.removeListener('peer:connect', this._onPeerConnect) - this.libp2p.removeListener('peer:disconnect', this._onPeerDisconnect) + this.libp2p.connectionManager.removeListener('peer:connect', this._onPeerConnect) + this.libp2p.connectionManager.removeListener('peer:disconnect', this._onPeerDisconnect) } /** @@ -92,12 +92,12 @@ class Network { } } - _onPeerConnect (peerInfo) { - this.bitswap._onPeerConnected(peerInfo.id) + _onPeerConnect (connection) { + this.bitswap._onPeerConnected(connection.remotePeer) } - _onPeerDisconnect (peerInfo) { - this.bitswap._onPeerDisconnected(peerInfo.id) + _onPeerDisconnect (connection) { + this.bitswap._onPeerDisconnected(connection.remotePeer) } /** @@ -181,7 +181,7 @@ class Network { /** * Connects to another peer * - * @param {PeerInfo|PeerId|Multiaddr} peer + * @param {PeerId|Multiaddr} peer * @param {Object} options * @param {AbortSignal} options.abortSignal * @returns {Promise} diff --git a/test/bitswap-stats.js b/test/bitswap-stats.js index 94167f93..cdf4afda 100644 --- a/test/bitswap-stats.js +++ b/test/bitswap-stats.js @@ -169,7 +169,8 @@ describe('bitswap stats', () => { bs2 = bitswaps[1] bs2.start() - await libp2pNodes[0].dial(libp2pNodes[1].peerInfo) + libp2pNodes[0].peerStore.addressBook.set(libp2pNodes[1].peerId, libp2pNodes[1].multiaddrs) + await libp2pNodes[0].dial(libp2pNodes[1].peerId) block = await makeBlock() @@ -212,7 +213,7 @@ describe('bitswap stats', () => { }) it('has peer stats', async () => { - const peerStats = bs2.stat().forPeer(libp2pNodes[0].peerInfo.id) + const peerStats = bs2.stat().forPeer(libp2pNodes[0].peerId) expect(peerStats).to.exist() const stats = await pEvent(peerStats, 'update') diff --git a/test/bitswap.js b/test/bitswap.js index 2679e437..a248f2ae 100644 --- a/test/bitswap.js +++ b/test/bitswap.js @@ -5,6 +5,7 @@ const { expect } = require('aegir/utils/chai') const delay = require('delay') const PeerId = require('peer-id') const sinon = require('sinon') +const pWaitFor = require('p-wait-for') const Bitswap = require('../src') @@ -38,9 +39,12 @@ describe('bitswap without DHT', function () { ]) // connect 0 -> 1 && 1 -> 2 + nodes[0].libp2pNode.peerStore.addressBook.set(nodes[1].libp2pNode.peerId, nodes[1].libp2pNode.multiaddrs) + nodes[1].libp2pNode.peerStore.addressBook.set(nodes[2].libp2pNode.peerId, nodes[2].libp2pNode.multiaddrs) + await Promise.all([ - nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerInfo), - nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerInfo) + nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerId), + nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerId) ]) }) @@ -132,10 +136,22 @@ describe('bitswap with DHT', function () { ]) // connect 0 -> 1 && 1 -> 2 + nodes[0].libp2pNode.peerStore.addressBook.set(nodes[1].libp2pNode.peerId, nodes[1].libp2pNode.multiaddrs) + nodes[1].libp2pNode.peerStore.addressBook.set(nodes[2].libp2pNode.peerId, nodes[2].libp2pNode.multiaddrs) + + await Promise.all([ + nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerId), + nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerId) + ]) + + // await dht routing table are updated await Promise.all([ - nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerInfo), - nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerInfo) + pWaitFor(() => nodes[0].libp2pNode._dht.routingTable.size >= 1), + pWaitFor(() => nodes[1].libp2pNode._dht.routingTable.size >= 1) ]) + + // Give time to process + await delay(300) }) after(async () => { @@ -151,7 +167,7 @@ describe('bitswap with DHT', function () { await nodes[2].bitswap.put(block) // Give put time to process - await delay(100) + await delay(300) const blockRetrieved = await nodes[0].bitswap.get(block.cid) expect(block.data).to.eql(blockRetrieved.data) diff --git a/test/network/gen-bitswap-network.node.js b/test/network/gen-bitswap-network.node.js index 4fb1087d..ef48044a 100644 --- a/test/network/gen-bitswap-network.node.js +++ b/test/network/gen-bitswap-network.node.js @@ -79,7 +79,7 @@ async function exchangeBlocks (nodes, blocksPerNode = 10) { const d = Date.now() // fetch all blocks on every node - await Promise.all(nodes.map(async (node) => { + await Promise.all(nodes.map(async (node, index) => { const bs = await Promise.all(cids.map((cid) => node.bitswap.get(cid))) expect(bs).to.deep.equal(blocks) })) diff --git a/test/network/network.node.js b/test/network/network.node.js index acad3053..e2144d00 100644 --- a/test/network/network.node.js +++ b/test/network/network.node.js @@ -76,7 +76,7 @@ describe('network', () => { it('connectTo fail', async () => { try { - await networkA.connectTo(p2pB.peerInfo.id) + await networkA.connectTo(p2pB.peerId) assert.fail() } catch (err) { expect(err).to.exist() @@ -87,16 +87,17 @@ describe('network', () => { var counter = 0 bitswapMockA._onPeerConnected = (peerId) => { - expect(peerId.toB58String()).to.equal(p2pB.peerInfo.id.toB58String()) + expect(peerId.toB58String()).to.equal(p2pB.peerId.toB58String()) counter++ } bitswapMockB._onPeerConnected = (peerId) => { - expect(peerId.toB58String()).to.equal(p2pA.peerInfo.id.toB58String()) + expect(peerId.toB58String()).to.equal(p2pA.peerId.toB58String()) counter++ } - await p2pA.dial(p2pB.peerInfo) + p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) + await p2pA.dial(p2pB.peerId) await pWaitFor(() => counter >= 2) bitswapMockA._onPeerConnected = () => {} @@ -104,7 +105,8 @@ describe('network', () => { }) it('connectTo success', async () => { - await networkA.connectTo(p2pB.peerInfo) + p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) + await networkA.connectTo(p2pB.peerId) }) const versions = [{ @@ -134,7 +136,8 @@ describe('network', () => { bitswapMockB._receiveError = (err) => deferred.reject(err) - const { stream } = await p2pA.dialProtocol(p2pB.peerInfo, '/ipfs/bitswap/' + version.num) + // TODO: set addr + const { stream } = await p2pA.dialProtocol(p2pB.peerId, '/ipfs/bitswap/' + version.num) await pipe( [version.serialize(msg)], lp.encode(), @@ -165,11 +168,12 @@ describe('network', () => { bitswapMockB._receiveError = deferred.reject - await networkA.sendMessage(p2pB.peerInfo.id, msg) + await networkA.sendMessage(p2pB.peerId, msg) }) it('dial to peer on Bitswap 1.0.0', async () => { - const { protocol } = await p2pA.dialProtocol(p2pC.peerInfo, ['/ipfs/bitswap/1.1.0', '/ipfs/bitswap/1.0.0']) + p2pA.peerStore.addressBook.set(p2pC.peerId, p2pC.multiaddrs) + const { protocol } = await p2pA.dialProtocol(p2pC.peerId, ['/ipfs/bitswap/1.1.0', '/ipfs/bitswap/1.0.0']) expect(protocol).to.equal('/ipfs/bitswap/1.0.0') }) @@ -194,7 +198,7 @@ describe('network', () => { bitswapMockC._receiveError = deferred.reject - await networkA.sendMessage(p2pC.peerInfo.id, msg) + await networkA.sendMessage(p2pC.peerId, msg) await deferred.promise }) @@ -208,8 +212,7 @@ describe('network', () => { networkA.start() networkB.start() - // FIXME: have to already be connected as sendMessage only accepts a peer id, not a PeerInfo - await p2pA.dial(p2pB.peerInfo) + p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) const deferred = pDefer() @@ -217,7 +220,7 @@ describe('network', () => { deferred.resolve() } - await networkA.sendMessage(p2pB.peerInfo.id, new Message(true)) + await networkA.sendMessage(p2pB.peerId, new Message(true)) return deferred }) diff --git a/test/utils/connect-all.js b/test/utils/connect-all.js index 100bebd2..a396d4ad 100644 --- a/test/utils/connect-all.js +++ b/test/utils/connect-all.js @@ -5,7 +5,8 @@ const without = require('lodash.without') module.exports = async (nodes) => { for (const node of nodes) { for (const otherNode of without(nodes, node)) { - await node.libp2pNode.dial(otherNode.bitswap.peerInfo) + // TODO: set addrs + await node.libp2pNode.dial(otherNode.bitswap.peerId) } } } diff --git a/test/utils/create-libp2p-node.js b/test/utils/create-libp2p-node.js index 4e630a35..98649980 100644 --- a/test/utils/create-libp2p-node.js +++ b/test/utils/create-libp2p-node.js @@ -5,8 +5,8 @@ const MPLEX = require('libp2p-mplex') const SECIO = require('libp2p-secio') const libp2p = require('libp2p') const KadDHT = require('libp2p-kad-dht') -const PeerInfo = require('peer-info') const PeerId = require('peer-id') + const defaultsDeep = require('@nodeutils/defaults-deep') class Node extends libp2p { @@ -38,10 +38,13 @@ class Node extends libp2p { async function createLibp2pNode (options = {}) { const id = await PeerId.create({ bits: 512 }) - const peerInfo = new PeerInfo(id) - peerInfo.multiaddrs.add('/ip4/0.0.0.0/tcp/0') - options.peerInfo = peerInfo - const node = new Node(options) + const node = new Node({ + peerId: id, + addresses: { + listen: ['/ip4/0.0.0.0/tcp/0'] + }, + ...options + }) await node.start() return node diff --git a/test/utils/mocks.js b/test/utils/mocks.js index 2179b371..293fc7d1 100644 --- a/test/utils/mocks.js +++ b/test/utils/mocks.js @@ -1,8 +1,9 @@ 'use strict' const range = require('lodash.range') + const PeerId = require('peer-id') -const PeerInfo = require('peer-info') + const PeerStore = require('libp2p/src/peer-store') const Node = require('./create-libp2p-node').bundle const tmpdir = require('ipfs-utils/src/temp-dir') @@ -15,17 +16,21 @@ const Bitswap = require('../../src') * Create a mock libp2p node */ exports.mockLibp2pNode = () => { - const peerInfo = new PeerInfo(PeerId.createFromHexString('122019318b6e5e0cf93a2314bf01269a2cc23cd3dcd452d742cdb9379d8646f6e4a9')) + const peerId = PeerId.createFromHexString('122019318b6e5e0cf93a2314bf01269a2cc23cd3dcd452d742cdb9379d8646f6e4a9') return Object.assign(new EventEmitter(), { - peerInfo: peerInfo, + peerId, + multiaddrs: [], handle () {}, unhandle () {}, contentRouting: { provide: async (cid) => {}, // eslint-disable-line require-await findProviders: async (cid, timeout) => { return [] } // eslint-disable-line require-await }, - on () {}, + connectionManager: { + on () {}, + removeListener () {} + }, async dial (peer) { // eslint-disable-line require-await }, async dialProtocol (peer, protocol) { // eslint-disable-line require-await @@ -139,58 +144,44 @@ exports.applyNetwork = (bs, n) => { bs.engine.network = n } -let basePort = 12000 - /** * @private * @param {number} n The number of nodes in the network * @param {boolean} enableDHT Whether or not to run the dht */ exports.genBitswapNetwork = async (n, enableDHT = false) => { - const netArray = [] // bitswap, peerStore, libp2p, peerInfo, repo + const netArray = [] // bitswap, peerStore, libp2p, peerId, repo - // create PeerInfo and libp2p.Node for each + // create PeerId and libp2p.Node for each const peers = await Promise.all( - range(n).map(i => PeerInfo.create()) + range(n).map(i => PeerId.create()) ) peers.forEach((p, i) => { - basePort++ - p.multiaddrs.add('/ip4/127.0.0.1/tcp/' + basePort + '/ipfs/' + p.id.toB58String()) - const l = new Node({ - peerInfo: p, + peerId: p, + addresses: { + listen: ['/ip4/127.0.0.1/tcp/0'] + }, config: { dht: { enabled: enableDHT } } }) - netArray.push({ peerInfo: p, libp2p: l }) - }) - - // create PeerStore and populate peerStore - netArray.forEach((net, i) => { - const pb = netArray[i].libp2p.peerStore - netArray.forEach((net, j) => { - if (i === j) { - return - } - pb.put(net.peerInfo) - }) - netArray[i].peerStore = pb + netArray.push({ peerId: p, libp2p: l }) }) // create the repos const tmpDir = tmpdir() netArray.forEach((net, i) => { - const repoPath = tmpDir + '/' + net.peerInfo.id.toB58String() + const repoPath = tmpDir + '/' + net.peerId.toB58String() net.repo = new Repo(repoPath) }) await Promise.all( netArray.map(async (net) => { - const repoPath = tmpDir + '/' + net.peerInfo.id.toB58String() + const repoPath = tmpDir + '/' + net.peerId.toB58String() net.repo = new Repo(repoPath) await net.repo.init({}) @@ -203,6 +194,18 @@ exports.genBitswapNetwork = async (n, enableDHT = false) => { netArray.map((net) => net.libp2p.start()) ) + // create PeerStore and populate peerStore + netArray.forEach((net, i) => { + const pb = netArray[i].libp2p.peerStore + netArray.forEach((net, j) => { + if (i === j) { + return + } + pb.addressBook.set(net.peerId, net.libp2p.multiaddrs) + }) + netArray[i].peerStore = pb + }) + // create every BitSwap netArray.forEach((net) => { net.bitswap = new Bitswap(net.libp2p, net.repo.blocks, net.peerStore) From a2114fa260fb8506baabfdf4c98e42e1f4cedec9 Mon Sep 17 00:00:00 2001 From: Jacob Heun Date: Wed, 27 May 2020 18:57:55 +0200 Subject: [PATCH 2/8] test: add better routing table check test: wait for provide to finish --- package.json | 3 +-- test/bitswap.js | 12 +++++------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/package.json b/package.json index 42400ca8..0d4680c3 100644 --- a/package.json +++ b/package.json @@ -90,8 +90,7 @@ "varint-decoder": "^0.4.0" }, "pre-push": [ - "lint", - "test" + "lint" ], "contributors": [ "David Dias ", diff --git a/test/bitswap.js b/test/bitswap.js index a248f2ae..dc6f7046 100644 --- a/test/bitswap.js +++ b/test/bitswap.js @@ -2,7 +2,6 @@ 'use strict' const { expect } = require('aegir/utils/chai') -const delay = require('delay') const PeerId = require('peer-id') const sinon = require('sinon') const pWaitFor = require('p-wait-for') @@ -147,11 +146,9 @@ describe('bitswap with DHT', function () { // await dht routing table are updated await Promise.all([ pWaitFor(() => nodes[0].libp2pNode._dht.routingTable.size >= 1), - pWaitFor(() => nodes[1].libp2pNode._dht.routingTable.size >= 1) + pWaitFor(() => nodes[1].libp2pNode._dht.routingTable.size >= 2), + pWaitFor(() => nodes[2].libp2pNode._dht.routingTable.size >= 1) ]) - - // Give time to process - await delay(300) }) after(async () => { @@ -164,10 +161,11 @@ describe('bitswap with DHT', function () { it('put a block in 2, get it in 0', async () => { const block = await makeBlock() + const provideSpy = sinon.spy(nodes[2].libp2pNode._dht, 'provide') await nodes[2].bitswap.put(block) - // Give put time to process - await delay(300) + // wait for the DHT to finish providing + await provideSpy.returnValues[0] const blockRetrieved = await nodes[0].bitswap.get(block.cid) expect(block.data).to.eql(blockRetrieved.data) From 5fc188bd8b1b6c2e14a23f78778fcb7e3a557132 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Thu, 28 May 2020 13:09:47 +0200 Subject: [PATCH 3/8] chore: use libp2p0.28rc0 --- package.json | 2 +- test/utils/connect-all.js | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/package.json b/package.json index 0d4680c3..7a58f6a4 100644 --- a/package.json +++ b/package.json @@ -52,7 +52,7 @@ "iso-random-stream": "^1.1.1", "it-all": "^1.0.2", "it-drain": "^1.0.1", - "libp2p": "libp2p/js-libp2p#0.28.x", + "libp2p": "^0.28.0-rc.0", "libp2p-kad-dht": "^0.19.1", "libp2p-mplex": "^0.9.2", "libp2p-secio": "^0.12.1", diff --git a/test/utils/connect-all.js b/test/utils/connect-all.js index a396d4ad..40a34e0d 100644 --- a/test/utils/connect-all.js +++ b/test/utils/connect-all.js @@ -5,7 +5,6 @@ const without = require('lodash.without') module.exports = async (nodes) => { for (const node of nodes) { for (const otherNode of without(nodes, node)) { - // TODO: set addrs await node.libp2pNode.dial(otherNode.bitswap.peerId) } } From b7552af1e89183d0902c54c7da66fd581e6f23b6 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Tue, 2 Jun 2020 17:23:22 +0200 Subject: [PATCH 4/8] chore: address review --- package.json | 3 ++- test/network/gen-bitswap-network.node.js | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/package.json b/package.json index 7a58f6a4..3adbb8ce 100644 --- a/package.json +++ b/package.json @@ -90,7 +90,8 @@ "varint-decoder": "^0.4.0" }, "pre-push": [ - "lint" + "lint", + "test" ], "contributors": [ "David Dias ", diff --git a/test/network/gen-bitswap-network.node.js b/test/network/gen-bitswap-network.node.js index ef48044a..4fb1087d 100644 --- a/test/network/gen-bitswap-network.node.js +++ b/test/network/gen-bitswap-network.node.js @@ -79,7 +79,7 @@ async function exchangeBlocks (nodes, blocksPerNode = 10) { const d = Date.now() // fetch all blocks on every node - await Promise.all(nodes.map(async (node, index) => { + await Promise.all(nodes.map(async (node) => { const bs = await Promise.all(cids.map((cid) => node.bitswap.get(cid))) expect(bs).to.deep.equal(blocks) })) From 06f01b9d958302e27a201e8c1075a38626f34653 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Thu, 4 Jun 2020 15:17:58 +0200 Subject: [PATCH 5/8] feat: use libp2p protocol topology --- package.json | 1 + src/network.js | 25 +++++++++++++++++-------- test/utils/mocks.js | 4 ++++ 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/package.json b/package.json index 3adbb8ce..4e2e4ef8 100644 --- a/package.json +++ b/package.json @@ -82,6 +82,7 @@ "it-length-prefixed": "^3.0.0", "it-pipe": "^1.1.0", "just-debounce-it": "^1.1.0", + "libp2p-interfaces": "^0.3.0", "moving-average": "^1.0.0", "multicodec": "^1.0.0", "multihashing-async": "^0.8.0", diff --git a/src/network.js b/src/network.js index fd79fd16..207a3088 100644 --- a/src/network.js +++ b/src/network.js @@ -3,6 +3,8 @@ const lp = require('it-length-prefixed') const pipe = require('it-pipe') +const MulticodecTopology = require('libp2p-interfaces/src/topology/multicodec-topology') + const Message = require('./types/message') const CONSTANTS = require('./constants') const logger = require('./utils').logger @@ -37,8 +39,15 @@ class Network { this._running = true this.libp2p.handle(this.protocols, this._onConnection) - this.libp2p.connectionManager.on('peer:connect', this._onPeerConnect) - this.libp2p.connectionManager.on('peer:disconnect', this._onPeerDisconnect) + // register protocol with topology + const topology = new MulticodecTopology({ + multicodecs: this.protocols, + handlers: { + onConnect: this._onPeerConnect, + onDisconnect: this._onPeerDisconnect + } + }) + this._registrarId = this.libp2p.registrar.register(topology) // All existing connections are like new ones for us for (const peer of this.libp2p.peerStore.peers.values()) { @@ -54,8 +63,8 @@ class Network { // Unhandle both, libp2p doesn't care if it's not already handled this.libp2p.unhandle(this.protocols) - this.libp2p.connectionManager.removeListener('peer:connect', this._onPeerConnect) - this.libp2p.connectionManager.removeListener('peer:disconnect', this._onPeerDisconnect) + // unregister protocol and handlers + this.libp2p.registrar.unregister(this._registrarId) } /** @@ -92,12 +101,12 @@ class Network { } } - _onPeerConnect (connection) { - this.bitswap._onPeerConnected(connection.remotePeer) + _onPeerConnect (peerId) { + this.bitswap._onPeerConnected(peerId) } - _onPeerDisconnect (connection) { - this.bitswap._onPeerDisconnected(connection.remotePeer) + _onPeerDisconnect (peerId) { + this.bitswap._onPeerDisconnected(peerId) } /** diff --git a/test/utils/mocks.js b/test/utils/mocks.js index 293fc7d1..a0635211 100644 --- a/test/utils/mocks.js +++ b/test/utils/mocks.js @@ -23,6 +23,10 @@ exports.mockLibp2pNode = () => { multiaddrs: [], handle () {}, unhandle () {}, + registrar: { + register () {}, + unregister () {} + }, contentRouting: { provide: async (cid) => {}, // eslint-disable-line require-await findProviders: async (cid, timeout) => { return [] } // eslint-disable-line require-await From 3b9c502b3963bfdad219349e04f7e2f1d1cdf338 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Fri, 5 Jun 2020 15:57:25 +0200 Subject: [PATCH 6/8] chore: use final libp2p release --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 4e2e4ef8..1d550bbc 100644 --- a/package.json +++ b/package.json @@ -52,7 +52,7 @@ "iso-random-stream": "^1.1.1", "it-all": "^1.0.2", "it-drain": "^1.0.1", - "libp2p": "^0.28.0-rc.0", + "libp2p": "^0.28.0", "libp2p-kad-dht": "^0.19.1", "libp2p-mplex": "^0.9.2", "libp2p-secio": "^0.12.1", From 43bdcadda2ba584fec0d78e684e674f1a07e5866 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Fri, 5 Jun 2020 16:30:22 +0200 Subject: [PATCH 7/8] chore: add addresses to address book before dial --- test/network/network.node.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/network/network.node.js b/test/network/network.node.js index e2144d00..bf033a8e 100644 --- a/test/network/network.node.js +++ b/test/network/network.node.js @@ -136,7 +136,7 @@ describe('network', () => { bitswapMockB._receiveError = (err) => deferred.reject(err) - // TODO: set addr + p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) const { stream } = await p2pA.dialProtocol(p2pB.peerId, '/ipfs/bitswap/' + version.num) await pipe( [version.serialize(msg)], From a4a19382bb40b1bc641949b14c9bea4f9b557e87 Mon Sep 17 00:00:00 2001 From: Vasco Santos Date: Fri, 5 Jun 2020 17:26:18 +0200 Subject: [PATCH 8/8] chore: dial to multiaddrs instead of peer-id --- test/bitswap-stats.js | 4 ++-- test/bitswap.js | 16 ++++++++-------- test/network/network.node.js | 18 ++++++++++-------- 3 files changed, 20 insertions(+), 18 deletions(-) diff --git a/test/bitswap-stats.js b/test/bitswap-stats.js index cdf4afda..82dfe9b7 100644 --- a/test/bitswap-stats.js +++ b/test/bitswap-stats.js @@ -169,8 +169,8 @@ describe('bitswap stats', () => { bs2 = bitswaps[1] bs2.start() - libp2pNodes[0].peerStore.addressBook.set(libp2pNodes[1].peerId, libp2pNodes[1].multiaddrs) - await libp2pNodes[0].dial(libp2pNodes[1].peerId) + const ma = `${libp2pNodes[1].multiaddrs[0]}/p2p/${libp2pNodes[1].peerId.toB58String()}` + await libp2pNodes[0].dial(ma) block = await makeBlock() diff --git a/test/bitswap.js b/test/bitswap.js index dc6f7046..727c4097 100644 --- a/test/bitswap.js +++ b/test/bitswap.js @@ -38,12 +38,12 @@ describe('bitswap without DHT', function () { ]) // connect 0 -> 1 && 1 -> 2 - nodes[0].libp2pNode.peerStore.addressBook.set(nodes[1].libp2pNode.peerId, nodes[1].libp2pNode.multiaddrs) - nodes[1].libp2pNode.peerStore.addressBook.set(nodes[2].libp2pNode.peerId, nodes[2].libp2pNode.multiaddrs) + const ma1 = `${nodes[1].libp2pNode.multiaddrs[0]}/p2p/${nodes[1].libp2pNode.peerId.toB58String()}` + const ma2 = `${nodes[2].libp2pNode.multiaddrs[0]}/p2p/${nodes[2].libp2pNode.peerId.toB58String()}` await Promise.all([ - nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerId), - nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerId) + nodes[0].libp2pNode.dial(ma1), + nodes[1].libp2pNode.dial(ma2) ]) }) @@ -135,12 +135,12 @@ describe('bitswap with DHT', function () { ]) // connect 0 -> 1 && 1 -> 2 - nodes[0].libp2pNode.peerStore.addressBook.set(nodes[1].libp2pNode.peerId, nodes[1].libp2pNode.multiaddrs) - nodes[1].libp2pNode.peerStore.addressBook.set(nodes[2].libp2pNode.peerId, nodes[2].libp2pNode.multiaddrs) + const ma1 = `${nodes[1].libp2pNode.multiaddrs[0]}/p2p/${nodes[1].libp2pNode.peerId.toB58String()}` + const ma2 = `${nodes[2].libp2pNode.multiaddrs[0]}/p2p/${nodes[2].libp2pNode.peerId.toB58String()}` await Promise.all([ - nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerId), - nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerId) + nodes[0].libp2pNode.dial(ma1), + nodes[1].libp2pNode.dial(ma2) ]) // await dht routing table are updated diff --git a/test/network/network.node.js b/test/network/network.node.js index bf033a8e..abfef09c 100644 --- a/test/network/network.node.js +++ b/test/network/network.node.js @@ -96,8 +96,8 @@ describe('network', () => { counter++ } - p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) - await p2pA.dial(p2pB.peerId) + const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}` + await p2pA.dial(ma) await pWaitFor(() => counter >= 2) bitswapMockA._onPeerConnected = () => {} @@ -105,8 +105,8 @@ describe('network', () => { }) it('connectTo success', async () => { - p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) - await networkA.connectTo(p2pB.peerId) + const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}` + await networkA.connectTo(ma) }) const versions = [{ @@ -136,8 +136,8 @@ describe('network', () => { bitswapMockB._receiveError = (err) => deferred.reject(err) - p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) - const { stream } = await p2pA.dialProtocol(p2pB.peerId, '/ipfs/bitswap/' + version.num) + const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}` + const { stream } = await p2pA.dialProtocol(ma, '/ipfs/bitswap/' + version.num) await pipe( [version.serialize(msg)], lp.encode(), @@ -172,8 +172,8 @@ describe('network', () => { }) it('dial to peer on Bitswap 1.0.0', async () => { - p2pA.peerStore.addressBook.set(p2pC.peerId, p2pC.multiaddrs) - const { protocol } = await p2pA.dialProtocol(p2pC.peerId, ['/ipfs/bitswap/1.1.0', '/ipfs/bitswap/1.0.0']) + const ma = `${p2pC.multiaddrs[0]}/p2p/${p2pC.peerId.toB58String()}` + const { protocol } = await p2pA.dialProtocol(ma, ['/ipfs/bitswap/1.1.0', '/ipfs/bitswap/1.0.0']) expect(protocol).to.equal('/ipfs/bitswap/1.0.0') }) @@ -212,6 +212,8 @@ describe('network', () => { networkA.start() networkB.start() + // In a real network scenario, peers will be discovered and their addresses + // will be added to the addressBook before bitswap kicks in p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs) const deferred = pDefer()