fix: pubsub promisify (#456)

* fix: allow pubsub sub/unsub via promises

* chore: fix linting errors

Author: achingbrain

package.json

@@ -81,6 +81,7 @@
     "chai": "^4.2.0",
     "chai-checkmark": "^1.0.1",
     "cids": "^0.7.1",
+    "delay": "^4.3.0",
     "dirty-chai": "^2.0.1",
     "electron-webrtc": "^0.3.0",
     "interface-datastore": "^0.6.0",

src/circuit/circuit/dialer.js

@@ -153,7 +153,7 @@ class Dialer {
       const relays = Array.from(this.relayPeers.values())
       const next = (nextRelay) => {
         if (!nextRelay) {
-          const err = `no relay peers were found or all relays failed to dial`
+          const err = 'no relay peers were found or all relays failed to dial'
           log.err(err)
           return cb(err)
         }
@@ -235,7 +235,7 @@ class Dialer {
         }
         const message = proto.CircuitRelay.decode(msg)
         if (message.type !== proto.CircuitRelay.Type.STATUS) {
-          return callback(new Error(`Got invalid message type - ` +
+          return callback(new Error('Got invalid message type - ' +
             `expected ${proto.CircuitRelay.Type.STATUS} got ${message.type}`))
         }
 

src/circuit/circuit/hop.js

@@ -203,7 +203,7 @@ class Hop extends EE {
 
           const message = proto.decode(msg)
           if (message.code !== proto.Status.SUCCESS) {
-            return callback(new Error(`Unable to create circuit!`))
+            return callback(new Error('Unable to create circuit!'))
           }
 
           return callback(null, msg)

src/circuit/circuit/stream-handler.js

@@ -49,7 +49,7 @@ class StreamHandler {
    */
   read (cb) {
     if (!this.isValid()) {
-      return cb(new Error(`handler is not in a valid state`))
+      return cb(new Error('handler is not in a valid state'))
     }
 
     lp.decodeFromReader(
@@ -77,7 +77,7 @@ class StreamHandler {
     cb = cb || (() => {})
 
     if (!this.isValid()) {
-      return cb(new Error(`handler is not in a valid state`))
+      return cb(new Error('handler is not in a valid state'))
     }
 
     pull(

src/circuit/listener.js

@@ -132,10 +132,10 @@ module.exports = (swarm, options, connHandler) => {
       if (!mafmt.Circuit.matches(addr)) {
         if (addr.getPeerId()) {
           // by default we're reachable over any relay
-          listenAddrs.push(multiaddr(`/p2p-circuit`).encapsulate(addr))
+          listenAddrs.push(multiaddr('/p2p-circuit').encapsulate(addr))
         } else {
           const ma = `${addr}/ipfs/${swarm._peerInfo.id.toB58String()}`
-          listenAddrs.push(multiaddr(`/p2p-circuit`).encapsulate(ma))
+          listenAddrs.push(multiaddr('/p2p-circuit').encapsulate(ma))
         }
       } else {
         listenAddrs.push(addr.encapsulate(`/ipfs/${swarm._peerInfo.id.toB58String()}`))

src/pubsub.js

@@ -32,27 +32,35 @@ module.exports = (node, Pubsub, config) => {
      * const handler = (message) => { }
      * libp2p.subscribe(topic, handler, callback)
      */
-    subscribe: promisify((topic, handler, options, callback) => {
+    subscribe: (topic, handler, options, callback) => {
+      // can't use promisify because it thinks the handler is a callback
       if (typeof options === 'function') {
         callback = options
         options = {}
       }
 
       if (!node.isStarted() && !pubsub.started) {
-        return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
-      }
+        const err = errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED)
 
-      function subscribe (cb) {
-        if (pubsub.listenerCount(topic) === 0) {
-          pubsub.subscribe(topic)
+        if (callback) {
+          return nextTick(() => callback(err))
         }
 
-        pubsub.on(topic, handler)
-        nextTick(cb)
+        return Promise.reject(err)
       }
 
-      subscribe(callback)
-    }),
+      if (pubsub.listenerCount(topic) === 0) {
+        pubsub.subscribe(topic)
+      }
+
+      pubsub.on(topic, handler)
+
+      if (callback) {
+        return nextTick(() => callback())
+      }
+
+      return Promise.resolve()
+    },
 
     /**
      * Unsubscribes from a pubsub topic
@@ -76,9 +84,16 @@ module.exports = (node, Pubsub, config) => {
      *
      * libp2p.unsubscribe(topic, handler, callback)
      */
-    unsubscribe: promisify((topic, handler, callback) => {
+    unsubscribe: (topic, handler, callback) => {
+      // can't use promisify because it thinks the handler is a callback
       if (!node.isStarted() && !pubsub.started) {
-        return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
+        const err = errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED)
+
+        if (callback) {
+          return nextTick(() => callback(err))
+        }
+
+        return Promise.reject(err)
       }
 
       if (!handler) {
@@ -91,12 +106,12 @@ module.exports = (node, Pubsub, config) => {
         pubsub.unsubscribe(topic)
       }
 
-      if (typeof callback === 'function') {
+      if (callback) {
         return nextTick(() => callback())
       }
 
       return Promise.resolve()
-    }),
+    },
 
     publish: promisify((topic, data, callback) => {
       if (!node.isStarted() && !pubsub.started) {

src/switch/connection/handler.js

@@ -5,7 +5,7 @@ const IncomingConnection = require('./incoming')
 const observeConn = require('../observe-connection')
 
 function listener (_switch) {
-  const log = debug(`libp2p:switch:listener`)
+  const log = debug('libp2p:switch:listener')
 
   /**
    * Takes a transport key and returns a connection handler function

src/switch/protocol-muxer.js

@@ -41,7 +41,7 @@ module.exports = function protocolMuxer (protocols, observer) {
 
     ms.handle(parentConn, (err) => {
       if (err) {
-        log.error(`multistream handshake failed`, err)
+        log.error('multistream handshake failed', err)
       }
     })
   }

test/circuit/dialer.spec.js

@@ -24,7 +24,7 @@ const dirtyChai = require('dirty-chai')
 const expect = chai.expect
 chai.use(dirtyChai)
 
-describe(`dialer tests`, function () {
+describe('dialer tests', function () {
   let dialer
 
   beforeEach(() => {
@@ -35,22 +35,22 @@ describe(`dialer tests`, function () {
     sinon.restore()
   })
 
-  describe(`.dial`, function () {
+  describe('.dial', function () {
     beforeEach(function () {
       dialer.relayPeers = new Map()
       dialer.relayPeers.set(nodes.node2.id, new Connection())
       dialer.relayPeers.set(nodes.node3.id, new Connection())
       dialer.dial.callThrough()
     })
 
-    it(`fail on non circuit addr`, function () {
+    it('fail on non circuit addr', function () {
       const dstMa = multiaddr(`/ipfs/${nodes.node4.id}`)
       expect(() => dialer.dial(dstMa, (err) => {
         err.to.match(/invalid circuit address/)
       }))
     })
 
-    it(`dial a peer`, function (done) {
+    it('dial a peer', function (done) {
       const dstMa = multiaddr(`/p2p-circuit/ipfs/${nodes.node3.id}`)
       dialer._dialPeer.callsFake(function (dstMa, relay, callback) {
         return callback(null, dialer.relayPeers.get(nodes.node3.id))
@@ -63,7 +63,7 @@ describe(`dialer tests`, function () {
       })
     })
 
-    it(`dial a peer over the specified relay`, function (done) {
+    it('dial a peer over the specified relay', function (done) {
       const dstMa = multiaddr(`/ipfs/${nodes.node3.id}/p2p-circuit/ipfs/${nodes.node4.id}`)
       dialer._dialPeer.callsFake(function (dstMa, relay, callback) {
         expect(relay.toString()).to.equal(`/ipfs/${nodes.node3.id}`)
@@ -78,7 +78,7 @@ describe(`dialer tests`, function () {
     })
   })
 
-  describe(`.canHop`, function () {
+  describe('.canHop', function () {
     let fromConn = null
     const peer = new PeerInfo(PeerId.createFromB58String('QmQWqGdndSpAkxfk8iyiJyz3XXGkrDNujvc8vEst3baubA'))
 
@@ -94,7 +94,7 @@ describe(`dialer tests`, function () {
       dialer._dialRelayHelper.callThrough()
     })
 
-    it(`should handle successful CAN_HOP`, (done) => {
+    it('should handle successful CAN_HOP', (done) => {
       dialer._dialRelay.callsFake((_, cb) => {
         pull(
           values([{
@@ -114,7 +114,7 @@ describe(`dialer tests`, function () {
       })
     })
 
-    it(`should handle failed CAN_HOP`, function (done) {
+    it('should handle failed CAN_HOP', function (done) {
       dialer._dialRelay.callsFake((_, cb) => {
         pull(
           values([{
@@ -135,7 +135,7 @@ describe(`dialer tests`, function () {
     })
   })
 
-  describe(`._dialPeer`, function () {
+  describe('._dialPeer', function () {
     beforeEach(function () {
       dialer.relayPeers = new Map()
       dialer.relayPeers.set(nodes.node1.id, new Connection())
@@ -144,14 +144,14 @@ describe(`dialer tests`, function () {
       dialer._dialPeer.callThrough()
     })
 
-    it(`should dial a peer over any relay`, function (done) {
+    it('should dial a peer over any relay', function (done) {
       const dstMa = multiaddr(`/ipfs/${nodes.node4.id}`)
       dialer._negotiateRelay.callsFake(function (conn, dstMa, callback) {
         if (conn === dialer.relayPeers.get(nodes.node3.id)) {
           return callback(null, dialer.relayPeers.get(nodes.node3.id))
         }
 
-        callback(new Error(`error`))
+        callback(new Error('error'))
       })
 
       dialer._dialPeer(dstMa, (err, conn) => {
@@ -162,22 +162,22 @@ describe(`dialer tests`, function () {
       })
     })
 
-    it(`should fail dialing a peer over any relay`, function (done) {
+    it('should fail dialing a peer over any relay', function (done) {
       const dstMa = multiaddr(`/ipfs/${nodes.node4.id}`)
       dialer._negotiateRelay.callsFake(function (conn, dstMa, callback) {
-        callback(new Error(`error`))
+        callback(new Error('error'))
       })
 
       dialer._dialPeer(dstMa, (err, conn) => {
         expect(conn).to.be.undefined()
         expect(err).to.not.be.null()
-        expect(err).to.equal(`no relay peers were found or all relays failed to dial`)
+        expect(err).to.equal('no relay peers were found or all relays failed to dial')
         done()
       })
     })
   })
 
-  describe(`._negotiateRelay`, function () {
+  describe('._negotiateRelay', function () {
     const dstMa = multiaddr(`/ipfs/${nodes.node4.id}`)
 
     let conn = null
@@ -188,7 +188,7 @@ describe(`dialer tests`, function () {
       PeerId.createFromJSON(nodes.node4, (_, peerId) => {
         PeerInfo.create(peerId, (err, peerInfo) => {
           peer = peerInfo
-          peer.multiaddrs.add(`/p2p-circuit/ipfs/QmSswe1dCFRepmhjAMR5VfHeokGLcvVggkuDJm7RMfJSrE`)
+          peer.multiaddrs.add('/p2p-circuit/ipfs/QmSswe1dCFRepmhjAMR5VfHeokGLcvVggkuDJm7RMfJSrE')
           done(err)
         })
       })
@@ -202,12 +202,12 @@ describe(`dialer tests`, function () {
       dialer.relayConns = new Map()
       dialer._negotiateRelay.callThrough()
       dialer._dialRelayHelper.callThrough()
-      peer = new PeerInfo(PeerId.createFromB58String(`QmSswe1dCFRepmhjAMR5VfHeokGLcvVggkuDJm7RMfJSrE`))
+      peer = new PeerInfo(PeerId.createFromB58String('QmSswe1dCFRepmhjAMR5VfHeokGLcvVggkuDJm7RMfJSrE'))
       p = pair()
       conn = new Connection(p[1])
     })
 
-    it(`should write the correct dst addr`, function (done) {
+    it('should write the correct dst addr', function (done) {
       dialer._dialRelay.callsFake((_, cb) => {
         pull(
           p[0],
@@ -228,7 +228,7 @@ describe(`dialer tests`, function () {
       dialer._negotiateRelay(peer, dstMa, done)
     })
 
-    it(`should negotiate relay`, function (done) {
+    it('should negotiate relay', function (done) {
       dialer._dialRelay.callsFake((_, cb) => {
         pull(
           p[0],
@@ -253,7 +253,7 @@ describe(`dialer tests`, function () {
       })
     })
 
-    it(`should fail with an invalid peer id`, function (done) {
+    it('should fail with an invalid peer id', function (done) {
       const dstMa = multiaddr('/ip4/127.0.0.1/tcp/4001')
       dialer._dialRelay.callsFake((_, cb) => {
         pull(
@@ -279,7 +279,7 @@ describe(`dialer tests`, function () {
       })
     })
 
-    it(`should handle failed relay negotiation`, function (done) {
+    it('should handle failed relay negotiation', function (done) {
       dialer._dialRelay.callsFake((_, cb) => {
         cb(null, conn)
         pull(
@@ -295,7 +295,7 @@ describe(`dialer tests`, function () {
       dialer._negotiateRelay(peer, dstMa, (err, conn) => {
         expect(err).to.not.be.null()
         expect(err).to.be.an.instanceOf(Error)
-        expect(err.message).to.be.equal(`Got 400 error code trying to dial over relay`)
+        expect(err.message).to.be.equal('Got 400 error code trying to dial over relay')
         done()
       })
     })