fix!: subscribe to topic separately from consuming messages

package.json

@@ -133,15 +133,15 @@
     "release": "aegir release"
   },
   "dependencies": {
-    "@libp2p/daemon-client": "^4.1.0",
+    "@libp2p/daemon-client": "^5.0.0",
     "@libp2p/interface-peer-id": "^2.0.1",
     "@libp2p/interface-peer-info": "^1.0.7",
     "@multiformats/multiaddr": "^11.4.0",
     "it-all": "^2.0.0",
     "it-first": "^2.0.0",
     "it-handshake": "^4.1.2",
     "it-length-prefixed": "^8.0.4",
-    "it-pb-stream": "^2.0.3",
+    "it-pb-stream": "^3.0.0",
     "it-pipe": "^2.0.4",
     "it-stream-types": "^1.0.5",
     "multiformats": "^11.0.0",

src/pubsub/floodsub.ts

@@ -52,9 +52,9 @@ function runFloodsubTests (factory: DaemonFactory, optionsA: SpawnOptions, optio
       const data = uint8ArrayFromString('test-data')
       const [peerA, peerB] = daemons
 
-      const subscribeIterator = peerB.client.pubsub.subscribe(topic)
+      const subscription = await peerB.client.pubsub.subscribe(topic)
       const subscriber = async (): Promise<void> => {
-        const message = await first(subscribeIterator)
+        const message = await first(subscription.messages())
 
         expect(message).to.exist()
         expect(message).to.have.property('data').that.equalBytes(data)

src/pubsub/gossipsub.ts

@@ -33,8 +33,9 @@ function runGossipsubTests (factory: DaemonFactory, optionsA: SpawnOptions, opti
         factory.spawn(optionsB)
       ])
 
-      const identify1 = await daemons[1].client.identify()
-      await daemons[0].client.connect(identify1.peerId, identify1.addrs)
+      const [peerA, peerB] = daemons
+      const identifyB = await peerB.client.identify()
+      await peerA.client.connect(identifyB.peerId, identifyB.addrs)
     })
 
     // Stop daemons
@@ -49,18 +50,19 @@ function runGossipsubTests (factory: DaemonFactory, optionsA: SpawnOptions, opti
     it(`${optionsA.type} peer to ${optionsB.type} peer`, async function () {
       const topic = 'test-topic'
       const data = uint8ArrayFromString('test-data')
+      const [peerA, peerB] = daemons
 
-      const subscribeIterator = daemons[1].client.pubsub.subscribe(topic)
+      const subscription = await peerB.client.pubsub.subscribe(topic)
       const subscriber = async (): Promise<void> => {
-        const message = await first(subscribeIterator)
+        const message = await first(subscription.messages())
 
         expect(message).to.exist()
         expect(message).to.have.property('data').that.equalBytes(data)
       }
 
       const publisher = async (): Promise<void> => {
-        await waitForBothSubscribed(topic, daemons[0], daemons[1])
-        await daemons[0].client.pubsub.publish(topic, data)
+        await waitForBothSubscribed(topic, peerA, peerB)
+        await peerA.client.pubsub.publish(topic, data)
       }
 
       return await Promise.all([

src/pubsub/hybrid.ts

@@ -33,8 +33,9 @@ function runHybridTests (factory: DaemonFactory, optionsA: SpawnOptions, options
         factory.spawn(optionsB)
       ])
 
-      const identify1 = await daemons[1].client.identify()
-      await daemons[0].client.connect(identify1.peerId, identify1.addrs)
+      const [peerA, peerB] = daemons
+      const identifyB = await peerB.client.identify()
+      await peerA.client.connect(identifyB.peerId, identifyB.addrs)
     })
 
     // Stop daemons
@@ -49,18 +50,19 @@ function runHybridTests (factory: DaemonFactory, optionsA: SpawnOptions, options
     it(`${optionsA.type} peer to ${optionsB.type} peer`, async function () {
       const topic = 'test-topic'
       const data = uint8ArrayFromString('test-data')
+      const [peerA, peerB] = daemons
 
-      const subscribeIterator = daemons[1].client.pubsub.subscribe(topic)
+      const subscription = await peerB.client.pubsub.subscribe(topic)
       const subscriber = async (): Promise<void> => {
-        const message = await first(subscribeIterator)
+        const message = await first(subscription.messages())
 
         expect(message).to.exist()
         expect(message).to.have.property('data').that.equalBytes(data)
       }
 
       const publisher = async (): Promise<void> => {
-        await waitForBothSubscribed(topic, daemons[0], daemons[1])
-        await daemons[0].client.pubsub.publish(topic, data)
+        await waitForBothSubscribed(topic, peerA, peerB)
+        await peerA.client.pubsub.publish(topic, data)
       }
 
       return await Promise.all([

src/pubsub/utils.ts

@@ -2,6 +2,9 @@ import pWaitFor from 'p-wait-for'
 import type { Daemon } from '..'
 
 export async function waitForBothSubscribed (topic: string, a: Daemon, b: Daemon): Promise<void> {
+  await a.client.pubsub.subscribe(topic)
+  await b.client.pubsub.subscribe(topic)
+
   const idA = await a.client.identify()
   const idB = await b.client.identify()