Pubsub

Author: haadcode

package.json

@@ -51,6 +51,7 @@
     "hapi": "^15.0.2",
     "interface-ipfs-core": "^0.14.2",
     "ipfsd-ctl": "^0.14.0",
+    "js-base64": "^2.1.9",
     "pre-commit": "^1.1.3",
     "socket.io": "^1.4.8",
     "socket.io-client": "^1.4.8",
@@ -100,4 +101,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,97 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+const bs58 = require('bs58')
+var Base64 = require('js-base64').Base64
+// const Wreck = require('wreck')
+var http = require('http')
+
+module.exports = (send) => {
+  const api = {
+    sub: promisify((topic, options, callback) => {
+      if (typeof options === 'function') {
+        callback = options
+        options = {}
+      }
+      if (!options) {
+        options = {}
+      }
+
+      console.log('Sub', topic)
+      http.get({
+        host: 'localhost',
+        port: 5001,
+        path: '/api/v0/pubsub/sub/' + topic
+      }, function (response) {
+        // Continuously update stream with data
+        var body = ''
+        response.on('data', function (d) {
+          // console.log('>', d.toString())
+          var parsed = JSON.parse(d)
+          parsed.from = bs58.encode(parsed.from)
+          parsed.data = Base64.decode(parsed.data)
+          parsed.seqno = Base64.decode(parsed.seqno)
+          console.log(parsed)
+          body += d
+        })
+        response.on('end', function () {
+          // Data reception is done, do whatever with it!
+          var parsed = JSON.parse(body)
+          console.log(parsed)
+          callback(null, parsed)
+        })
+      })
+
+      // Wreck.get('http://localhost:5001/api/v0/pubsub/sub/' + topic, (err, res, payload) => {
+      //   if (err) {
+      //     return callback(err)
+      //   }
+
+      //   console.log(payload.toString())
+      //   // const result = JSON.parse(res)
+      //   callback(null, payload)
+      // })
+
+      // send({
+      //   path: 'pubsub/sub/' + topic
+      // }, (err, result) => {
+      //   console.log('RESULT', err, result)
+      //   if (err) {
+      //     return callback(err)
+      //   }
+
+      //   callback(null, result) // result is a Stream
+      // })
+    }),
+    pub: promisify((topic, data, options, callback) => {
+      if (typeof options === 'function') {
+        callback = options
+        options = {}
+      }
+      if (!options) {
+        options = {}
+      }
+
+      let buf
+      if (Buffer.isBuffer(data)) {
+        buf = data
+      } else {
+        buf = new Buffer(data)
+      }
+
+      send({
+        path: 'pubsub/pub',
+        qs: { topic: topic },
+        files: buf
+      }, (err, result) => {
+        if (err) {
+          return callback(err)
+        }
+
+        callback(null, true)
+      })
+    })
+  }
+
+  return api
+}

src/load-commands.js

@@ -22,6 +22,7 @@ function requireCommands () {
     object: require('./api/object'),
     pin: require('./api/pin'),
     ping: require('./api/ping'),
+    pubsub: require('./api/pubsub'),
     refs: require('./api/refs'),
     repo: require('./api/repo'),
     swarm: require('./api/swarm'),

test.js

@@ -0,0 +1,29 @@
+'use strict'
+
+var http = require('http')
+var Base64 = require('js-base64').Base64
+const bs58 = require('bs58')
+
+http.get({
+  host: 'localhost',
+  port: 5001,
+  path: '/api/v0/pubsub/sub/testi1'
+}, function (response) {
+  // Continuously update stream with data
+  var body = ''
+  response.on('data', function (d) {
+    // console.log(">", d.toString())
+    var parsed = JSON.parse(d)
+    parsed.from = bs58.encode(parsed.from)
+    parsed.data = Base64.decode(parsed.data)
+    parsed.seqno = Base64.decode(parsed.seqno)
+    console.log(parsed)
+    body += d
+  })
+  response.on('end', function () {
+    // Data reception is done, do whatever with it!
+    var parsed = JSON.parse(body)
+    console.log(parsed)
+    // callback(null, parsed)
+  })
+})

test/ipfs-api/pubsub.spec.js

@@ -0,0 +1,65 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ['error', 8] */
+'use strict'
+
+const expect = require('chai').expect
+const isNode = require('detect-node')
+const FactoryClient = require('../factory/factory-client')
+
+const useLocalDaemon = true
+
+describe('.pubsub', () => {
+  if (!isNode) {
+    return
+  }
+
+  let ipfs
+  let fc
+
+  before(function (done) {
+    this.timeout(20 * 1000) // slow CI
+    fc = new FactoryClient()
+    fc.spawnNode(null, null, useLocalDaemon, (err, node) => {
+      expect(err).to.not.exist
+      ipfs = node
+      done()
+    })
+  })
+
+  after((done) => {
+    fc.dismantle(done)
+  })
+
+  it.only('sub', (done) => {
+    console.log('1')
+    ipfs.pubsub.sub('testi1', (err, result) => {
+      console.log('RESULT1', err, result)
+      expect(err).to.not.exist
+      expect(result.length).to.equal(1)
+      done()
+    })
+    // setTimeout(() => {
+    //   ipfs.pubsub.pub('testi1', 'hi')
+    // }, 1000)
+  })
+
+  describe('.pub', () => {
+    it('publishes a message - from string', (done) => {
+      const data = 'hello friend'
+      ipfs.pubsub.pub('testi1', data, (err, result) => {
+        expect(err).to.not.exist
+        expect(result).to.equal(true)
+        done()
+      })
+    })
+
+    it('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()
+      })
+    })
+  })
+})