chore: add pubsub tests

* chore: update deps

* chore: add prepare script

CHANGELOG.md

@@ -1,3 +1,7 @@
+## [0.8.3](https://github.com/libp2p/js-interfaces/compare/v0.8.2...v0.8.3) (2021-01-26)
+
+
+
 ## [0.8.2](https://github.com/libp2p/js-interfaces/compare/v0.8.1...v0.8.2) (2021-01-20)
 
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p-interfaces",
-  "version": "0.8.2",
+  "version": "0.8.3",
   "description": "Interfaces for JS Libp2p",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
@@ -24,6 +24,7 @@
   "scripts": {
     "lint": "aegir lint",
     "build": "aegir build",
+    "prepare": "aegir build --no-bundle",
     "test": "aegir test",
     "test:node": "aegir test --target node",
     "test:browser": "aegir test --target browser",
@@ -51,33 +52,33 @@
     "abortable-iterator": "^3.0.0",
     "chai": "^4.2.0",
     "chai-checkmark": "^1.0.1",
-    "debug": "^4.1.1",
-    "delay": "^4.3.0",
+    "debug": "^4.3.1",
+    "delay": "^4.4.0",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
-    "err-code": "^2.0.0",
-    "it-goodbye": "^2.0.1",
+    "err-code": "^2.0.3",
+    "it-goodbye": "^2.0.2",
     "it-length-prefixed": "^3.1.0",
     "it-pair": "^1.0.0",
     "it-pipe": "^1.1.0",
     "it-pushable": "^1.4.0",
-    "libp2p-crypto": "^0.18.0",
+    "libp2p-crypto": "^0.19.0",
     "libp2p-tcp": "^0.15.0",
-    "multiaddr": "^8.0.0",
-    "multibase": "^3.0.0",
-    "multihashes": "^3.0.1",
+    "multiaddr": "^8.1.2",
+    "multibase": "^3.1.1",
+    "multihashes": "^3.1.1",
     "p-defer": "^3.0.0",
-    "p-limit": "^2.3.0",
-    "p-wait-for": "^3.1.0",
-    "peer-id": "^0.14.0",
+    "p-limit": "^3.1.0",
+    "p-wait-for": "^3.2.0",
+    "peer-id": "^0.14.2",
     "protons": "^2.0.0",
-    "sinon": "^9.0.2",
-    "streaming-iterables": "^5.0.2",
-    "uint8arrays": "^1.1.0"
+    "sinon": "^9.2.4",
+    "streaming-iterables": "^5.0.4",
+    "uint8arrays": "^2.0.5"
   },
   "devDependencies": {
     "aegir": "^29.2.0",
-    "it-handshake": "^1.0.1",
+    "it-handshake": "^1.0.2",
     "rimraf": "^3.0.2"
   },
   "contributors": [

src/pubsub/peer-streams.js

@@ -6,8 +6,6 @@ const EventEmitter = require('events')
 
 const lp = require('it-length-prefixed')
 
-/** @type {typeof import('it-pushable').default} */
-// @ts-ignore
 const pushable = require('it-pushable')
 const { pipe } = require('it-pipe')
 const { source: abortable } = require('abortable-iterator')

src/pubsub/tests/multiple-nodes.js

@@ -307,9 +307,71 @@ module.exports = (common) => {
         // line
         // ◉────◎────◉
         // a    b    c
+        // A subscribes and C publishes
 
-        before(() => { })
-        after(() => { })
+        let psA, psB, psC
+
+        // Create and start pubsub nodes
+        beforeEach(async () => {
+          [psA, psB, psC] = await common.setup(3)
+
+          // Start pubsub mpdes
+          ;[psA, psB, psC].map((p) => p.start())
+        })
+
+        // Connect nodes
+        beforeEach(async () => {
+          await psA._libp2p.dial(psB.peerId)
+          await psB._libp2p.dial(psC.peerId)
+
+          // Wait for peers to be ready in pubsub
+          await pWaitFor(() =>
+            psA.peers.size === 1 &&
+            psC.peers.size === 1 &&
+            psA.peers.size === 1
+          )
+        })
+
+        afterEach(async () => {
+          sinon.restore()
+
+          ;[psA, psB, psC].map((p) => p.stop())
+          await common.teardown()
+        })
+
+        it('exchange pubsub messages', async () => {
+          const defer = pDefer()
+          let counter = 0
+          const topic = 'Z'
+
+          function incMsg (msg) {
+            expect(uint8ArrayToString(msg.data)).to.equal('hey')
+            check()
+          }
+
+          function check () {
+            if (++counter === 1) {
+              psA.removeListener(topic, incMsg)
+              psB.removeListener(topic, incMsg)
+              psC.removeListener(topic, incMsg)
+              defer.resolve()
+            }
+          }
+
+          psA.on(topic, incMsg)
+          psA.subscribe(topic)
+
+          // await subscription change
+          await Promise.all([
+            new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve(null)))
+          ])
+
+          // await a cycle
+          await delay(1000)
+          psC.publish(topic, uint8ArrayFromString('hey'))
+
+          return defer.promise
+        })
       })
 
       describe('1 level tree', () => {

src/pubsub/utils.js

@@ -111,10 +111,10 @@ exports.normalizeInRpcMessage = (message, peerId) => {
  */
 exports.normalizeOutRpcMessage = (message) => {
   const m = Object.assign({}, message)
-  if (typeof message.from === 'string' || message.from instanceof String) {
+  if (typeof message.from === 'string') {
     m.from = uint8ArrayFromString(message.from, 'base58btc')
   }
-  if (typeof message.data === 'string' || message.data instanceof String) {
+  if (typeof message.data === 'string') {
     m.data = uint8ArrayFromString(message.data)
   }
   return m

src/stream-muxer/tests/spawner.js

@@ -4,7 +4,7 @@ const { expect } = require('chai')
 const pair = require('it-pair/duplex')
 const { pipe } = require('it-pipe')
 
-const pLimit = require('p-limit').default
+const pLimit = require('p-limit')
 const { collect, tap, consume } = require('streaming-iterables')
 
 module.exports = async (Muxer, nStreams, nMsg, limit) => {