Feature: PubSub 🌟

package.json

@@ -50,17 +50,18 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs#readme",
   "devDependencies": {
-    "aegir": "^9.2.1",
+    "aegir": "^9.2.2",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
+    "delay": "^1.3.1",
     "detect-node": "^2.0.3",
     "eslint-plugin-react": "^6.8.0",
     "execa": "^0.5.0",
     "expose-loader": "^0.7.1",
     "form-data": "^2.1.2",
     "fs-pull-blob-store": "^0.4.1",
     "gulp": "^3.9.1",
-    "interface-ipfs-core": "^0.22.1",
+    "interface-ipfs-core": "ipfs/interface-ipfs-core#501c304",
     "left-pad": "^1.1.3",
     "lodash": "^4.17.2",
     "ncp": "^2.0.0",
@@ -80,20 +81,22 @@
     "glob": "^7.1.1",
     "hapi": "^16.0.1",
     "hapi-set-header": "^1.0.2",
+    "hoek": "^4.1.0",
     "idb-pull-blob-store": "^0.5.1",
-    "ipfs-api": "^12.1.0",
-    "ipfs-bitswap": "^0.8.2",
+    "ipfs-api": "ipfs/js-ipfs-api#32908ae",
+    "ipfs-bitswap": "^0.8.3",
     "ipfs-block": "^0.5.3",
-    "ipfs-block-service": "^0.7.1",
+    "ipfs-block-service": "^0.7.2",
     "ipfs-multipart": "^0.1.0",
-    "ipfs-repo": "^0.11.1",
-    "ipfs-unixfs": "^0.1.8",
-    "ipfs-unixfs-engine": "^0.14.1",
-    "ipld-resolver": "^0.4.0",
+    "ipfs-repo": "^0.11.2",
+    "ipfs-unixfs": "^0.1.9",
+    "ipfs-unixfs-engine": "^0.14.2",
+    "ipld-resolver": "^0.4.1",
     "isstream": "^0.1.2",
     "joi": "^10.0.5",
-    "libp2p-ipfs-nodejs": "^0.17.0",
+    "libp2p-floodsub": "0.5.0",
     "libp2p-ipfs-browser": "^0.17.2",
+    "libp2p-ipfs-nodejs": "^0.17.0",
     "lodash.flatmap": "^4.5.0",
     "lodash.get": "^4.4.2",
     "lodash.has": "^4.5.2",

src/cli/commands/pubsub.js

@@ -0,0 +1,14 @@
+'use strict'
+
+module.exports = {
+  command: 'pubsub',
+
+  description: 'pubsub commands',
+
+  builder (yargs) {
+    return yargs
+      .commandDir('pubsub')
+  },
+
+  handler (argv) {}
+}

src/cli/commands/pubsub/ls.js

@@ -0,0 +1,32 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:pubsub')
+log.error = debug('cli:pubsub:error')
+
+module.exports = {
+  command: 'ls',
+
+  describe: 'Get your list of subscriptions',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.pubsub.ls((err, subscriptions) => {
+        if (err) {
+          throw err
+        }
+
+        subscriptions.forEach((sub) => {
+          console.log(sub)
+        })
+      })
+    })
+  }
+}

src/cli/commands/pubsub/peers.js

@@ -0,0 +1,32 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:pubsub')
+log.error = debug('cli:pubsub:error')
+
+module.exports = {
+  command: 'peers <topic>',
+
+  describe: 'Get all peers subscribed to a topic',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.pubsub.peers(argv.topic, (err, peers) => {
+        if (err) {
+          throw err
+        }
+
+        peers.forEach((peer) => {
+          console.log(peer)
+        })
+      })
+    })
+  }
+}

src/cli/commands/pubsub/pub.js

@@ -0,0 +1,28 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:pubsub')
+log.error = debug('cli:pubsub:error')
+
+module.exports = {
+  command: 'pub <topic> <data>',
+
+  describe: 'Publish data to a topic',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.pubsub.publish(argv.topic, argv.data, (err) => {
+        if (err) {
+          throw err
+        }
+      })
+    })
+  }
+}

src/cli/commands/pubsub/sub.js

@@ -0,0 +1,32 @@
+'use strict'
+
+const utils = require('../../utils')
+const debug = require('debug')
+const log = debug('cli:pubsub')
+log.error = debug('cli:pubsub:error')
+
+module.exports = {
+  command: 'sub <topic>',
+
+  describe: 'Subscribe to a topic',
+
+  builder: {},
+
+  handler (argv) {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      const handler = (msg) => {
+        console.log(msg.data.toString())
+      }
+
+      ipfs.pubsub.subscribe(argv.topic, handler, (err) => {
+        if (err) {
+          throw err
+        }
+      })
+    })
+  }
+}

