Fix pubsub load test progress output. Add check for streams buffering too much. Add bigger timeout for 10k load test so that js-ipfs has enough time to run it.

Author: haadcode

src/pubsub.js

@@ -332,8 +332,14 @@ module.exports = (common) => {
           })
         })
 
-        describe('load tests', () => {
-          it('send/receive 10k messages', (done) => {
+        describe('load tests', function () {
+          // Write the progress to stdout when in Node.js, silent when in the browser
+          const log = process && process.stdout ? (s) => process.stdout.write(s) : () => {}
+
+          it('send/receive 10k messages', function (done) {
+            // js-ipfs is a little slow atm, so make sure we have enough time
+            this.timeout(2 * 60 * 1000)
+
             const expectedString = 'hello'
             const count = 10000
             let sendCount = 0
@@ -350,23 +356,36 @@ module.exports = (common) => {
               expect(err).to.not.exists
 
               const outputProgress = () => {
-                /*
-                process.stdout.write('                                                                        \r')
-                process.stdout.write('Sent: ' + sendCount + ' of ' + count + ', Received: ' + receivedCount + '\r')
-                */
+                log('                                                                         \r')
+                log('Sent: ' + sendCount + ' of ' + count + ', Received: ' + receivedCount + '\r')
               }
 
               subscription.on('data', (d) => {
                 expect(d.data.toString()).to.equal(expectedString)
+
                 receivedCount++
                 outputProgress()
                 if (receivedCount >= count) {
                   const duration = new Date().getTime() - startTime
-                  console.log(`Send/Receive 10k messages took: ${duration} ms, ${Math.floor(count / (duration / 1000))} ops / s`)
+                  log(`Send/Receive 10k messages took: ${duration} ms, ${Math.floor(count / (duration / 1000))} ops / s\n`)
                   subscription.cancel()
                     .then(() => subscription2.cancel())
                     .then(done)
                 }
+
+                // Check for early cancel.
+                // We had bugs in the past where all 10k messages would be
+                // buffered first and only then the receiving stream would
+                // process them. 
+                // This check will make sure, that while the sending peer is
+                // sending messages the receiving peer has started receiving 
+                // and processing them.
+                if (sendCount - (count / 10) > receivedCount) {
+                  sendCount = count
+                  subscription.cancel()
+                    .then(() => subscription2.cancel())
+                    .then(() => done(new Error(`Didn't receive enough messages on time!`)))
+                }
               })
 
               function loop () {
@@ -396,10 +415,8 @@ module.exports = (common) => {
             function loop () {
               if (sendCount < count) {
                 sendCount++
-                /*
-                process.stdout.write('                                               \r')
-                process.stdout.write('Sending messages: ' + sendCount + ' of ' + count + '\r')
-                */
+                log('                                                    \r')
+                log('Sending messages: ' + sendCount + ' of ' + count + '\r')
                 ipfs1.pubsub.publish(topic, expectedString, (err) => {
                   expect(err).to.not.exist
                   process.nextTick(() => loop())
@@ -420,10 +437,8 @@ module.exports = (common) => {
             function loop () {
               if (sendCount < count) {
                 sendCount++
-                /*
-                process.stdout.write('                                               \r')
-                process.stdout.write('Subscribing: ' + sendCount + ' of ' + count + '\r')
-                */
+                log('                                               \r')
+                log('Subscribing: ' + sendCount + ' of ' + count + '\r')
                 ipfs1.pubsub.subscribe(topic, (err, res) => {
                   receivedCount++
                   // First call should go through normally
@@ -447,13 +462,10 @@ module.exports = (common) => {
           it('subscribe/unsubscribe 1k times', (done) => {
             const count = 1000
             let sendCount = 0
-            // let receivedCount = 0
 
             function outputProgress () {
-              /*
-              process.stdout.write('                                                                                 \r')
-              process.stdout.write('Subscribe: ' + sendCount + ' of ' + count + ', Unsubscribe: ' + receivedCount + '\r')
-              */
+              log('                                                         \r')
+              log('Subscribe/Unsubscribe: ' + sendCount + ' of ' + count + '\r')
             }
 
             function loop () {
@@ -464,7 +476,6 @@ module.exports = (common) => {
                   expect(err).to.not.exist
                   subscription.cancel((err) => {
                     expect(err).to.not.exist
-                    // receivedCount++
                     outputProgress()
                     process.nextTick(() => loop())
                   })