test: tests js/go pubsub interop

Author: richardschneider

test/interop/pubsub.js

@@ -11,7 +11,24 @@ const parallel = require('async/parallel')
 const GODaemon = require('../utils/interop-daemon-spawner/go')
 const JSDaemon = require('../utils/interop-daemon-spawner/js')
 
-describe('pubsub', () => {
+/*
+ * Wait for a condition to become true.  When its true, callback is called.
+ */
+function waitFor (predicate, callback) {
+  const ttl = Date.now() + (2 * 1000)
+  const self = setInterval(() => {
+    if (predicate()) {
+      clearInterval(self)
+      callback()
+    }
+    if (Date.now() > ttl) {
+      clearInterval(self)
+      callback(new Error("waitFor time expired"))
+    }
+  }, 500)
+}
+
+describe('pubsub', function () {
   let jsD
   let goD
   let jsId
@@ -34,47 +51,115 @@ describe('pubsub', () => {
   })
 
   after((done) => {
-    series([
+    parallel([
       (cb) => goD.stop(cb),
       (cb) => jsD.stop(cb)
     ], done)
   })
 
   it('make connections', (done) => {
-    parallel([
+    series([
       (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
+      jsId = ids[0].id
+      goId = ids[1].id
 
-      console.log('jsId:', jsId)
-      console.log('goId:', goId)
+      const jsLocalAddr = ids[0].addresses.find(a => a.includes('127.0.0.1'))
+      const goLocalAddr = ids[1].addresses.find(a => a.includes('127.0.0.1'))
 
       parallel([
-        (cb) => jsD.api.swarm.connect(ids[1].addresses[0], cb),
-        (cb) => goD.api.swarm.connect(ids[0].addresses[0], cb)
+        (cb) => jsD.api.swarm.connect(goLocalAddr, cb),
+        (cb) => goD.api.swarm.connect(jsLocalAddr, cb)
       ], done)
     })
   })
 
-  it.skip('publish from JS, subscribe on Go', (done) => {
-    // TODO write this test
+  it('publish from Go, subscribe on Go', (done) => {
+    const topic = 'pubsub-go-go'
+    const data = Buffer.from('hello world')
+    let n = 0
+
+    function checkMessage (msg) {
+      ++n
+      expect(msg.data.toString()).to.equal(data.toString())
+      expect(msg).to.have.property('seqno')
+      expect(Buffer.isBuffer(msg.seqno)).to.be.eql(true)
+      // TODO: expect(msg).to.have.property('topicIDs').eql([topic])
+      expect(msg).to.have.property('from', goId)
+    }
+
+    series([
+      (cb) => goD.api.pubsub.subscribe(topic, checkMessage, cb),
+      (cb) => goD.api.pubsub.publish(topic, data, cb),
+      (cb) => waitFor(() => n === 1, cb)
+    ], done)
+  })
+
+  it('publish from JS, subscribe on JS', (done) => {
+    const topic = 'pubsub-js-js'
+    const data = Buffer.from('hello world')
+    let n = 0
+
+    function checkMessage (msg) {
+      ++n
+      expect(msg.data.toString()).to.equal(data.toString())
+      expect(msg).to.have.property('seqno')
+      expect(Buffer.isBuffer(msg.seqno)).to.be.eql(true)
+      expect(msg).to.have.property('topicIDs').eql([topic])
+      expect(msg).to.have.property('from', jsId)
+    }
+
+    series([
+      (cb) => jsD.api.pubsub.subscribe(topic, checkMessage, cb),
+      (cb) => jsD.api.pubsub.publish(topic, data, cb),
+      (cb) => waitFor(() => n === 1, cb)
+    ], done)
+  })
+
+  it('publish from JS, subscribe on Go', (done) => {
+    const topic = 'pubsub-js-go'
+    const data = Buffer.from('hello world')
+    let n = 0
+
+    function checkMessage (msg) {
+      console.log('check message', msg)
+      ++n
+      expect(msg.data.toString()).to.equal(data.toString())
+      expect(msg).to.have.property('seqno')
+      expect(Buffer.isBuffer(msg.seqno)).to.be.eql(true)
+      expect(msg).to.have.property('topicIDs').eql([topic])
+      expect(msg).to.have.property('from', jsId)
+    }
+
+    series([
+      (cb) => goD.api.pubsub.subscribe(topic, checkMessage, cb),
+      (cb) => jsD.api.pubsub.publish(topic, data, cb),
+      (cb) => waitFor(() => n === 1, cb)
+    ], done)
   })
 
-  it.skip('publish from Go, subscribe on JS', (done) => {
+  it('publish from Go, subscribe on JS', (done) => {
     const topic = 'pubsub-go-js'
     const data = Buffer.from('hello world')
+    let n = 0
 
-    function checkMessage () {
-      console.log('check message', arguments)
+    function checkMessage (msg) {
+      console.log('check message', msg)
+      ++n
+      expect(msg.data.toString()).to.equal(data.toString())
+      expect(msg).to.have.property('seqno')
+      expect(Buffer.isBuffer(msg.seqno)).to.be.eql(true)
+      expect(msg).to.have.property('topicIDs').eql([topic])
+      expect(msg).to.have.property('from', goId)
     }
 
     series([
-      cb => jsD.api.pubsub.subscribe(topic, checkMessage, cb),
-      cb => goD.api.pubsub.publish(topic, data, cb)
+      (cb) => jsD.api.pubsub.subscribe(topic, checkMessage, cb),
+      (cb) => goD.api.pubsub.publish(topic, data, cb),
+      (cb) => waitFor(() => n === 1, cb),
     ], done)
   })
 })

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

@@ -40,8 +40,7 @@ class GoDaemon {
         this.node = node
         this.node.setConfig('Bootstrap', '[]', cb)
       },
-      (res, cb) => this.node.startDaemon(cb),
-      // (res, cb) => this.node.startDaemon(this.flags, cb),
+      (res, cb) => this.node.startDaemon(this.flags, cb),
       (api, cb) => {
         this.api = api
 

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

@@ -37,14 +37,17 @@ class JsDaemon extends EventEmitter {
     this.path = opts.path || tmpDir()
     this._started = false
 
+    const extras = {
+      enablePubsubExperiment: true
+    }
     if (this.init) {
       const p = portConfig(this.port)
       this.node = new HttpApi(this.path, {
         Bootstrap: [],
         Addresses: p
-      })
+      }, extras)
     } else {
-      this.node = new HttpApi(this.path)
+      this.node = new HttpApi(this.path, null, extras)
     }
 
     this.node.start(this.init, (err) => {

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

@@ -7,7 +7,7 @@ const path = require('path')
 exports.tmpDir = (prefix) => {
   return path.join(
     os.tmpdir(),
-    prefix || 'tmp',
+    prefix || 'js-ipfs-interop',
     crypto.randomBytes(32).toString('hex')
   )
 }