refactor(pubsub): Refactored and completed pubsub API

Author: haadcode

package.json

@@ -34,6 +34,7 @@
     "ipld-dag-pb": "^0.9.2",
     "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",
@@ -63,7 +64,7 @@
     "gulp": "^3.9.1",
     "hapi": "^16.0.1",
     "interface-ipfs-core": "^0.22.0",
-    "ipfsd-ctl": "^0.17.0",
+    "@haad/ipfsd-ctl": "^0.18.0-beta.5",
     "pre-commit": "^1.1.3",
     "socket.io": "^1.7.1",
     "socket.io-client": "^1.7.1",
@@ -115,4 +116,4 @@
     "url": "https://github.com/ipfs/js-ipfs-api/issues"
   },
   "homepage": "https://github.com/ipfs/js-ipfs-api"
-}
+}

src/api/pubsub.js

@@ -0,0 +1,98 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+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)
+  }
+})
+
+/* Public API */
+module.exports = (send) => {
+  return {
+    subscribe: promisify((topic, options, callback) => {
+      const defaultOptions = {
+        discover: false
+      }
+
+      if (typeof options === 'function') {
+        callback = 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}'`))
+      }
+
+      // 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
+      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)
+
+      const request = {
+        path: 'pubsub/pub',
+        args: [topic, buf]
+      }
+
+      send(request, callback)
+    }),
+    ls: promisify((callback) => {
+      const request = {
+        path: 'pubsub/ls'
+      }
+
+      send.andTransform(request, stringlistToArray, callback)
+    }),
+    peers: promisify((topic, callback) => {
+      if (!subscriptions[topic]) {
+        return callback(new Error(`Not subscribed to '${topic}'`))
+      }
+
+      const request = {
+        path: 'pubsub/peers',
+        args: [topic]
+      }
+
+      send.andTransform(request, stringlistToArray, callback)
+    })
+  }
+}

src/load-commands.js

@@ -25,6 +25,7 @@ function requireCommands () {
     refs: require('./api/refs'),
     repo: require('./api/repo'),
     swarm: require('./api/swarm'),
+    pubsub: require('./api/pubsub'),
     update: require('./api/update'),
     version: require('./api/version')
   }

src/pubsub-message-stream.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const TransformStream = require('readable-stream').Transform
+const PubsubMessage = require('./pubsub-message-utils')
+
+class PubsubMessageStream extends TransformStream {
+  constructor (options) {
+    const opts = Object.assign(options || {}, { objectMode: true })
+    super(opts)
+  }
+
+  static from (inputStream, callback) {
+    let outputStream = inputStream.pipe(new PubsubMessageStream())
+    inputStream.on('end', () => outputStream.emit('end'))
+    callback(null, outputStream)
+  }
+
+  _transform (obj, enc, callback) {
+    try {
+      const message = PubsubMessage.deserialize(obj, 'base64')
+      this.push(message)
+    } catch (e) {
+      // Not a valid pubsub message
+      // go-ipfs returns '{}' as the very first object atm, we skip that
+    }
+    callback()
+  }
+}
+
+module.exports = PubsubMessageStream

src/pubsub-message-utils.js

@@ -0,0 +1,47 @@
+'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') {
+    enc = enc ? enc.toLowerCase() : null
+
+    if (enc === 'json') {
+      return PubsubMessageUtils._deserializeFromJson(data)
+    } else if (enc === 'base64') {
+      return PubsubMessageUtils._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
+
+    return PubsubMessageUtils.create(senderId, payload, seqno, topics)
+  }
+
+  static _isPubsubMessage (obj) {
+    return obj && obj.from && obj.seqno && obj.data && obj.topicIDs
+  }
+}
+
+module.exports = PubsubMessageUtils

src/pubsub-message.js

@@ -0,0 +1,28 @@
+'use strict'
+
+class PubsubMessage {
+  constructor (senderId, data, seqNo, topics) {
+    this._senderId = senderId
+    this._data = data
+    this._seqNo = seqNo
+    this._topics = topics
+  }
+
+  get from () {
+    return this._senderId
+  }
+
+  get data () {
+    return this._data
+  }
+
+  get seqno () {
+    return this._seqNo
+  }
+
+  get topicIDs () {
+    return this._topics
+  }
+}
+
+module.exports = PubsubMessage

test/factory/daemon-spawner.js

@@ -1,7 +1,7 @@
 'use strict'
 
 // const defaultConfig = require('./default-config.json')
-const ipfsd = require('ipfsd-ctl')
+const ipfsd = require('@haad/ipfsd-ctl')
 const series = require('async/series')
 const eachSeries = require('async/eachSeries')
 const once = require('once')
@@ -73,17 +73,19 @@ function spawnEphemeralNode (callback) {
       (cb) => {
         const configValues = {
           Bootstrap: [],
-          Discovery: {},
-          'HTTPHeaders.Access-Control-Allow-Origin': ['*'],
-          'HTTPHeaders.Access-Control-Allow-Credentials': 'true',
-          'HTTPHeaders.Access-Control-Allow-Methods': ['PUT', 'POST', 'GET']
+          // Discovery: {},
+          API: {
+            'HTTPHeaders.Access-Control-Allow-Origin': ['*'],
+            'HTTPHeaders.Access-Control-Allow-Credentials': 'true',
+            'HTTPHeaders.Access-Control-Allow-Methods': ['PUT', 'POST', 'GET']
+          }
         }
 
         eachSeries(Object.keys(configValues), (configKey, cb) => {
-          node.setConfig(`API.${configKey}`, JSON.stringify(configValues[configKey]), cb)
+          node.setConfig(`${configKey}`, JSON.stringify(configValues[configKey]), cb)
         }, cb)
       },
-      (cb) => node.startDaemon(cb)
+      (cb) => node.startDaemon(['--enable-pubsub-experiment'], cb)
     ], (err) => {
       if (err) {
         return callback(err)

test/setup/spawn-daemons.js

@@ -5,7 +5,7 @@
 const gulp = require('gulp')
 const fs = require('fs')
 const path = require('path')
-const ipfsd = require('ipfsd-ctl')
+const ipfsd = require('@haad/ipfsd-ctl')
 const eachSeries = require('async/eachSeries')
 const parallel = require('async/parallel')
 
@@ -27,7 +27,7 @@ function startDisposableDaemons (callback) {
 
       const configValues = {
         Bootstrap: [],
-        Discovery: {},
+        // Discovery: {},
         API: {
           'HTTPHeaders.Access-Control-Allow-Origin': ['*'],
           'HTTPHeaders.Access-Control-Allow-Credentials': 'true',
@@ -36,13 +36,13 @@ function startDisposableDaemons (callback) {
       }
 
       eachSeries(Object.keys(configValues), (configKey, cb) => {
-        nodes[key].setConfig(`API.${configKey}`, JSON.stringify(configValues[configKey]), cb)
+        nodes[key].setConfig(`${configKey}`, JSON.stringify(configValues[configKey]), cb)
       }, (err) => {
         if (err) {
           return cb(err)
         }
 
-        nodes[key].startDaemon(cb)
+        nodes[key].startDaemon(['--enable-pubsub-experiment'], cb)
       })
     })
   }