src/core/components/go-online.js

@@ -2,6 +2,7 @@
 
 const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
+const FloodSub = require('libp2p-floodsub')
 
 module.exports = function goOnline (self) {
   return (cb) => {
@@ -21,6 +22,8 @@ module.exports = function goOnline (self) {
       )
       self._bitswap.start()
       self._blockService.goOnline(self._bitswap)
+      self._pubsub = new FloodSub(self._libp2pNode)
+
       cb()
     })
   }

src/core/components/pubsub.js

@@ -0,0 +1,98 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+const setImmediate = require('async/setImmediate')
+
+const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
+
+module.exports = function pubsub (self) {
+  return {
+    subscribe: (topic, options, handler, callback) => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      if (typeof options === 'function') {
+        callback = handler
+        handler = options
+        options = {}
+      }
+
+      if (!callback) {
+        return new Promise((resolve, reject) => {
+          subscribe(topic, options, handler, (err) => {
+            if (err) {
+              return reject(err)
+            }
+            resolve()
+          })
+        })
+      }
+
+      subscribe(topic, options, handler, callback)
+    },
+
+    unsubscribe: (topic, handler) => {
+      const ps = self._pubsub
+
+      ps.removeListener(topic, handler)
+
+      if (ps.listenerCount(topic) === 0) {
+        ps.unsubscribe(topic)
+      }
+    },
+
+    publish: promisify((topic, data, callback) => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      if (typeof data === 'string') {
+        data = new Buffer(data)
+      }
+
+      self._pubsub.publish(topic, data)
+
+      setImmediate(callback)
+    }),
+
+    ls: promisify((callback) => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      const subscriptions = Array.from(
+        self._pubsub.subscriptions
+      )
+
+      setImmediate(() => callback(null, subscriptions))
+    }),
+
+    peers: promisify((topic, callback) => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      const peers = Array.from(self._pubsub.peers.values())
+          .filter((peer) => peer.topics.has(topic))
+          .map((peer) => peer.info.id.toB58String())
+
+      setImmediate(() => callback(null, peers))
+    }),
+
+    setMaxListeners (n) {
+      return self._pubsub.setMaxListeners(n)
+    }
+  }
+
+  function subscribe (topic, options, handler, callback) {
+    const ps = self._pubsub
+
+    if (ps.listenerCount(topic) === 0) {
+      ps.subscribe(topic)
+    }
+
+    ps.on(topic, handler)
+    setImmediate(callback)
+  }
+}

src/core/index.js

@@ -23,6 +23,7 @@ const swarm = require('./components/swarm')
 const ping = require('./components/ping')
 const files = require('./components/files')
 const bitswap = require('./components/bitswap')
+const pubsub = require('./components/pubsub')
 
 exports = module.exports = IPFS
 
@@ -44,6 +45,7 @@ function IPFS (repoInstance) {
   this._bitswap = null
   this._blockService = new BlockService(this._repo)
   this._ipldResolver = new IPLDResolver(this._blockService)
+  this._pubsub = null
 
   // IPFS Core exposed components
 
@@ -67,4 +69,5 @@ function IPFS (repoInstance) {
   this.files = files(this)
   this.bitswap = bitswap(this)
   this.ping = ping(this)
+  this.pubsub = pubsub(this)
 }

src/http-api/error-handler.js

@@ -0,0 +1,54 @@
+'use strict'
+
+const Hoek = require('hoek')
+
+module.exports = (server) => {
+  server.ext('onRequest', (request, reply) => {
+    request.handleError = handleError
+    reply.continue()
+  })
+
+  server.ext('onPreResponse', (request, reply) => {
+    const res = request.response
+    const req = request.raw.req
+
+    let statusCode = 200
+    let msg = 'Sorry, something went wrong, please retrace your steps.'
+
+    if (res.isBoom) {
+      statusCode = res.output.payload.statusCode
+
+      if (res.message && res.isDeveloperError) {
+        // we caught it!
+        msg = res.message.replace('Uncaught error: ', '')
+      }
+
+      const debug = {
+        method: req.method,
+        url: request.url.path,
+        headers: request.raw.req.headers,
+        info: request.info,
+        payload: request.payload,
+        response: res.output.payload
+      }
+      // ALWAYS Log the error
+      server.log('error', debug)
+
+      reply({
+        Message: msg,
+        Code: 0
+      }).code(statusCode)
+      return
+    }
+
+    reply.continue()
+  })
+}
+
+function handleError (error, errorMessage) {
+  if (errorMessage) {
+    return Hoek.assert(!error, errorMessage)
+  }
+
+  return Hoek.assert(!error, error)
+}