feat: promisify all api methods that accept callbacks (#381)

* feat: promisify all api methods that accept callbacks

This is a stop-gap until the full async/await migration can be
completed.  It means we can refactor tests of other modules that
depend on this module without having to mix async flow control
strategies.

N.b. some methods that were previously callable without callbacks
(e.g. `node.start()`, `node.stop()`, etc) now require callbacks
otherwise a promise is returned which, if rejected, can cause
`unhandledPromiseRejection` events and lead to memory leaks.

* docs: add a global note to the api about promisify

* fix: update the logic for unsubscribe

* test(fix): correct pubsub unsubscribe usage for api change

* test(fix): update content routing tests for latest delegate version

README.md

@@ -204,6 +204,8 @@ class Node extends Libp2p {
 
 ### API
 
+**IMPORTANT NOTE**: All the methods listed in the API section that take a callback are also now Promisified. Libp2p is migrating away from callbacks to async/await, and in a future release (that will be announced in advance), callback support will be removed entirely. You can follow progress of the async/await endeavor at https://github.com/ipfs/js-ipfs/issues/1670.
+
 #### Create a Node - `Libp2p.createLibp2p(options, callback)`
 
 > Behaves exactly like `new Libp2p(options)`, but doesn't require a PeerInfo. One will be generated instead

package.json

@@ -58,6 +58,7 @@
     "peer-book": "^0.9.1",
     "peer-id": "^0.12.2",
     "peer-info": "^0.15.1",
+    "promisify-es6": "^1.0.3",
     "superstruct": "^0.6.0"
   },
   "devDependencies": {

src/content-routing.js

@@ -3,6 +3,7 @@
 const tryEach = require('async/tryEach')
 const parallel = require('async/parallel')
 const errCode = require('err-code')
+const promisify = require('promisify-es6')
 
 module.exports = (node) => {
   const routers = node._modules.contentRouting || []
@@ -24,7 +25,7 @@ module.exports = (node) => {
      * @param {function(Error, Result<Array>)} callback
      * @returns {void}
      */
-    findProviders: (key, options, callback) => {
+    findProviders: promisify((key, options, callback) => {
       if (typeof options === 'function') {
         callback = options
         options = {}
@@ -60,7 +61,7 @@ module.exports = (node) => {
         results = results || []
         callback(null, results)
       })
-    },
+    }),
 
     /**
      * Iterates over all content routers in parallel to notify it is
@@ -70,7 +71,7 @@ module.exports = (node) => {
      * @param {function(Error)} callback
      * @returns {void}
      */
-    provide: (key, callback) => {
+    provide: promisify((key, callback) => {
       if (!routers.length) {
         return callback(errCode(new Error('No content routers available'), 'NO_ROUTERS_AVAILABLE'))
       }
@@ -78,6 +79,6 @@ module.exports = (node) => {
       parallel(routers.map((router) => {
         return (cb) => router.provide(key, cb)
       }), callback)
-    }
+    })
   }
 }

src/dht.js

@@ -2,19 +2,20 @@
 
 const nextTick = require('async/nextTick')
 const errCode = require('err-code')
+const promisify = require('promisify-es6')
 
 const { messages, codes } = require('./errors')
 
 module.exports = (node) => {
   return {
-    put: (key, value, callback) => {
+    put: promisify((key, value, callback) => {
       if (!node._dht) {
         return nextTick(callback, errCode(new Error(messages.DHT_DISABLED), codes.DHT_DISABLED))
       }
 
       node._dht.put(key, value, callback)
-    },
+    }),
-    get: (key, options, callback) => {
+    get: promisify((key, options, callback) => {
       if (typeof options === 'function') {
         callback = options
         options = {}
@@ -25,8 +26,8 @@ module.exports = (node) => {
       }
 
       node._dht.get(key, options, callback)
-    },
+    }),
-    getMany: (key, nVals, options, callback) => {
+    getMany: promisify((key, nVals, options, callback) => {
       if (typeof options === 'function') {
         callback = options
         options = {}
@@ -37,6 +38,6 @@ module.exports = (node) => {
       }
 
       node._dht.getMany(key, nVals, options, callback)
-    }
+    })
   }
 }

src/get-peer-info.js

@@ -4,12 +4,13 @@ const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
 const errCode = require('err-code')
+const promisify = require('promisify-es6')
 
 module.exports = (node) => {
   /*
    * Helper method to check the data type of peer and convert it to PeerInfo
    */
-  return function (peer, callback) {
+  return promisify(function (peer, callback) {
     let p
     // PeerInfo
     if (PeerInfo.isPeerInfo(peer)) {
@@ -62,5 +63,5 @@ module.exports = (node) => {
     }
 
     callback(null, p)
-  }
+  })
 }

src/index.js

@@ -6,6 +6,7 @@ const debug = require('debug')
 const log = debug('libp2p')
 log.error = debug('libp2p:error')
 const errCode = require('err-code')
+const promisify = require('promisify-es6')
 
 const each = require('async/each')
 const series = require('async/series')
@@ -186,6 +187,13 @@ class Libp2p extends EventEmitter {
     })
 
     this._peerDiscovered = this._peerDiscovered.bind(this)
+
+    // promisify all instance methods
+    ;['start', 'stop', 'dial', 'dialProtocol', 'dialFSM', 'hangUp', 'ping'].forEach(method => {
+      this[method] = promisify(this[method], {
+        context: this
+      })
+    })
   }
 
   /**
@@ -557,7 +565,7 @@ module.exports = Libp2p
  * @param {function(Error, Libp2p)} callback
  * @returns {void}
  */
-module.exports.createLibp2p = (options, callback) => {
+module.exports.createLibp2p = promisify((options, callback) => {
   if (options.peerInfo) {
     return nextTick(callback, null, new Libp2p(options))
   }
@@ -566,4 +574,4 @@ module.exports.createLibp2p = (options, callback) => {
     options.peerInfo = peerInfo
     callback(null, new Libp2p(options))
   })
-}
+})

src/peer-routing.js

@@ -2,6 +2,7 @@
 
 const tryEach = require('async/tryEach')
 const errCode = require('err-code')
+const promisify = require('promisify-es6')
 
 module.exports = (node) => {
   const routers = node._modules.peerRouting || []
@@ -21,7 +22,7 @@ module.exports = (node) => {
      * @param {function(Error, Result<Array>)} callback
      * @returns {void}
      */
-    findPeer: (id, options, callback) => {
+    findPeer: promisify((id, options, callback) => {
       if (typeof options === 'function') {
         callback = options
         options = {}
@@ -53,6 +54,6 @@ module.exports = (node) => {
         results = results || []
         callback(null, results)
       })
-    }
+    })
   }
 }

src/pubsub.js

@@ -3,6 +3,7 @@
 const nextTick = require('async/nextTick')
 const { messages, codes } = require('./errors')
 const FloodSub = require('libp2p-floodsub')
+const promisify = require('promisify-es6')
 
 const errCode = require('err-code')
 
@@ -12,7 +13,7 @@ module.exports = (node) => {
   node._floodSub = floodSub
 
   return {
-    subscribe: (topic, options, handler, callback) => {
+    subscribe: promisify((topic, options, handler, callback) => {
       if (typeof options === 'function') {
         callback = handler
         handler = options
@@ -33,13 +34,36 @@ module.exports = (node) => {
       }
 
       subscribe(callback)
-    },
+    }),
 
-    unsubscribe: (topic, handler, callback) => {
+    /**
+     * Unsubscribes from a pubsub topic
+     *
+     * @param {string} topic
+     * @param {function|null} handler The handler to unsubscribe from
+     * @param {function} [callback] An optional callback
+     *
+     * @returns {Promise|void} A promise is returned if no callback is provided
+     *
+     * @example <caption>Unsubscribe a topic for all handlers</caption>
+     *
+     * // `null` must be passed until unsubscribe is no longer using promisify
+     * await libp2p.unsubscribe(topic, null)
+     *
+     * @example <caption>Unsubscribe a topic for 1 handler</caption>
+     *
+     * await libp2p.unsubscribe(topic, handler)
+     *
+     * @example <caption>Use a callback instead of the Promise api</caption>
+     *
+     * libp2p.unsubscribe(topic, handler, callback)
+     */
+    unsubscribe: promisify((topic, handler, callback) => {
       if (!node.isStarted() && !floodSub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
-      if (!handler && !callback) {
+
+      if (!handler) {
         floodSub.removeAllListeners(topic)
       } else {
         floodSub.removeListener(topic, handler)
@@ -50,11 +74,13 @@ module.exports = (node) => {
       }
 
       if (typeof callback === 'function') {
-        nextTick(() => callback())
+        return nextTick(() => callback())
       }
-    },
 
-    publish: (topic, data, callback) => {
+      return Promise.resolve()
+    }),
+
+    publish: promisify((topic, data, callback) => {
       if (!node.isStarted() && !floodSub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
@@ -64,9 +90,9 @@ module.exports = (node) => {
       }
 
       floodSub.publish(topic, data, callback)
-    },
+    }),
 
-    ls: (callback) => {
+    ls: promisify((callback) => {
       if (!node.isStarted() && !floodSub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
@@ -74,9 +100,9 @@ module.exports = (node) => {
       const subscriptions = Array.from(floodSub.subscriptions)
 
       nextTick(() => callback(null, subscriptions))
-    },
+    }),
 
-    peers: (topic, callback) => {
+    peers: promisify((topic, callback) => {
       if (!node.isStarted() && !floodSub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
@@ -91,7 +117,7 @@ module.exports = (node) => {
         .map((peer) => peer.info.id.toB58String())
 
       nextTick(() => callback(null, peers))
-    },
+    }),
 
     setMaxListeners (n) {
       return floodSub.setMaxListeners(n)

test/content-routing.node.js

@@ -185,19 +185,10 @@ describe('.contentRouting', () => {
       it('should be able to register as a provider', (done) => {
         const cid = new CID('QmU621oD8AhHw6t25vVyfYKmL9VV3PTgc52FngEhTGACFB')
         const mockApi = nock('http://0.0.0.0:60197')
-          // mock the swarm connect
-          .post('/api/v0/swarm/connect')
-          .query({
-            arg: `/ip4/0.0.0.0/tcp/60194/p2p-circuit/ipfs/${nodeA.peerInfo.id.toB58String()}`,
-            'stream-channels': true
-          })
-          .reply(200, {
-            Strings: [`connect ${nodeA.peerInfo.id.toB58String()} success`]
-          }, ['Content-Type', 'application/json'])
           // mock the refs call
           .post('/api/v0/refs')
           .query({
-            recursive: true,
+            recursive: false,
             arg: cid.toBaseEncodedString(),
             'stream-channels': true
           })
@@ -216,10 +207,11 @@ describe('.contentRouting', () => {
       it('should handle errors when registering as a provider', (done) => {
         const cid = new CID('QmU621oD8AhHw6t25vVyfYKmL9VV3PTgc52FngEhTGACFB')
         const mockApi = nock('http://0.0.0.0:60197')
-          // mock the swarm connect
+          // mock the refs call
-          .post('/api/v0/swarm/connect')
+          .post('/api/v0/refs')
           .query({
-            arg: `/ip4/0.0.0.0/tcp/60194/p2p-circuit/ipfs/${nodeA.peerInfo.id.toB58String()}`,
+            recursive: false,
+            arg: cid.toBaseEncodedString(),
             'stream-channels': true
           })
           .reply(502, 'Bad Gateway', ['Content-Type', 'application/json'])

test/fsm.spec.js

@@ -60,9 +60,9 @@ describe('libp2p state machine (fsm)', () => {
         node.once('stop', done)
 
         // stop the stopped node
-        node.stop()
+        node.stop(() => {})
       })
-      node.start()
+      node.start(() => {})
     })
 
     it('should callback with an error when it occurs on stop', (done) => {
@@ -79,7 +79,7 @@ describe('libp2p state machine (fsm)', () => {
       expect(2).checks(done)
 
       sinon.stub(node._switch, 'stop').callsArgWith(0, error)
-      node.start()
+      node.start(() => {})
     })
 
     it('should noop when starting a started node', (done) => {
@@ -89,13 +89,13 @@ describe('libp2p state machine (fsm)', () => {
         })
         node.once('start', () => {
           node.once('stop', done)
-          node.stop()
+          node.stop(() => {})
         })
 
         // start the started node
-        node.start()
+        node.start(() => {})
       })
-      node.start()
+      node.start(() => {})
     })
 
     it('should error on start with no transports', (done) => {
@@ -115,7 +115,7 @@ describe('libp2p state machine (fsm)', () => {
 
       expect(2).checks(done)
 
-      node.start()
+      node.start(() => {})
     })
 
     it('should not start if the switch fails to start', (done) => {
@@ -150,7 +150,7 @@ describe('libp2p state machine (fsm)', () => {
         })
       })
 
-      node.stop()
+      node.stop(() => {})
     })
 
     it('should not dial (fsm) when the node is stopped', (done) => {
@@ -162,7 +162,7 @@ describe('libp2p state machine (fsm)', () => {
         })
       })
 
-      node.stop()
+      node.stop(() => {})
     })
   })
 })

test/pubsub.node.js

@@ -117,12 +117,8 @@ describe('.pubsub', () => {
         (cb) => nodes[1].pubsub.publish('pubsub', data, cb),
         // Wait a moment before unsubscribing
         (cb) => setTimeout(cb, 500),
-        // unsubscribe on the first
+        // unsubscribe from all
-        (cb) => {
+        (cb) => nodes[0].pubsub.unsubscribe('pubsub', null, cb),
-          nodes[0].pubsub.unsubscribe('pubsub')
-          // Wait a moment to make sure the ubsubscribe-from-all worked
-          setTimeout(cb, 500)
-        },
         // Verify unsubscribed
         (cb) => {
           nodes[0].pubsub.ls((err, topics) => {