chore: use new version of pubsub pr

Author: vasco-santos

README.md

@@ -29,20 +29,29 @@ Gossipsub is an implementation of pubsub based on meshsub and floodsub. You can
 ```javascript
 const Gossipsub = require('libp2p-gossipsub')
 
-const gsub = new Gossipsub(node)
+const registrar = {
+  handle: (multicodecs, handle) => {
+    // register multicodec to libp2p
+    // handle function is called everytime a remote peer opens a stream to the peer.
+  },
+  register: (multicodecs, handlers) => {
+    // handlers will be used to notify pubsub of peer connection establishment or closing
+  },
+  unregister: (id) => {
 
-gsub.start((err) => {
-  if (err) {
-    console.log('Upsy', err)
   }
-  gsub.on('fruit', (data) => {
-    console.log(data)
-  })
-  gsub.subscribe('fruit')
+}
 
-  gsub.publish('fruit', new Buffer('banana'))
+const gsub = new Gossipsub(peerInfo, registrar, options)
+
+await gsub.start()
+
+gsub.on('fruit', (data) => {
+  console.log(data)
 })
+gsub.subscribe('fruit')
 
+gsub.publish('fruit', new Buffer('banana'))
 ```
 
 ## API
@@ -51,14 +60,16 @@ gsub.start((err) => {
 
 ```js
 const options = {…}
-const gossipsub = new Gossipsub(libp2pNode, options)
+const gossipsub = new Gossipsub(peerInfo, registrar, options)
 ```
 
 Options is an optional object with the following key-value pairs:
 
 * **`fallbackToFloodsub`**: boolean identifying whether the node should fallback to the floodsub protocol, if another connecting peer does not support gossipsub (defaults to **true**).
 * **`emitSelf`**: boolean identifying whether the node should emit to self on publish, in the event of the topic being subscribed (defaults to **false**).
 
+For the remaining API, see https://github.com/libp2p/js-libp2p-pubsub
+
 ## Contribute
 
 This module is actively under development. Please check out the issues and submit PRs!

src/index.js

@@ -16,6 +16,7 @@ class GossipSub extends BasicPubsub {
   /**
    * @param {PeerInfo} peerInfo instance of the peer's PeerInfo
    * @param {Object} registrar
+   * @param {function} registrar.handle
    * @param {function} registrar.register
    * @param {function} registrar.unregister
    * @param {Object} [options]
@@ -28,6 +29,7 @@ class GossipSub extends BasicPubsub {
     assert(PeerInfo.isPeerInfo(peerInfo), 'peer info must be an instance of `peer-info`')
 
     // registrar handling
+    assert(registrar && typeof registrar.handle === 'function', 'a handle function must be provided in registrar')
     assert(registrar && typeof registrar.register === 'function', 'a register function must be provided in registrar')
     assert(registrar && typeof registrar.unregister === 'function', 'a unregister function must be provided in registrar')
 
@@ -588,3 +590,4 @@ class GossipSub extends BasicPubsub {
 }
 
 module.exports = GossipSub
+module.exports.multicodec = constants.GossipSubID

src/pubsub.js

@@ -27,6 +27,7 @@ class BasicPubSub extends Pubsub {
    * @param {string} props.multicodec protocol identificer to connect
    * @param {PeerInfo} props.peerInfo peer's peerInfo
    * @param {Object} props.registrar registrar for libp2p protocols
+   * @param {function} props.registrar.handle
    * @param {function} props.registrar.register
    * @param {function} props.registrar.unregister
    * @param {Object} [props.options]
@@ -82,9 +83,10 @@ class BasicPubSub extends Pubsub {
    * @override
    * @param {PeerInfo} peerInfo peer info
    * @param {Connection} conn connection to the peer
+   * @returns {Promise<void>}
    */
-  _onPeerConnected (peerInfo, conn) {
-    super._onPeerConnected(peerInfo, conn)
+  async _onPeerConnected (peerInfo, conn) {
+    await super._onPeerConnected(peerInfo, conn)
     const idB58Str = peerInfo.id.toB58String()
     const peer = this.peers.get(idB58Str)
 
@@ -110,7 +112,7 @@ class BasicPubSub extends Pubsub {
       await pipe(
         conn,
         lp.decode(),
-        async function collect (source) {
+        async function (source) {
           for await (const data of source) {
             const rpcMsg = Buffer.isBuffer(data) ? data : data.slice()
 

test/2-nodes.spec.js

@@ -7,14 +7,14 @@ chai.use(require('chai-spies'))
 const expect = chai.expect
 
 const { GossipSubID: multicodec } = require('../src/constants')
-const DuplexPair = require('it-pair/duplex')
 
 const {
   createGossipsub,
   createGossipsubNodes,
   createGossipsubConnectedNodes,
   mockRegistrar,
   expectSet,
+  ConnectionPair,
   first
 } = require('./utils')
 
@@ -66,7 +66,7 @@ describe('2 nodes', () => {
       const onConnect1 = registrarRecords[1][multicodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnect0(nodes[1].peerInfo, d0)
       onConnect1(nodes[0].peerInfo, d1)
 
@@ -315,7 +315,7 @@ describe('2 nodes', () => {
       const onConnect1 = registrarRecords[1][multicodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnect0(nodes[1].peerInfo, d0)
       onConnect1(nodes[0].peerInfo, d1)
 

test/floodsub.spec.js

@@ -6,14 +6,15 @@ chai.use(require('dirty-chai'))
 
 const expect = chai.expect
 const times = require('lodash/times')
-const DuplexPair = require('it-pair/duplex')
 
 const { multicodec: floodsubMulticodec } = require('libp2p-floodsub')
 
 const {
   createGossipsub,
   createFloodsubNode,
   expectSet,
+  createMockRegistrar,
+  ConnectionPair,
   first
 } = require('./utils')
 
@@ -22,19 +23,6 @@ const shouldNotHappen = () => expect.fail()
 describe('gossipsub fallbacks to floodsub', () => {
   let registrarRecords = Array.from({ length: 2 })
 
-  const registrar = (registrarRecord) => ({
-    register: (multicodecs, handlers) => {
-      multicodecs.forEach((multicodec) => {
-        registrarRecord[multicodec] = handlers
-      })
-    },
-    unregister: (multicodecs) => {
-      multicodecs.forEach((multicodec) => {
-        delete registrarRecord[multicodec]
-      })
-    }
-  })
-
   describe('basics', () => {
     let nodeGs
     let nodeFs
@@ -43,8 +31,8 @@ describe('gossipsub fallbacks to floodsub', () => {
       registrarRecords[0] = {}
       registrarRecords[1] = {}
 
-      nodeGs = await createGossipsub(registrar(registrarRecords[0]), true)
-      nodeFs = await createFloodsubNode(registrar(registrarRecords[1]), true)
+      nodeGs = await createGossipsub(createMockRegistrar(registrarRecords[0]), true)
+      nodeFs = await createFloodsubNode(createMockRegistrar(registrarRecords[1]), true)
     })
 
     afterEach(async function () {
@@ -63,7 +51,7 @@ describe('gossipsub fallbacks to floodsub', () => {
       expect(onConnectFs).to.exist()
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnectGs(nodeFs.peerInfo, d0)
       onConnectFs(nodeGs.peerInfo, d1)
 
@@ -80,8 +68,8 @@ describe('gossipsub fallbacks to floodsub', () => {
       registrarRecords[0] = {}
       registrarRecords[1] = {}
 
-      nodeGs = await createGossipsub(registrar(registrarRecords[0]), true, { fallbackToFloodsub: false })
-      nodeFs = await createFloodsubNode(registrar(registrarRecords[1]), true)
+      nodeGs = await createGossipsub(createMockRegistrar(registrarRecords[0]), true, { fallbackToFloodsub: false })
+      nodeFs = await createFloodsubNode(createMockRegistrar(registrarRecords[1]), true)
     })
 
     after(async function () {
@@ -120,14 +108,14 @@ describe('gossipsub fallbacks to floodsub', () => {
       registrarRecords[0] = {}
       registrarRecords[1] = {}
 
-      nodeGs = await createGossipsub(registrar(registrarRecords[0]), true)
-      nodeFs = await createFloodsubNode(registrar(registrarRecords[1]), true)
+      nodeGs = await createGossipsub(createMockRegistrar(registrarRecords[0]), true)
+      nodeFs = await createFloodsubNode(createMockRegistrar(registrarRecords[1]), true)
 
       const onConnectGs = registrarRecords[0][floodsubMulticodec].onConnect
       const onConnectFs = registrarRecords[1][floodsubMulticodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnectGs(nodeFs.peerInfo, d0)
       onConnectFs(nodeGs.peerInfo, d1)
     })
@@ -175,14 +163,14 @@ describe('gossipsub fallbacks to floodsub', () => {
       registrarRecords[0] = {}
       registrarRecords[1] = {}
 
-      nodeGs = await createGossipsub(registrar(registrarRecords[0]), true)
-      nodeFs = await createFloodsubNode(registrar(registrarRecords[1]), true)
+      nodeGs = await createGossipsub(createMockRegistrar(registrarRecords[0]), true)
+      nodeFs = await createFloodsubNode(createMockRegistrar(registrarRecords[1]), true)
 
       const onConnectGs = registrarRecords[0][floodsubMulticodec].onConnect
       const onConnectFs = registrarRecords[1][floodsubMulticodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnectGs(nodeFs.peerInfo, d0)
       onConnectFs(nodeGs.peerInfo, d1)
 
@@ -286,16 +274,16 @@ describe('gossipsub fallbacks to floodsub', () => {
       registrarRecords[0] = {}
       registrarRecords[1] = {}
 
-      nodeGs = await createGossipsub(registrar(registrarRecords[0]), true)
-      nodeFs = await createFloodsubNode(registrar(registrarRecords[1]), true)
+      nodeGs = await createGossipsub(createMockRegistrar(registrarRecords[0]), true)
+      nodeFs = await createFloodsubNode(createMockRegistrar(registrarRecords[1]), true)
 
       const onConnectGs = registrarRecords[0][floodsubMulticodec].onConnect
       const onConnectFs = registrarRecords[1][floodsubMulticodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
-      onConnectGs(nodeFs.peerInfo, d0)
-      onConnectFs(nodeGs.peerInfo, d1)
+      const [d0, d1] = ConnectionPair()
+      await onConnectGs(nodeFs.peerInfo, d0)
+      await onConnectFs(nodeGs.peerInfo, d1)
 
       nodeGs.subscribe(topic)
       nodeFs.subscribe(topic)

test/mesh.spec.js

@@ -2,10 +2,12 @@
 /* eslint-env mocha */
 
 const { expect } = require('chai')
-const DuplexPair = require('it-pair/duplex')
 
 const { GossipSubDhi, GossipSubID: multicodec } = require('../src/constants')
-const { createGossipsubNodes } = require('./utils')
+const {
+  createGossipsubNodes,
+  ConnectionPair
+} = require('./utils')
 
 describe('mesh overlay', () => {
   let nodes, registrarRecords
@@ -39,7 +41,7 @@ describe('mesh overlay', () => {
       const onConnectN = registrarRecords[n][multicodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnect0(nodes[n].peerInfo, d0)
       onConnectN(nodes[0].peerInfo, d1)
     }
@@ -66,7 +68,7 @@ describe('mesh overlay', () => {
       const onConnectN = registrarRecords[i][multicodec].onConnect
 
       // Notice peers of connection
-      const [d0, d1] = DuplexPair()
+      const [d0, d1] = ConnectionPair()
       onConnect0(nodes[i].peerInfo, d0)
       onConnectN(nodes[0].peerInfo, d1)
     }

test/multiple-nodes.spec.js

@@ -7,11 +7,11 @@ chai.use(require('dirty-chai'))
 const expect = chai.expect
 const promisify = require('promisify-es6')
 
-const DuplexPair = require('it-pair/duplex')
 const { GossipSubID: multicodec } = require('../src/constants')
 const {
   createGossipsubNodes,
-  expectSet
+  expectSet,
+  ConnectionPair
 } = require('./utils')
 
 describe('multiple nodes (more than 2)', () => {
@@ -40,11 +40,11 @@ describe('multiple nodes (more than 2)', () => {
           const onConnectC = registrarRecords[2][multicodec].onConnect
 
           // Notice peers of connection
-          const [d0, d1] = DuplexPair()
+          const [d0, d1] = ConnectionPair()
           onConnectA(b.peerInfo, d0)
           onConnectB(a.peerInfo, d1)
 
-          const [d2, d3] = DuplexPair()
+          const [d2, d3] = ConnectionPair()
           onConnectB(c.peerInfo, d2)
           onConnectC(b.peerInfo, d3)
         })
@@ -105,11 +105,11 @@ describe('multiple nodes (more than 2)', () => {
           const onConnectC = registrarRecords[2][multicodec].onConnect
 
           // Notice peers of connection
-          const [d0, d1] = DuplexPair()
+          const [d0, d1] = ConnectionPair()
           onConnectA(b.peerInfo, d0)
           onConnectB(a.peerInfo, d1)
 
-          const [d2, d3] = DuplexPair()
+          const [d2, d3] = ConnectionPair()
           onConnectB(c.peerInfo, d2)
           onConnectC(b.peerInfo, d3)
 
@@ -200,11 +200,11 @@ describe('multiple nodes (more than 2)', () => {
         const onConnectC = registrarRecords[2][multicodec].onConnect
 
         // Notice peers of connection
-        const [d0, d1] = DuplexPair()
+        const [d0, d1] = ConnectionPair()
         onConnectA(b.peerInfo, d0)
         onConnectB(a.peerInfo, d1)
 
-        const [d2, d3] = DuplexPair()
+        const [d2, d3] = ConnectionPair()
         onConnectB(c.peerInfo, d2)
         onConnectC(b.peerInfo, d3)
 
@@ -265,19 +265,19 @@ describe('multiple nodes (more than 2)', () => {
         const onConnectE = registrarRecords[4][multicodec].onConnect
 
         // Notice peers of connection
-        const [d0, d1] = DuplexPair()
+        const [d0, d1] = ConnectionPair()
         onConnectA(b.peerInfo, d0)
         onConnectB(a.peerInfo, d1)
 
-        const [d2, d3] = DuplexPair()
+        const [d2, d3] = ConnectionPair()
         onConnectB(c.peerInfo, d2)
         onConnectC(b.peerInfo, d3)
 
-        const [d4, d5] = DuplexPair()
+        const [d4, d5] = ConnectionPair()
         onConnectC(d.peerInfo, d4)
         onConnectD(c.peerInfo, d5)
 
-        const [d6, d7] = DuplexPair()
+        const [d6, d7] = ConnectionPair()
         onConnectD(e.peerInfo, d6)
         onConnectE(d.peerInfo, d7)