feat: use libp2p protocol topology

BREAKING CHANGE: bitswap start and stop return a promise

Author: vasco-santos

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",

src/index.js

@@ -360,23 +360,23 @@ class Bitswap {
   /**
    * Start the bitswap node.
    *
-   * @returns {void}
+   * @returns {Promise<void>}
    */
-  start () {
+  async start () {
     this.wm.start()
-    this.network.start()
+    await this.network.start()
     this.engine.start()
   }
 
   /**
    * Stop the bitswap node.
    *
-   * @returns {void}
+   * @returns {Promise<void>}
    */
-  stop () {
+  async stop () {
     this._stats.stop()
     this.wm.stop()
-    this.network.stop()
+    await this.network.stop()
     this.engine.stop()
   }
 }

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
@@ -33,12 +35,19 @@ class Network {
     this._onConnection = this._onConnection.bind(this)
   }
 
-  start () {
+  async start () {
     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 = await this.libp2p.registrar.register(topology)
 
     // All existing connections are like new ones for us
     for (const peer of this.libp2p.peerStore.peers.values()) {
@@ -48,14 +57,14 @@ class Network {
     }
   }
 
-  stop () {
+  async stop () {
     this._running = false
 
     // 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
+    await 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)
   }
 
   /**

test/bitswap-mock-internals.js

@@ -51,7 +51,7 @@ describe('bitswap with mocks', function () {
   describe('receive message', () => {
     it('simple block message', async () => {
       const bs = new Bitswap(mockLibp2pNode(), repo.blocks)
-      bs.start()
+      await bs.start()
 
       const other = ids[1]
 
@@ -85,7 +85,7 @@ describe('bitswap with mocks', function () {
 
     it('simple want message', async () => {
       const bs = new Bitswap(mockLibp2pNode(), repo.blocks)
-      bs.start()
+      await bs.start()
 
       const other = ids[1]
       const b1 = blocks[0]
@@ -110,7 +110,7 @@ describe('bitswap with mocks', function () {
       this.timeout(80 * 1000)
       const bs = new Bitswap(mockLibp2pNode(), repo.blocks)
 
-      bs.start()
+      await bs.start()
 
       const others = await makePeerId(5)
       const blocks = await makeBlock(10)
@@ -139,7 +139,7 @@ describe('bitswap with mocks', function () {
 
     it('ignore unwanted blocks', async () => {
       const bs = new Bitswap(mockLibp2pNode(), repo.blocks)
-      bs.start()
+      await bs.start()
 
       const other = ids[1]
 
@@ -238,7 +238,7 @@ describe('bitswap with mocks', function () {
       bs.network = net
       bs.wm.network = net
       bs.engine.network = net
-      bs.start()
+      await bs.start()
       const get = bs.get(block.cid)
 
       setTimeout(() => {
@@ -317,13 +317,13 @@ describe('bitswap with mocks', function () {
       // Create and start bs1
       const bs1 = new Bitswap(mockLibp2pNode(), repo.blocks)
       applyNetwork(bs1, n1)
-      bs1.start()
+      await bs1.start()
 
       // Create and start bs2
       const repo2 = await createTempRepo()
       const bs2 = new Bitswap(mockLibp2pNode(), repo2.blocks)
       applyNetwork(bs2, n2)
-      bs2.start()
+      await bs2.start()
 
       bs1._onPeerConnected(other)
       bs2._onPeerConnected(me)
@@ -443,7 +443,7 @@ describe('bitswap with mocks', function () {
   describe('unwant', () => {
     it('removes blocks that are wanted multiple times', async () => {
       const bs = new Bitswap(mockLibp2pNode(), repo.blocks)
-      bs.start()
+      await bs.start()
 
       const b = blocks[12]
       const p = Promise.all([

test/bitswap-stats.js

@@ -65,7 +65,7 @@ describe('bitswap stats', () => {
     bs.wm.wantBlocks(blocks.map(b => b.cid))
 
     // start the first bitswap
-    bs.start()
+    await bs.start()
   })
 
   after(async () => {
@@ -167,7 +167,7 @@ describe('bitswap stats', () => {
 
     before(async () => {
       bs2 = bitswaps[1]
-      bs2.start()
+      await bs2.start()
 
       libp2pNodes[0].peerStore.addressBook.set(libp2pNodes[1].peerId, libp2pNodes[1].multiaddrs)
       await libp2pNodes[0].dial(libp2pNodes[1].peerId)

test/bitswap.js

@@ -21,7 +21,7 @@ async function createThing (dht) {
     DHT: dht
   })
   const bitswap = new Bitswap(libp2pNode, repo.blocks)
-  bitswap.start()
+  await bitswap.start()
   return { repo, libp2pNode, bitswap }
 }
 

test/network/gen-bitswap-network.node.js

@@ -66,7 +66,7 @@ async function exchangeBlocks (nodes, blocksPerNode = 10) {
 
   // put blocksPerNode amount of blocks per node
   await Promise.all(nodes.map(async (node, i) => {
-    node.bitswap.start()
+    await node.bitswap.start()
 
     const data = range(blocksPerNode).map((j) => {
       const index = i * blocksPerNode + j

test/network/network.node.js

@@ -59,7 +59,7 @@ describe('network', () => {
     _onPeerDisconnected: async () => {}
   }
 
-  it('instantiate the network obj', () => {
+  it('instantiate the network obj', async () => {
     networkA = new Network(p2pA, bitswapMockA)
     networkB = new Network(p2pB, bitswapMockB)
     // only bitswap100
@@ -69,9 +69,9 @@ describe('network', () => {
     expect(networkB).to.exist()
     expect(networkC).to.exist()
 
-    networkA.start()
-    networkB.start()
-    networkC.start()
+    await networkA.start()
+    await networkB.start()
+    await networkC.start()
   })
 
   it('connectTo fail', async () => {
@@ -209,8 +209,8 @@ describe('network', () => {
     networkB = new Network(p2pB, bitswapMockB)
     networkB.protocols = ['/ipfs/bitswap/1.2.0']
 
-    networkA.start()
-    networkB.start()
+    await networkA.start()
+    await networkB.start()
 
     p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs)
 

test/utils/create-bitswap.js

@@ -15,6 +15,6 @@ module.exports = async () => {
     }
   })
   const bitswap = new Bitswap(libp2pNode, repo.blocks)
-  bitswap.start()
+  await bitswap.start()
   return { bitswap, repo, libp2pNode }
 }

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