chore: use libp2p as parameter

.aegir.js

@@ -0,0 +1,60 @@
+'use strict'
+
+/**
+ * This file uses aegir hooks to
+ * set up a libp2p instance for browser nodes to relay through
+ * before tests start
+ */
+
+const Libp2p = require('libp2p')
+const PeerId = require('peer-id')
+
+const WS = require('libp2p-websockets')
+const MPLEX = require('libp2p-mplex')
+const { NOISE } = require('libp2p-noise')
+
+const RelayPeer = require('./test/fixtures/relay')
+
+let libp2p
+
+const before = async () => {
+  // Use the last peer
+  const peerId = await PeerId.createFromJSON(RelayPeer)
+
+  libp2p = new Libp2p({
+    addresses: {
+      listen: [RelayPeer.multiaddr]
+    },
+    peerId,
+    modules: {
+      transport: [WS],
+      streamMuxer: [MPLEX],
+      connEncryption: [NOISE]
+    },
+    config: {
+      relay: {
+        enabled: true,
+        hop: {
+          enabled: true,
+          active: false
+        }
+      },
+      pubsub: {
+        enabled: false
+      }
+    }
+  })
+
+  await libp2p.start()
+}
+
+const after = async () => {
+  await libp2p.stop()
+}
+
+module.exports = {
+  hooks: {
+    pre: before,
+    post: after
+  }
+}

package.json

@@ -48,14 +48,17 @@
     "chai-spies": "^1.0.0",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
-    "it-pair": "^1.0.0",
+    "libp2p": "https://github.com/libp2p/js-libp2p#0.29.x",
+    "libp2p-mplex": "^0.9.5",
+    "libp2p-noise": "^1.1.2",
+    "libp2p-websockets": "^0.13.6",
     "lodash": "^4.17.15",
     "multiaddr": "^7.1.0",
     "p-defer": "^3.0.0",
+    "p-wait-for": "^3.1.0",
     "sinon": "^9.0.1"
   },
   "dependencies": {
-    "async.nexttick": "^0.5.2",
     "buffer": "^5.6.0",
     "debug": "^4.1.1",
     "it-pipe": "^1.0.1",
@@ -65,6 +68,9 @@
     "protons": "^1.0.1",
     "time-cache": "^0.3.0"
   },
