diff --git a/src/pubsub.js b/src/pubsub.js index 344c2f65..2c067d99 100644 --- a/src/pubsub.js +++ b/src/pubsub.js @@ -46,6 +46,12 @@ module.exports = (node, Pubsub, config) => { } }, + /** + * Publish messages to the given topics. + * @param {Array|string} topic + * @param {Buffer} data + * @returns {Promise} + */ publish: (topic, data) => { if (!node.isStarted() && !pubsub.started) { throw errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED) @@ -60,6 +66,10 @@ module.exports = (node, Pubsub, config) => { return pubsub.publish(topic, data) }, + /** + * Get a list of topics the node is subscribed to. + * @returns {Array} topics + */ getTopics: () => { if (!node.isStarted() && !pubsub.started) { throw errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED) @@ -68,6 +78,11 @@ module.exports = (node, Pubsub, config) => { return pubsub.getTopics() }, + /** + * Get a list of the peer-ids that are subscribed to one topic. + * @param {string} topic + * @returns {Array} + */ getPeersSubscribed: (topic) => { if (!node.isStarted() && !pubsub.started) { throw errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED) diff --git a/test/pubsub/configuration.node.js b/test/pubsub/configuration.node.js index 7ec68b30..829e303f 100644 --- a/test/pubsub/configuration.node.js +++ b/test/pubsub/configuration.node.js @@ -6,12 +6,11 @@ chai.use(require('dirty-chai')) const { expect } = chai const mergeOptions = require('merge-options') - const multiaddr = require('multiaddr') -const PeerInfo = require('peer-info') const { create } = require('../../src') const { baseOptions, subsystemOptions } = require('./utils') +const peerUtils = require('../utils/creators/peer') const listenAddr = multiaddr('/ip4/127.0.0.1/tcp/0') @@ -33,7 +32,7 @@ describe('Pubsub subsystem is configurable', () => { }) it('should start and stop by default once libp2p starts', async () => { - const peerInfo = await PeerInfo.create() + const [peerInfo] = await peerUtils.createPeerInfoFromFixture(1) peerInfo.multiaddrs.add(listenAddr) const customOptions = mergeOptions(subsystemOptions, { @@ -51,7 +50,7 @@ describe('Pubsub subsystem is configurable', () => { }) it('should not start if disabled once libp2p starts', async () => { - const peerInfo = await PeerInfo.create() + const [peerInfo] = await peerUtils.createPeerInfoFromFixture(1) peerInfo.multiaddrs.add(listenAddr) const customOptions = mergeOptions(subsystemOptions, { @@ -71,7 +70,7 @@ describe('Pubsub subsystem is configurable', () => { }) it('should allow a manual start', async () => { - const peerInfo = await PeerInfo.create() + const [peerInfo] = await peerUtils.createPeerInfoFromFixture(1) peerInfo.multiaddrs.add(listenAddr) const customOptions = mergeOptions(subsystemOptions, { diff --git a/test/pubsub/implementations.node.js b/test/pubsub/implementations.node.js index 120ee84d..cb2ee9f2 100644 --- a/test/pubsub/implementations.node.js +++ b/test/pubsub/implementations.node.js @@ -15,10 +15,10 @@ const { multicodec: floodsubMulticodec } = require('libp2p-floodsub') const { multicodec: gossipsubMulticodec } = require('libp2p-gossipsub') const multiaddr = require('multiaddr') -const PeerInfo = require('peer-info') const { create } = require('../../src') const { baseOptions } = require('./utils') +const peerUtils = require('../utils/creators/peer') const listenAddr = multiaddr('/ip4/127.0.0.1/tcp/0') const remoteListenAddr = multiaddr('/ip4/127.0.0.1/tcp/0') @@ -29,10 +29,7 @@ describe('Pubsub subsystem is able to use different implementations', () => { let remAddr beforeEach(async () => { - [peerInfo, remotePeerInfo] = await Promise.all([ - PeerInfo.create(), - PeerInfo.create() - ]) + [peerInfo, remotePeerInfo] = await peerUtils.createPeerInfoFromFixture(2) peerInfo.multiaddrs.add(listenAddr) remotePeerInfo.multiaddrs.add(remoteListenAddr) @@ -79,8 +76,6 @@ describe('Pubsub subsystem is able to use different implementations', () => { remAddr = remoteLibp2p.transportManager.getAddrs()[0] const connection = await libp2p.dialProtocol(remAddr, multicodec) - - await new Promise((resolve) => setTimeout(resolve, 1000)) expect(connection).to.exist() libp2p.pubsub.subscribe(topic, (msg) => { @@ -93,9 +88,8 @@ describe('Pubsub subsystem is able to use different implementations', () => { const subscribedPeers = remoteLibp2p.pubsub.getPeersSubscribed(topic) return subscribedPeers.includes(libp2pId) }) - await new Promise((resolve) => setTimeout(resolve, 1000)) - remoteLibp2p.pubsub.publish(topic, data) + remoteLibp2p.pubsub.publish(topic, data) await defer.promise } }) diff --git a/test/pubsub/operation.node.js b/test/pubsub/operation.node.js index b52d6c13..b7e5a77b 100644 --- a/test/pubsub/operation.node.js +++ b/test/pubsub/operation.node.js @@ -9,12 +9,11 @@ const sinon = require('sinon') const pWaitFor = require('p-wait-for') const pDefer = require('p-defer') const mergeOptions = require('merge-options') - const multiaddr = require('multiaddr') -const PeerInfo = require('peer-info') const { create } = require('../../src') const { subsystemOptions, subsystemMulticodecs } = require('./utils') +const peerUtils = require('../utils/creators/peer') const listenAddr = multiaddr('/ip4/127.0.0.1/tcp/0') const remoteListenAddr = multiaddr('/ip4/127.0.0.1/tcp/0') @@ -25,10 +24,7 @@ describe('Pubsub subsystem operates correctly', () => { let remAddr beforeEach(async () => { - [peerInfo, remotePeerInfo] = await Promise.all([ - PeerInfo.create(), - PeerInfo.create() - ]) + [peerInfo, remotePeerInfo] = await peerUtils.createPeerInfoFromFixture(2) peerInfo.multiaddrs.add(listenAddr) remotePeerInfo.multiaddrs.add(remoteListenAddr) @@ -44,8 +40,10 @@ describe('Pubsub subsystem operates correctly', () => { peerInfo: remotePeerInfo })) - await libp2p.start() - await remoteLibp2p.start() + await Promise.all([ + libp2p.start(), + remoteLibp2p.start() + ]) remAddr = remoteLibp2p.transportManager.getAddrs()[0] })