Better tests + implementation

Author: victorb

src/api/pubsub.js

@@ -7,6 +7,21 @@ var Stream = require('stream')
 // const Wreck = require('wreck')
 var http = require('http')
 
+let activeSubscriptions = []
+
+const subscriptionExists = (subscriptions, topic) => {
+  return subscriptions.indexOf(topic) !== -1
+}
+const removeSubscription = (subscriptions, topic) => {
+  const indexToRemove = subscriptions.indexOf(topic)
+  return subscriptions.filter((el, index) => {
+    return index !== indexToRemove
+  })
+}
+const addSubscription = (subscriptions, topic) => {
+  return subscriptions.concat([topic])
+}
+
 module.exports = (send) => {
   const api = {
     sub: promisify((topic, options, callback) => {
@@ -18,12 +33,17 @@ module.exports = (send) => {
         options = {}
       }
 
-
       var rs = new Stream()
       rs.readable = true
 
-      console.log('Sub', topic)
-      http.get({
+      if (!subscriptionExists(activeSubscriptions, topic)) {
+        activeSubscriptions = addSubscription(activeSubscriptions, topic)
+      } else {
+        return callback(new Error('Already subscribed to ' + topic), null)
+      }
+
+      // we're using http.get here to have more control
+      const request = http.get({
         host: 'localhost',
         port: 5001,
         path: '/api/v0/pubsub/sub/' + topic
@@ -32,7 +52,7 @@ module.exports = (send) => {
           var parsed = JSON.parse(d)
 
           // skip "double subscription" error
-          if(!parsed.Message) {
+          if (!parsed.Message) {
             parsed.from = bs58.encode(parsed.from)
             parsed.data = Base64.decode(parsed.data)
             parsed.seqno = Base64.decode(parsed.seqno)
@@ -42,21 +62,17 @@ module.exports = (send) => {
         response.on('end', function () {
           rs.emit('end')
         })
-
-        callback(null, rs)
+        rs.cancel = () => {
+          request.abort()
+          response.destroy()
+          activeSubscriptions = removeSubscription(activeSubscriptions, topic)
+        }
       })
-
-      // send({
-      //   path: 'pubsub/sub/' + topic
-      // }, (err, response) => {
-      //   console.log('RESULT', err, response)
-      //   if (err) {
-      //     return callback(err)
-      //   }
-
-      //   callback(null, response.pipe(ndjson.parse()))
-      //   // callback(null, result) // result is a Stream
-      // })
+      rs.cancel = () => {
+        request.abort()
+        activeSubscriptions = removeSubscription(activeSubscriptions, topic)
+      }
+      callback(null, rs)
     }),
     pub: promisify((topic, data, options, callback) => {
       if (typeof options === 'function') {
@@ -81,7 +97,6 @@ module.exports = (send) => {
         if (err) {
           return callback(err)
         }
-
         callback(null, true)
       })
     })

test/ipfs-api/pubsub.spec.js

@@ -9,6 +9,16 @@ const ipfsApi = require('../../src')
 
 const useLocalDaemon = true
 
+const topicName = 'js-ipfs-api-tests'
+
+const publish = (ipfs, data, callback) => {
+  ipfs.pubsub.pub(topicName, data, (err, result) => {
+    expect(err).to.not.exist
+    expect(result).to.equal(true)
+    callback()
+  })
+}
+
 describe('.pubsub', () => {
   if (!isNode) {
     return
@@ -33,40 +43,62 @@ describe('.pubsub', () => {
   //   // fc.dismantle(done)
   // })
 
-  it.only('sub', (done) => {
-    ipfs.pubsub.sub('testi1', (err, result) => {
-      expect(err).to.not.exist
-      result.on('data', function (d) {
-        // console.log("-->", d)
-        expect(d.data).to.equal('hi')
-        done()
-      })
-      result.on('end', function () {
-        console.log("END!!")
-      })
+  describe('.publish', () => {
+    it('message from string', (done) => {
+      publish(ipfs, 'hello friend', done)
+    })
+    it('message from buffer', (done) => {
+      publish(ipfs, new Buffer('hello friend'), done)
     })
-    setTimeout(() => {
-      ipfs.pubsub.pub('testi1', 'hi')
-    }, 100)
   })
 
-  describe('.pub', () => {
-    it.only('publishes a message - from string', (done) => {
-      const data = 'hello friend'
-      ipfs.pubsub.pub('testi1', data, (err, result) => {
+  describe('.subscribe', () => {
+    it('one topic', (done) => {
+      ipfs.pubsub.sub(topicName, (err, subscription) => {
         expect(err).to.not.exist
-        expect(result).to.equal(true)
-        done()
+        subscription.on('data', (d) => {
+          expect(d.data).to.equal('hi')
+          subscription.cancel()
+        })
+        subscription.on('end', () => {
+          done()
+        })
       })
+      setTimeout(publish.bind(null, ipfs, 'hi', () => {}), 0)
+    })
+    it('fails when already subscribed', (done) => {
+      ipfs.pubsub.sub(topicName, (firstErr, firstSub) => {
+        expect(firstErr).to.not.exist
+        ipfs.pubsub.sub(topicName, (secondErr, secondSub) => {
+          expect(secondErr).to.be.an('error')
+          expect(secondErr.toString()).to.equal('Error: Already subscribed to ' + topicName)
+          firstSub.cancel()
+          done()
+        }).catch(done)
+      }).catch(done)
     })
+    it('receive multiple messages', (done) => {
+      let receivedMessages = []
+      let interval = null
+      const expectedMessages = 2
+      ipfs.pubsub.sub(topicName, (err, subscription) => {
+        expect(err).to.not.exists
+        subscription.on('data', (d) => {
+          receivedMessages.push(d.data)
+          if (receivedMessages.length === expectedMessages) {
+            receivedMessages.forEach((msg) => {
+              expect(msg).to.be.equal('hi')
+            })
+            clearInterval(interval)
+            subscription.cancel()
+            done()
+          }
+        })
+      }).catch(done)
 
-    it.only('publishes a message - from Buffer', (done) => {
-      const data = new Buffer('hello buffer')
-      ipfs.pubsub.pub('testi1', data, (err, result) => {
-        expect(err).to.not.exist
-        expect(result).to.equal(true)
-        done()
-      })
+      setTimeout(() => {
+        interval = setInterval(publish.bind(null, ipfs, 'hi', () => {}), 10)
+      }, 10)
     })
   })
 })