+  "peerDependencies": {
+    "libp2p": "https://github.com/libp2p/js-libp2p#0.29.x"
+  },
   "contributors": [
     "David Dias <[email protected]>",
     "Vasco Santos <[email protected]>",

src/index.js

@@ -15,52 +15,24 @@ const { multicodec } = require('./config')
 
 const ensureArray = utils.ensureArray
 
-function validateRegistrar (registrar) {
-  if (typeof registrar !== 'object') {
-    throw new Error('a registrar object is required')
-  }
-
-  if (typeof registrar.handle !== 'function') {
-    throw new Error('a handle function must be provided in registrar')
-  }
-
-  if (typeof registrar.register !== 'function') {
-    throw new Error('a register function must be provided in registrar')
-  }
-
-  if (typeof registrar.unregister !== 'function') {
-    throw new Error('a unregister function must be provided in registrar')
-  }
-}
-
 /**
  * FloodSub (aka dumbsub is an implementation of pubsub focused on
  * delivering an API for Publish/Subscribe, but with no CastTree Forming
  * (it just floods the network).
  */
 class FloodSub extends BaseProtocol {
   /**
-   * @param {PeerId} peerId instance of the peer's PeerId
-   * @param {Object} registrar
-   * @param {function} registrar.handle
-   * @param {function} registrar.register
-   * @param {function} registrar.unregister
+   * @param {Libp2p} libp2p instance of libp2p
    * @param {Object} [options]
    * @param {boolean} options.emitSelf if publish should emit to self, if subscribed, defaults to false
    * @constructor
    */
-  constructor (peerId, registrar, options = {}) {
-    if (!PeerId.isPeerId(peerId)) {
-      throw new Error('peerId must be an instance of `peer-id`')
-    }
-
-    validateRegistrar(registrar)
-
+  constructor (libp2p, options = {}) {
     super({
       debugName: debugName,
       multicodecs: multicodec,
-      peerId: peerId,
-      registrar: registrar,
+      peerId: libp2p.peerId,
+      registrar: libp2p.registrar,
       ...options
     })
 

test/2-nodes.spec.js

@@ -6,42 +6,36 @@ const chai = require('chai')
 chai.use(require('dirty-chai'))
 chai.use(require('chai-spies'))
 const expect = chai.expect
+
 const { Buffer } = require('buffer')
 const pDefer = require('p-defer')
+const pWaitFor = require('p-wait-for')
 const times = require('lodash/times')
 
 const FloodSub = require('../src')
-const { multicodec } = require('../src')
 const {
   defOptions,
   first,
-  createPeerId,
-  createMockRegistrar,
-  expectSet,
-  ConnectionPair
+  expectSet
 } = require('./utils')
 
+const { createPeers } = require('./utils/create-peer')
+
 function shouldNotHappen (_) {
   expect.fail()
 }
 
 describe('basics between 2 nodes', () => {
   describe('fresh nodes', () => {
-    let peerIdA, peerIdB
+    let peerA, peerB
     let fsA, fsB
 
-    const registrarRecordA = {}
-    const registrarRecordB = {}
-
     // Mount pubsub protocol
     before(async () => {
-      [peerIdA, peerIdB] = await Promise.all([
-        createPeerId(),
-        createPeerId()
-      ])
+      [peerA, peerB] = await createPeers({ number: 2 })
 
-      fsA = new FloodSub(peerIdA, createMockRegistrar(registrarRecordA), defOptions)
-      fsB = new FloodSub(peerIdB, createMockRegistrar(registrarRecordB), defOptions)
+      fsA = new FloodSub(peerA, defOptions)
+      fsB = new FloodSub(peerB, defOptions)
 
       expect(fsA.peers.size).to.be.eql(0)
       expect(fsA.subscriptions.size).to.eql(0)
@@ -57,34 +51,10 @@ describe('basics between 2 nodes', () => {
 
     // Connect floodsub nodes
     before(async () => {
-      const onConnectA = registrarRecordA[multicodec].onConnect
-      const onConnectB = registrarRecordB[multicodec].onConnect
-      const handleA = registrarRecordA[multicodec].handler
-      const handleB = registrarRecordB[multicodec].handler
-
-      // Notice peers of connection
-      const [c0, c1] = ConnectionPair()
-      await onConnectA(peerIdB, c0)
-      await onConnectB(peerIdA, c1)
-
-      await handleB({
-        protocol: multicodec,
-        stream: c1.stream,
-        connection: {
-          remotePeer: peerIdA
-        }
-      })
-
-      await handleA({
-        protocol: multicodec,
-        stream: c0.stream,
-        connection: {
-          remotePeer: peerIdB
-        }
-      })
+      await peerA.dialProtocol(peerB.peerId, FloodSub.multicodec)
 
-      expect(fsA.peers.size).to.be.eql(1)
-      expect(fsB.peers.size).to.be.eql(1)
+      // Wait for peers to be ready in pubsub
+      await pWaitFor(() => fsA.peers.size === 1 && fsB.peers.size === 1)
     })
 
     after(() => {
@@ -213,21 +183,20 @@ describe('basics between 2 nodes', () => {
   })
 
   describe('nodes send state on connection', () => {
-    let peerIdA, peerIdB
+    let peerA, peerB
     let fsA, fsB
 
-    const registrarRecordA = {}
-    const registrarRecordB = {}
-
     // Mount pubsub protocol
     before(async () => {
-      [peerIdA, peerIdB] = await Promise.all([
-        createPeerId(),
-        createPeerId()
-      ])
+      [peerA, peerB] = await createPeers({ number: 2 })
+
+      fsA = new FloodSub(peerA, defOptions)
+      fsB = new FloodSub(peerB, defOptions)
 
-      fsA = new FloodSub(peerIdA, createMockRegistrar(registrarRecordA), defOptions)
-      fsB = new FloodSub(peerIdB, createMockRegistrar(registrarRecordB), defOptions)
+      expect(fsA.peers.size).to.be.eql(0)
+      expect(fsA.subscriptions.size).to.eql(0)
+      expect(fsB.peers.size).to.be.eql(0)
+      expect(fsB.subscriptions.size).to.eql(0)
     })
 
     // Start pubsub
@@ -255,35 +224,8 @@ describe('basics between 2 nodes', () => {
     })
 
     it('existing subscriptions are sent upon peer connection', async () => {
-      const dial = async () => {
-        const onConnectA = registrarRecordA[multicodec].onConnect
-        const onConnectB = registrarRecordB[multicodec].onConnect
-        const handleA = registrarRecordA[multicodec].handler
-        const handleB = registrarRecordB[multicodec].handler
-
-        // Notice peers of connection
-        const [c0, c1] = ConnectionPair()
-        await onConnectA(peerIdB, c0)
-        await handleB({
-          protocol: multicodec,
-          stream: c1.stream,
-          connection: {
-            remotePeer: peerIdA
-          }
-        })
-
-        await onConnectB(peerIdA, c1)
-        await handleA({
-          protocol: multicodec,
-          stream: c0.stream,
-          connection: {
-            remotePeer: peerIdB
-          }
-        })
-      }
-
       await Promise.all([
-        dial(),
+        peerA.dialProtocol(peerB.peerId, FloodSub.multicodec),
         new Promise((resolve) => fsA.once('floodsub:subscription-change', resolve)),
         new Promise((resolve) => fsB.once('floodsub:subscription-change', resolve))
       ])

test/emit-self.spec.js

@@ -8,21 +8,19 @@ const expect = chai.expect
 const { Buffer } = require('buffer')
 const FloodSub = require('../src')
 
-const {
-  createPeerId, mockRegistrar
-} = require('./utils')
+const { createPeers } = require('./utils/create-peer')
 
 const shouldNotHappen = (_) => expect.fail()
 
 describe('emit self', () => {
   let floodsub
-  let peerId
+  let peer
   const topic = 'Z'
 
   describe('enabled', () => {
     before(async () => {
-      peerId = await createPeerId()
-      floodsub = new FloodSub(peerId, mockRegistrar, { emitSelf: true })
+      [peer] = await createPeers()
+      floodsub = new FloodSub(peer, { emitSelf: true })
     })
 
     before(async () => {
@@ -44,8 +42,8 @@ describe('emit self', () => {
 
   describe('disabled', () => {
     before(async () => {
-      peerId = await createPeerId()
-      floodsub = new FloodSub(peerId, mockRegistrar, { emitSelf: false })
+      [peer] = await createPeers()
+      floodsub = new FloodSub(peer, { emitSelf: false })
     })
 
     before(async () => {