fix: pubsub message fields (#1077)

* fix(pubsub): topicCIDs should be topicIDs

* fix(pubsub): msg.from field is base64 encoding of peer id

* fix: oops

* chore: update deps

* write interop test setup for richard

* fix: allow go daemon flags to be specified

* test: pubsub interop with go (WIP)

* chore: fix linting

* fix more linting

* fix interop tests

Author: richardschneider

package.json

@@ -75,7 +75,7 @@
     "form-data": "^2.3.1",
     "gulp": "^3.9.1",
     "hat": "0.0.3",
-    "interface-ipfs-core": "~0.34.3",
+    "interface-ipfs-core": "~0.35.0",
     "ipfsd-ctl": "~0.24.1",
     "left-pad": "^1.1.3",
     "lodash": "^4.17.4",
@@ -94,6 +94,7 @@
     "async": "^2.6.0",
     "bl": "^1.2.1",
     "boom": "^7.1.1",
+    "bs58": "^4.0.1",
     "byteman": "^1.3.5",
     "cids": "^0.5.2",
     "debug": "^3.1.0",
@@ -105,7 +106,7 @@
     "hapi": "^16.6.2",
     "hapi-set-header": "^1.0.2",
     "hoek": "^5.0.2",
-    "ipfs-api": "^15.1.0",
+    "ipfs-api": "^16.0.0",
     "ipfs-bitswap": "~0.17.4",
     "ipfs-block": "~0.6.1",
     "ipfs-block-service": "~0.13.0",
@@ -119,7 +120,7 @@
     "joi": "^13.0.2",
     "libp2p": "~0.13.1",
     "libp2p-circuit": "~0.1.4",
-    "libp2p-floodsub": "~0.11.1",
+    "libp2p-floodsub": "~0.12.0",
     "libp2p-kad-dht": "~0.6.0",
     "libp2p-mdns": "~0.9.1",
     "libp2p-multiplex": "~0.5.0",

src/http/api/resources/pubsub.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const PassThrough = require('stream').PassThrough
+const bs58 = require('bs58')
 
 exports = module.exports
 
@@ -20,10 +21,10 @@ exports.subscribe = {
 
     const handler = (msg) => {
       res.write(JSON.stringify({
-        from: msg.from,
+        from: bs58.decode(msg.from).toString('base64'),
         data: msg.data.toString('base64'),
         seqno: msg.seqno.toString('base64'),
-        topicCIDs: msg.topicCIDs
+        topicIDs: msg.topicIDs
       }) + '\n', 'utf8')
     }
 

test/interop/node.js

@@ -5,3 +5,4 @@ require('./repo')
 require('./exchange-files')
 require('./circuit-relay')
 require('./kad-dht')
+require('./pubsub')

test/interop/pubsub.js

@@ -0,0 +1,80 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const dirtyChai = require('dirty-chai')
+const expect = chai.expect
+chai.use(dirtyChai)
+const series = require('async/series')
+const parallel = require('async/parallel')
+
+const GODaemon = require('../utils/interop-daemon-spawner/go')
+const JSDaemon = require('../utils/interop-daemon-spawner/js')
+
+describe('pubsub', () => {
+  let jsD
+  let goD
+  let jsId
+  let goId
+
+  before(function (done) {
+    this.timeout(50 * 1000)
+
+    goD = new GODaemon({
+      disposable: true,
+      init: true,
+      flags: ['--enable-pubsub-experiment']
+    })
+    jsD = new JSDaemon()
+
+    parallel([
+      (cb) => goD.start(cb),
+      (cb) => jsD.start(cb)
+    ], (done))
+  })
+
+  after((done) => {
+    series([
+      (cb) => goD.stop(cb),
+      (cb) => jsD.stop(cb)
+    ], done)
+  })
+
+  it('make connections', (done) => {
+    parallel([
+      (cb) => jsD.api.id(cb),
+      (cb) => goD.api.id(cb)
+    ], (err, ids) => {
+      expect(err).to.not.exist()
+
+      jsId = ids[0].ID
+      goId = ids[0].ID
+
+      console.log('jsId:', jsId)
+      console.log('goId:', goId)
+
+      parallel([
+        (cb) => jsD.api.swarm.connect(ids[1].addresses[0], cb),
+        (cb) => goD.api.swarm.connect(ids[0].addresses[0], cb)
+      ], done)
+    })
+  })
+
+  it.skip('publish from JS, subscribe on Go', (done) => {
+    // TODO write this test
+  })
+
+  it.skip('publish from Go, subscribe on JS', (done) => {
+    const topic = 'pubsub-go-js'
+    const data = Buffer.from('hello world')
+
+    function checkMessage () {
+      console.log('check message', arguments)
+    }
+
+    series([
+      cb => jsD.api.pubsub.subscribe(topic, checkMessage, cb),
+      cb => goD.api.pubsub.publish(topic, data, cb)
+    ], done)
+  })
+})

test/utils/interop-daemon-spawner/go.js

@@ -16,6 +16,7 @@ class GoDaemon {
     this.node = null
     this.api = null
     this.config = opts.config || {}
+    this.flags = opts.flags || {}
   }
 
   start (callback) {
@@ -40,6 +41,7 @@ class GoDaemon {
         this.node.setConfig('Bootstrap', '[]', cb)
       },
       (res, cb) => this.node.startDaemon(cb),
+      // (res, cb) => this.node.startDaemon(this.flags, cb),
       (api, cb) => {
         this.api = api