refactor to new event based pubsub api

Author: dignifiedquire

package.json

@@ -34,7 +34,6 @@
     "ipld-dag-pb": "^0.9.3",
     "is-ipfs": "^0.2.1",
     "isstream": "^0.1.2",
-    "js-base64": "^2.1.9",
     "lru-cache": "^4.0.2",
     "multiaddr": "^2.1.1",
     "multipart-stream": "^2.0.1",

src/api/pubsub.js

@@ -1,70 +1,54 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const EventEmitter = require('events')
+const eos = require('end-of-stream')
 const PubsubMessageStream = require('../pubsub-message-stream')
 const stringlistToArray = require('../stringlist-to-array')
 
 /* Internal subscriptions state and functions */
-let subscriptions = {}
-
-const addSubscription = (topic, request) => {
-  subscriptions[topic] = { request: request }
-}
-
-const removeSubscription = promisify((topic, callback) => {
-  if (!subscriptions[topic]) {
-    return callback(new Error(`Not subscribed to ${topic}`))
-  }
-
-  subscriptions[topic].request.abort()
-  delete subscriptions[topic]
-
-  if (callback) {
-    callback(null)
-  }
-})
+const ps = new EventEmitter()
+const subscriptions = {}
 
 /* Public API */
 module.exports = (send) => {
   return {
-    subscribe: promisify((topic, options, callback) => {
+    subscribe: (topic, options, handler, callback) => {
       const defaultOptions = {
         discover: false
       }
 
       if (typeof options === 'function') {
-        callback = options
+        callback = handler
+        handler = options
         options = defaultOptions
       }
 
       if (!options) {
         options = defaultOptions
       }
 
-      // If we're already subscribed, return an error
-      if (subscriptions[topic]) {
-        return callback(new Error(`Already subscribed to '${topic}'`))
+      // promisify doesn't work as we always pass a
+      // function as last argument (`handler`)
+      if (!callback) {
+        return new Promise((resolve, reject) => {
+          subscribe(topic, options, handler, (err) => {
+            if (err) {
+              return reject(err)
+            }
+            resolve()
+          })
+        })
       }
 
-      // Request params
-      const request = {
-        path: 'pubsub/sub',
-        args: [topic],
-        qs: { discover: options.discover }
+      subscribe(topic, options, handler, callback)
+    },
+    unsubscribe (topic, handler) {
+      ps.removeListener(topic, handler)
+      if (ps.listenerCount(topic) === 0) {
+        subscriptions[topic].abort()
       }
-
-      // Start the request and transform the response stream to Pubsub messages stream
-      const req = send.andTransform(request, PubsubMessageStream.from, (err, stream) => {
-        if (err) {
-          return callback(err)
-        }
-        // Add a cancel method to the stream so that the subscription can be cleanly cancelled
-        stream.cancel = promisify((cb) => removeSubscription(topic, cb))
-        // Add the request to the active subscriptions and return the stream
-        addSubscription(topic, req)
-        callback(null, stream)
-      })
-    }),
+    },
     publish: promisify((topic, data, callback) => {
       const buf = Buffer.isBuffer(data) ? data : new Buffer(data)
 
@@ -93,6 +77,53 @@ module.exports = (send) => {
       }
 
       send.andTransform(request, stringlistToArray, callback)
+    }),
+    setMaxListeners (n) {
+      return ps.setMaxListeners(n)
+    }
+  }
+
+  function subscribe (topic, options, handler, callback) {
+    ps.on(topic, handler)
+
+    if (subscriptions[topic]) {
+      return callback()
+    }
+
+    // Request params
+    const request = {
+      path: 'pubsub/sub',
+      args: [topic],
+      qs: {
+        discover: options.discover
+      }
+    }
+
+    // Start the request and transform the response
+    // stream to Pubsub messages stream
+    subscriptions[topic] = send.andTransform(request, PubsubMessageStream.from, (err, stream) => {
+      if (err) {
+        subscriptions[topic] = null
+        return callback(err)
+      }
+
+      stream.on('data', (msg) => {
+        ps.emit(topic, msg)
+      })
+
+      stream.on('error', (err) => {
+        ps.emit('error', err)
+      })
+
+      eos(stream, (err) => {
+        if (err) {
+          ps.emit('error', err)
+        }
+
+        subscriptions[topic] = null
+      })
+
+      callback()
     })
   }
 }

src/pubsub-message-stream.js

@@ -16,13 +16,16 @@ class PubsubMessageStream extends TransformStream {
   }
 
   _transform (obj, enc, callback) {
+    let msg
     try {
-      const message = PubsubMessage.deserialize(obj, 'base64')
-      this.push(message)
+      msg = PubsubMessage.deserialize(obj, 'base64')
     } catch (e) {
       // Not a valid pubsub message
       // go-ipfs returns '{}' as the very first object atm, we skip that
+      return callback()
     }
+
+    this.push(msg)
     callback()
   }
 }

src/pubsub-message-utils.js

@@ -1,47 +1,38 @@
 'use strict'
 
-const Base58 = require('bs58')
-const Base64 = require('js-base64').Base64
-const PubsubMessage = require('./pubsub-message')
-
-class PubsubMessageUtils {
-  static create (senderId, data, seqNo, topics) {
-    return new PubsubMessage(senderId, data, seqNo, topics)
-  }
-
-  static deserialize (data, enc = 'json') {
+module.exports = {
+  deserialize (data, enc = 'json') {
     enc = enc ? enc.toLowerCase() : null
 
     if (enc === 'json') {
-      return PubsubMessageUtils._deserializeFromJson(data)
+      return deserializeFromJson(data)
     } else if (enc === 'base64') {
-      return PubsubMessageUtils._deserializeFromBase64(data)
+      return deserializeFromBase64(data)
     }
 
     throw new Error(`Unsupported encoding: '${enc}'`)
   }
+}
 
-  static _deserializeFromJson (data) {
-    const json = JSON.parse(data)
-    return PubsubMessageUtils._deserializeFromBase64(json)
-  }
-
-  static _deserializeFromBase64 (obj) {
-    if (!PubsubMessageUtils._isPubsubMessage(obj)) {
-      throw new Error(`Not a pubsub message`)
-    }
-
-    const senderId = Base58.encode(obj.from)
-    const payload = Base64.decode(obj.data)
-    const seqno = Base64.decode(obj.seqno)
-    const topics = obj.topicIDs || obj.topicCIDs
+function deserializeFromJson (data) {
+  const json = JSON.parse(data)
+  return deserializeFromBase64(json)
+}
 
-    return PubsubMessageUtils.create(senderId, payload, seqno, topics)
+function deserializeFromBase64 (obj) {
+  if (!isPubsubMessage(obj)) {
+    throw new Error(`Not a pubsub message`)
   }
 
-  static _isPubsubMessage (obj) {
-    return obj && obj.from && obj.seqno && obj.data && (obj.topicIDs || obj.topicCIDs)
+  return {
+    // TODO: broken see https://github.com/ipfs/go-ipfs/issues/3522
+    from: obj.from,
+    seqno: new Buffer(obj.seqno, 'base64'),
+    data: new Buffer(obj.data, 'base64'),
+    topicCIDs: obj.topicIDs || obj.topicCIDs
   }
 }
 
-module.exports = PubsubMessageUtils
+function isPubsubMessage (obj) {
+  return obj && obj.from && obj.seqno && obj.data && (obj.topicIDs || obj.topicCIDs)
+}

src/pubsub-message.js

@@ -78,7 +78,7 @@ function spawnEphemeralNode (callback) {
           Discovery: {},
           API: {
             'HTTPHeaders.Access-Control-Allow-Origin': ['*'],
-            'HTTPHeaders.Access-Control-Allow-Credentials': 'true',
+            'HTTPHeaders.Access-Control-Allow-Credentials': ['true'],
             'HTTPHeaders.Access-Control-Allow-Methods': ['PUT', 'POST', 'GET']
           }
         }

test/factory/daemon-spawner.js

@@ -2,86 +2,24 @@
 'use strict'
 
 const expect = require('chai').expect
-// const PubsubMessage = require('../../src/pubsub-message')
 const PubsubMessageUtils = require('../../src/pubsub-message-utils')
 
-// NOTE!
-// (Most of) these tests are skipped for now until we figure out the
-// final data types for the messages coming over the wire
-
 const topicName = 'js-ipfs-api-tests'
 
 describe('.pubsub-message', () => {
-  it.skip('create message', () => {
-    // TODO
-  })
-
-  it.skip('deserialize message from JSON object', () => {
+  it('deserialize message from JSON object', () => {
     const obj = {
       from: 'BI:ۛv�m�uyѱ����tU�+��#���V',
       data: 'aGk=',
       seqno: 'FIlj2BpyEgI=',
       topicIDs: [ topicName ]
     }
-    try {
-      const message = PubsubMessageUtils.deserialize(obj)
-      expect(message.from).to.equal('AAA')
-      expect(message.data).to.equal('hi')
-      expect(message.seqno).to.equal('\u0014�c�\u001ar\u0012\u0002')
-      expect(message.topicIDs.length).to.equal(1)
-      expect(message.topicIDs[0]).to.equal(topicName)
-    } catch (e) {
-      expect(e).to.not.exist
-    }
-  })
-
-  describe('immutable properties', () => {
-    const sender = 'A'
-    const data = 'hello'
-    const seqno = '123'
-    const topicIDs = ['hello world']
-
-    const message = PubsubMessageUtils.create(sender, data, seqno, topicIDs)
-
-    it('from', () => {
-      try {
-        message.from = 'not allowed'
-      } catch (e) {
-        expect(e).to.be.an('error')
-        expect(e.toString()).to.equal(`TypeError: Cannot set property from of #<PubsubMessage> which has only a getter`)
-      }
-      expect(message.from).to.equal(sender)
-    })
-
-    it('data', () => {
-      try {
-        message.data = 'not allowed'
-      } catch (e) {
-        expect(e).to.be.an('error')
-        expect(e.toString()).to.equal(`TypeError: Cannot set property data of #<PubsubMessage> which has only a getter`)
-      }
-      expect(message.data).to.equal(data)
-    })
-
-    it('seqno', () => {
-      try {
-        message.seqno = 'not allowed'
-      } catch (e) {
-        expect(e).to.be.an('error')
-        expect(e.toString()).to.equal(`TypeError: Cannot set property seqno of #<PubsubMessage> which has only a getter`)
-      }
-      expect(message.seqno).to.equal(seqno)
-    })
 
-    it('topicIDs', () => {
-      try {
-        message.topicIDs = ['not allowed']
-      } catch (e) {
-        expect(e).to.be.an('error')
-        expect(e.toString()).to.equal(`TypeError: Cannot set property topicIDs of #<PubsubMessage> which has only a getter`)
-      }
-      expect(message.topicIDs[0]).to.equal(topicIDs[0])
-      expect(message.topicIDs.length).to.equal(topicIDs.length)
-    })
+    const message = PubsubMessageUtils.deserialize(obj)
+    expect(message.from).to.equal('AAA')
+    expect(message.data).to.equal('hi')
+    expect(message.seqno).to.equal('\u0014�c�\u001ar\u0012\u0002')
+    expect(message.topicIDs.length).to.equal(1)
+    expect(message.topicIDs[0]).to.equal(topicName)
   })
 })

test/ipfs-api/pubsub-message.spec.js

@@ -29,7 +29,7 @@ function startDisposableDaemons (callback) {
         Bootstrap: [],
         Discovery: {},
         'API.HTTPHeaders.Access-Control-Allow-Origin': ['*'],
-        'API.HTTPHeaders.Access-Control-Allow-Credentials': 'true',
+        'API.HTTPHeaders.Access-Control-Allow-Credentials': ['true'],
         'API.HTTPHeaders.Access-Control-Allow-Methods': ['PUT', 'POST', 'GET']
       }