Fix "not subscribed" errors in multi-node tests

Author: haadcode

src/pubsub.js

@@ -55,7 +55,6 @@ module.exports = (common) => {
               return callback(err)
             }
 
-            console.log('->', peers)
             const hasAllPeers = peersToWait
                                   .map((e) => peers.includes(e))
                                   .filter((e) => e === false)
@@ -134,6 +133,18 @@ module.exports = (common) => {
       })
 
       describe('.peers', () => {
+        before((done) => {
+          ipfs2.id((err, id) => {
+            expect(err).to.not.exist
+            const ipfs2Addr = id.addresses[0]
+            ipfs1.swarm.connect(ipfs2Addr, (err) => {
+              expect(err).to.not.exist
+              // We need to fix this on libp2p level
+              setTimeout(done, 3000)
+            })
+          })
+        })
+
         // TODO clarify what is the goal of pubsub.peers
         it('returns an error when not subscribed to a topic', (done) => {
           ipfs1.pubsub.peers(topic, (err, peers) => {
@@ -143,7 +154,7 @@ module.exports = (common) => {
           })
         })
 
-        it.skip('returns no peers within 10 seconds', (done) => {
+        it('returns no peers within 10 seconds', (done) => {
           // Currently go-ipfs returns peers that have not been
           // subscribed to the topic. Enable when go-ipfs has been fixed
           ipfs1.pubsub.subscribe(topic, (err, subscription) => {
@@ -159,7 +170,7 @@ module.exports = (common) => {
           })
         })
 
-        it.skip('doesn\'t return extra peers', (done) => {
+        it('doesn\'t return extra peers', (done) => {
           // Currently go-ipfs returns peers that have not been
           // subscribed to the topic. Enable when go-ipfs has been fixed
           ipfs1.pubsub.subscribe(topic, (err, subscription1) => {
@@ -182,10 +193,16 @@ module.exports = (common) => {
           })
         })
 
-        it.skip('returns peers for a topic - one peer', (done) => {
+        it('returns peers for a topic - one peer', (done) => {
           // Currently go-ipfs returns peers that have not been subscribed to the topic
           // Enable when go-ipfs has been fixed
           const peersToWait = [ipfs2.peerId]
+          let subscription2
+
+          ipfs1.pubsub.subscribe(topic, (err, subscription) => {
+            expect(err).to.not.exist
+            subscription2 = subscription
+          })
 
           ipfs2.pubsub.subscribe(topic, (err, subscription) => {
             expect(err).to.not.exist
@@ -197,8 +214,6 @@ module.exports = (common) => {
                   done(err)
                 }
 
-                console.log(peers)
-
                 const hasAllPeers = peersToWait
                   .map((e) => peers.indexOf(e) !== -1)
                   .filter((e) => e === false)
@@ -207,7 +222,9 @@ module.exports = (common) => {
                 if (hasAllPeers) {
                   clearInterval(i)
                   expect(peers.length).to.equal(peersToWait.length)
-                  subscription.cancel(done)
+                  subscription.cancel()
+                    .then(() => subscription2.cancel())
+                    .then(done)
                 }
               })
             }, 1000)
@@ -264,21 +281,30 @@ module.exports = (common) => {
             const ipfs2Addr = id.addresses[0]
             ipfs1.swarm.connect(ipfs2Addr, (err) => {
               expect(err).to.not.exist
+              // We need to fix this on libp2p level
               setTimeout(done, 3000)
             })
           })
         })
 
         it('receive messages from different node', (done) => {
           const expectedString = 'hello from the other side'
+          let subscription2
+
+          ipfs2.pubsub.subscribe(topic, (err, subscription) => {
+            expect(err).to.not.exist
+            subscription2 = subscription
+          })
 
           ipfs1.pubsub.subscribe(topic, (err, subscription) => {
             expect(err).to.not.exist
             expect(subscription).to.exist
 
             subscription.on('data', (d) => {
               expect(d.data).to.be.equal(expectedString)
-              subscription.cancel(done)
+              subscription.cancel()
+                .then(() => subscription2.cancel())
+                .then(done)
             })
 
             waitForPeers(ipfs2, [ipfs1.peerId], (err) => {
@@ -293,6 +319,12 @@ module.exports = (common) => {
         it('receive multiple messages', (done) => {
           let receivedMessages = []
           const expectedMessages = 2
+          let subscription2
+
+          ipfs2.pubsub.subscribe(topic, (err, subscription) => {
+            expect(err).to.not.exist
+            subscription2 = subscription
+          })
 
           ipfs1.pubsub.subscribe(topic, (err, subscription) => {
             expect(err).to.not.exists
@@ -303,7 +335,9 @@ module.exports = (common) => {
                 receivedMessages.forEach((msg) => {
                   expect(msg).to.be.equal('hi')
                 })
-                subscription.cancel(done)
+                subscription.cancel()
+                  .then(() => subscription2.cancel())
+                  .then(done)
               }
             })
 
@@ -317,12 +351,30 @@ module.exports = (common) => {
       })
 
       describe('load tests', () => {
+        before((done) => {
+          ipfs2.id((err, id) => {
+            expect(err).to.not.exist
+            const ipfs2Addr = id.addresses[0]
+            ipfs1.swarm.connect(ipfs2Addr, (err) => {
+              expect(err).to.not.exist
+              // We need to fix this on libp2p level
+              setTimeout(done, 3000)
+            })
+          })
+        })
+
         it('send/receive 10k messages', (done) => {
           const expectedString = 'hello'
           const count = 10000
           let sendCount = 0
           let receivedCount = 0
           let startTime
+          let subscription2
+
+          ipfs2.pubsub.subscribe(topic, (err, subscription) => {
+            expect(err).to.not.exists
+            subscription2 = subscription
+          })
 
           ipfs1.pubsub.subscribe(topic, (err, subscription) => {
             expect(err).to.not.exists
@@ -340,7 +392,9 @@ module.exports = (common) => {
                 const duration = new Date().getTime() - startTime
                 process.stdout.write('                                                                        \r')
                 console.log(`Send/Receive 10k messages took: ${duration} ms, ${Math.floor(count / (duration / 1000))} ops / s`)
-                subscription.cancel(done)
+                subscription.cancel()
+                  .then(() => subscription2.cancel())
+                  .then(done)
               }
             })