Merge pull request #49 from ChainSafe/refactor/async

refactor: async migration

Author: wemeetagain

README.md

@@ -29,20 +29,29 @@ Gossipsub is an implementation of pubsub based on meshsub and floodsub. You can
 ```javascript
 const Gossipsub = require('libp2p-gossipsub')
 
-const gsub = new Gossipsub(node)
+const registrar = {
+  handle: (multicodecs, handle) => {
+    // register multicodec to libp2p
+    // handle function is called everytime a remote peer opens a stream to the peer.
+  },
+  register: (multicodecs, handlers) => {
+    // handlers will be used to notify pubsub of peer connection establishment or closing
+  },
+  unregister: (id) => {
 
-gsub.start((err) => {
-  if (err) {
-    console.log('Upsy', err)
   }
-  gsub.on('fruit', (data) => {
-    console.log(data)
-  })
-  gsub.subscribe('fruit')
+}
 
-  gsub.publish('fruit', new Buffer('banana'))
+const gsub = new Gossipsub(peerInfo, registrar, options)
+
+await gsub.start()
+
+gsub.on('fruit', (data) => {
+  console.log(data)
 })
+gsub.subscribe('fruit')
 
+gsub.publish('fruit', new Buffer('banana'))
 ```
 
 ## API
@@ -51,14 +60,16 @@ gsub.start((err) => {
 
 ```js
 const options = {…}
-const gossipsub = new Gossipsub(libp2pNode, options)
+const gossipsub = new Gossipsub(peerInfo, registrar, options)
 ```
 
 Options is an optional object with the following key-value pairs:
 
 * **`fallbackToFloodsub`**: boolean identifying whether the node should fallback to the floodsub protocol, if another connecting peer does not support gossipsub (defaults to **true**).
 * **`emitSelf`**: boolean identifying whether the node should emit to self on publish, in the event of the topic being subscribed (defaults to **false**).
 
+For the remaining API, see https://github.com/libp2p/js-libp2p-pubsub
+
 ## Contribute
 
 This module is actively under development. Please check out the issues and submit PRs!

package.json

@@ -32,35 +32,33 @@
     "lint"
   ],
   "dependencies": {
-    "async": "^2.6.2",
-    "err-code": "^1.1.2",
-    "libp2p-floodsub": "~0.17.1",
-    "libp2p-pubsub": "~0.2.0",
-    "multistream-select": "~0.14.6",
-    "peer-id": "~0.12.2",
-    "peer-info": "~0.15.1",
+    "debug": "^4.1.1",
+    "err-code": "^2.0.0",
+    "it-length-prefixed": "^2.0.0",
+    "it-pipe": "^1.0.1",
+    "libp2p-floodsub": "^0.19.0",
+    "libp2p-pubsub": "~0.3.1",
+    "p-map": "^3.0.0",
+    "peer-id": "~0.13.3",
+    "peer-info": "~0.17.0",
     "protons": "^1.0.1",
-    "pull-length-prefixed": "^1.3.3",
-    "pull-stream": "^3.6.13"
+    "time-cache": "^0.3.0"
   },
   "devDependencies": {
-    "@types/chai": "^4.1.7",
+    "@types/chai": "^4.2.3",
     "@types/mocha": "^5.2.7",
-    "aegir": "^20.0.0",
+    "aegir": "^20.4.1",
     "benchmark": "^2.1.4",
     "chai": "^4.2.0",
     "chai-spies": "^1.0.0",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
-    "libp2p": "~0.25.5",
-    "libp2p-secio": "~0.11.1",
-    "libp2p-spdy": "~0.13.3",
-    "libp2p-tcp": "~0.13.0",
-    "libp2p-websockets": "~0.12.2",
+    "it-pair": "^1.0.0",
     "lodash": "^4.17.15",
-    "mocha": "^5.2.0",
+    "mocha": "^6.2.1",
+    "p-times": "^2.1.0",
     "promisify-es6": "^1.0.3",
-    "sinon": "^7.3.2"
+    "sinon": "^7.5.0"
   },
   "contributors": [
     "Cayman <[email protected]>",

src/heartbeat.js

@@ -12,11 +12,11 @@ class Heartbeat {
     this.gossipsub = gossipsub
   }
 
-  start (callback) {
+  start () {
     if (this._heartbeatTimer) {
       const errMsg = 'Heartbeat timer is already running'
       this.gossipsub.log(errMsg)
-      return callback(errcode(new Error(errMsg), 'ERR_HEARTBEAT_ALREADY_RUNNING'))
+      throw errcode(new Error(errMsg), 'ERR_HEARTBEAT_ALREADY_RUNNING')
     }
 
     const heartbeatTimer = {
@@ -25,39 +25,35 @@ class Heartbeat {
       runPeriodically: (fn, period) => {
         heartbeatTimer._timeoutId = setInterval(fn, period)
       },
-      cancel: (cb) => {
+      cancel: () => {
         clearTimeout(heartbeatTimer._timeoutId)
-        cb()
       }
     }
 
     const heartbeat = this._heartbeat.bind(this)
+
     setTimeout(() => {
       heartbeat()
       heartbeatTimer.runPeriodically(heartbeat, constants.GossipSubHeartbeatInterval)
     }, constants.GossipSubHeartbeatInitialDelay)
 
     this._heartbeatTimer = heartbeatTimer
-    callback()
   }
 
   /**
    * Unmounts the gossipsub protocol and shuts down every connection
-   *
    * @override
-   * @param {Function} callback
    * @returns {void}
    */
-  stop (callback) {
+  stop () {
     if (!this._heartbeatTimer) {
       const errMsg = 'Heartbeat timer is not running'
       this.gossipsub.log(errMsg)
-      return callback(errcode(new Error(errMsg), 'ERR_HEARTBEAT_NO_RUNNING'))
+      throw errcode(new Error(errMsg), 'ERR_HEARTBEAT_NO_RUNNING')
     }
-    this._heartbeatTimer.cancel(() => {
-      this._heartbeatTimer = null
-      callback()
-    })
+
+    this._heartbeatTimer.cancel()
+    this._heartbeatTimer = null
   }
 
   /**