chore: restructure pubsub tests

.github/workflows/main.yml

@@ -27,7 +27,7 @@ jobs:
     strategy:
       matrix:
         os: [windows-latest, ubuntu-latest, macos-latest]
-        node: [14]
+        node: [14, 16]
       fail-fast: true
     steps:
       - uses: actions/checkout@v2

examples/package.json

@@ -12,6 +12,7 @@
     "fs-extra": "^8.1.0",
     "libp2p-pubsub-peer-discovery": "^4.0.0",
     "libp2p-relay-server": "^0.2.0",
+    "libp2p-gossipsub": "^0.9.1",
     "p-defer": "^3.0.0",
     "which": "^2.0.1"
   },

examples/pubsub/README.md

@@ -8,6 +8,10 @@ We've seen many interesting use cases appear with this, here are some highlights
 - [IPFS PubSub (using libp2p-floodsub) for IoT](https://www.youtube.com/watch?v=qLpM5pBDGiE).
 - [Real Time distributed Applications](https://www.youtube.com/watch?v=vQrbxyDPSXg)
 
+## 0. Set up the example
+
+Before moving into the examples, you should run `npm install` on the top level `js-libp2p` folder, in order to install all the dependencies needed for this example. In addition, you will need to install the example related dependencies by doing `cd examples && npm install`. Once the install finishes, you should move into the example folder with `cd pubsub`.
+
 ## 1. Setting up a simple PubSub network on top of libp2p
 
 For this example, we will use MulticastDNS for automatic Peer Discovery. This example is based the previous examples found in [Discovery Mechanisms](../discovery-mechanisms). You can find the complete version at [1.js](./1.js).

test/pubsub/configuration.node.js → test/configuration/pubsub.spec.js

@@ -3,14 +3,13 @@
 
 const { expect } = require('aegir/utils/chai')
 const mergeOptions = require('merge-options')
-const { Multiaddr } = require('multiaddr')
+const pDefer = require('p-defer')
+const delay = require('delay')
 
 const { create } = require('../../src')
-const { baseOptions, subsystemOptions } = require('./utils')
+const { baseOptions, pubsubSubsystemOptions } = require('./utils')
 const peerUtils = require('../utils/creators/peer')
 
-const listenAddr = new Multiaddr('/ip4/127.0.0.1/tcp/0')
-
 describe('Pubsub subsystem is configurable', () => {
   let libp2p
 
@@ -24,18 +23,15 @@ describe('Pubsub subsystem is configurable', () => {
   })
 
   it('should exist if the module is provided', async () => {
-    libp2p = await create(subsystemOptions)
+    libp2p = await create(pubsubSubsystemOptions)
     expect(libp2p.pubsub).to.exist()
   })
 
   it('should start and stop by default once libp2p starts', async () => {
     const [peerId] = await peerUtils.createPeerId()
 
-    const customOptions = mergeOptions(subsystemOptions, {
-      peerId,
-      addresses: {
-        listen: [listenAddr]
-      }
+    const customOptions = mergeOptions(pubsubSubsystemOptions, {
+      peerId
     })
 
     libp2p = await create(customOptions)
@@ -51,11 +47,8 @@ describe('Pubsub subsystem is configurable', () => {
   it('should not start if disabled once libp2p starts', async () => {
     const [peerId] = await peerUtils.createPeerId()
 
-    const customOptions = mergeOptions(subsystemOptions, {
+    const customOptions = mergeOptions(pubsubSubsystemOptions, {
       peerId,
-      addresses: {
-        listen: [listenAddr]
-      },
       config: {
         pubsub: {
           enabled: false
@@ -73,11 +66,8 @@ describe('Pubsub subsystem is configurable', () => {
   it('should allow a manual start', async () => {
     const [peerId] = await peerUtils.createPeerId()
 
-    const customOptions = mergeOptions(subsystemOptions, {
+    const customOptions = mergeOptions(pubsubSubsystemOptions, {
       peerId,
-      addresses: {
-        listen: [listenAddr]
-      },
       config: {
         pubsub: {
           enabled: false
@@ -93,3 +83,47 @@ describe('Pubsub subsystem is configurable', () => {
     expect(libp2p.pubsub.started).to.equal(true)
   })
 })
+
+describe('Pubsub subscription handlers adapter', () => {
+  let libp2p
+
+  beforeEach(async () => {
+    const [peerId] = await peerUtils.createPeerId()
+
+    libp2p = await create(mergeOptions(pubsubSubsystemOptions, {
+      peerId
+    }))
+
+    await libp2p.start()
+  })
+
+  afterEach(async () => {
+    libp2p && await libp2p.stop()
+  })
+
+  it('extends pubsub with subscribe handler', async () => {
+    let countMessages = 0
+    const topic = 'topic'
+    const defer = pDefer()
+
+    const handler = () => {
+      countMessages++
+      if (countMessages > 1) {
+        throw new Error('only one message should be received')
+      }
+
+      defer.resolve()
+    }
+
+    await libp2p.pubsub.subscribe(topic, handler)
+
+    libp2p.pubsub.emit(topic, 'useless-data')
+    await defer.promise
+
+    await libp2p.pubsub.unsubscribe(topic, handler)
+    libp2p.pubsub.emit(topic, 'useless-data')
+
+    // wait to guarantee that the handler is not called twice
+    await delay(100)
+  })
+})

test/configuration/utils.js

@@ -0,0 +1,52 @@
+'use strict'
+
+const Pubsub = require('libp2p-interfaces/src/pubsub')
+const { NOISE: Crypto } = require('libp2p-noise')
+const Muxer = require('libp2p-mplex')
+const Transport = require('libp2p-websockets')
+const filters = require('libp2p-websockets/src/filters')
+const transportKey = Transport.prototype[Symbol.toStringTag]
+
+const { MULTIADDRS_WEBSOCKETS } = require('../fixtures/browser')
+const relayAddr = MULTIADDRS_WEBSOCKETS[0]
+
+const mergeOptions = require('merge-options')
+
+const baseOptions = {
+  modules: {
+    transport: [Transport],
+    streamMuxer: [Muxer],
+    connEncryption: [Crypto]
+  }
+}
+
+module.exports.baseOptions = baseOptions
+
+class MockPubsub extends Pubsub {
+  constructor (libp2p, options = {}) {
+    super({
+      debugName: 'mock-pubsub',
+      multicodecs: '/mock-pubsub',
+      libp2p,
+      ...options
+    })
+  }
+}
+
+const pubsubSubsystemOptions = mergeOptions(baseOptions, {
+  modules: {
+    pubsub: MockPubsub
+  },
+  addresses: {
+    listen: [`${relayAddr}/p2p-circuit`]
+  },
+  config: {
+    transport: {
+      [transportKey]: {
+        filter: filters.all
+      }
+    }
+  }
+})
+
+module.exports.pubsubSubsystemOptions = pubsubSubsystemOptions