refactor: switch to async iterators

BREAKING CHANGE: Switch to using async/await and async iterators for all the API. Moreover, gossipsub does not need the libp2p instance anymore, receiving a registerar that enables it to receive the necessary events from libp2p

Author: vasco-santos

package.json

@@ -32,33 +32,31 @@
     "lint"
   ],
   "dependencies": {
-    "async": "^2.6.2",
+    "debug": "^4.1.1",
     "err-code": "^2.0.0",
-    "libp2p-floodsub": "~0.18.0",
-    "libp2p-pubsub": "~0.2.1",
-    "multistream-select": "~0.14.6",
-    "peer-id": "~0.12.5",
-    "peer-info": "~0.15.1",
+    "it-length-prefixed": "^2.0.0",
+    "it-pipe": "^1.0.1",
+    "libp2p-floodsub": "libp2p/js-libp2p-floodsub#refactor/async",
+    "libp2p-pubsub": "libp2p/js-libp2p-pubsub#refactor/async",
+    "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.14"
+    "time-cache": "^0.3.0"
   },
   "devDependencies": {
     "@types/chai": "^4.2.3",
     "@types/mocha": "^5.2.7",
-    "aegir": "^20.3.1",
+    "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.26.2",
-    "libp2p-secio": "~0.11.1",
-    "libp2p-spdy": "~0.13.3",
-    "libp2p-tcp": "~0.13.2",
-    "libp2p-websockets": "~0.12.4",
+    "it-pair": "^1.0.0",
     "lodash": "^4.17.15",
     "mocha": "^6.2.1",
+    "p-times": "^2.1.0",
     "promisify-es6": "^1.0.3",
     "sinon": "^7.5.0"
   },

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
   }
 
   /**

src/index.js

@@ -3,6 +3,8 @@
 const assert = require('assert')
 const { utils } = require('libp2p-pubsub')
 
+const PeerInfo = require('peer-info')
+
 const BasicPubsub = require('./pubsub')
 const { MessageCache } = require('./messageCache')
 
@@ -12,15 +14,24 @@ const Heartbeat = require('./heartbeat')
 
 class GossipSub extends BasicPubsub {
   /**
-   * @param {Object} libp2p an instance of Libp2p
-   * @param {Object} options
-   * @param {bool} options.emitSelf if publish should emit to self, if subscribed, defaults to false
-   * @param {bool} options.gossipIncoming if incoming messages on a subscribed topic should be automatically gossiped, defaults to true
-   * @param {bool} options.fallbackToFloodsub if dial should fallback to floodsub, defaults to true
+   * @param {PeerInfo} peerInfo instance of the peer's PeerInfo
+   * @param {Object} registrar
+   * @param {function} registrar.register
+   * @param {function} registrar.unregister
+   * @param {Object} [options]
+   * @param {bool} [options.emitSelf] if publish should emit to self, if subscribed, defaults to false
+   * @param {bool} [options.gossipIncoming] if incoming messages on a subscribed topic should be automatically gossiped, defaults to true
+   * @param {bool} [options.fallbackToFloodsub] if dial should fallback to floodsub, defaults to true
    * @constructor
    */
-  constructor (libp2p, options) {
-    super('libp2p:gossipsub', constants.GossipSubID, libp2p, options)
+  constructor (peerInfo, registrar, options = {}) {
+    assert(PeerInfo.isPeerInfo(peerInfo), 'peer info must be an instance of `peer-info`')
+
+    // registrar handling
+    assert(registrar && typeof registrar.register === 'function', 'a register function must be provided in registrar')
+    assert(registrar && typeof registrar.unregister === 'function', 'a unregister function must be provided in registrar')
+
+    super('libp2p:gossipsub', constants.GossipSubID, peerInfo, registrar, options)
 
     /**
      * Map of topic meshes
@@ -71,37 +82,35 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Removes a peer from the router
-   *
    * @override
    * @param {Peer} peer
    * @returns {PeerInfo}
    */
   _removePeer (peer) {
     super._removePeer(peer)
-    // Only delete when no one else if referencing this peer.
-    if (peer._references === 0) {
-      // Remove this peer from the mesh
-      // eslint-disable-next-line no-unused-vars
-      for (const [_, peers] of this.mesh.entries()) {
-        peers.delete(peer)
-      }
-      // Remove this peer from the fanout
-      // eslint-disable-next-line no-unused-vars
-      for (const [_, peers] of this.fanout.entries()) {
-        peers.delete(peer)
-      }
 
-      // Remove from gossip mapping
-      this.gossip.delete(peer)
-      // Remove from control mapping
-      this.control.delete(peer)
+    // Remove this peer from the mesh
+    // eslint-disable-next-line no-unused-vars
+    for (const [_, peers] of this.mesh.entries()) {
+      peers.delete(peer)
     }
+
+    // Remove this peer from the fanout
+    // eslint-disable-next-line no-unused-vars
+    for (const [_, peers] of this.fanout.entries()) {
+      peers.delete(peer)
+    }
+
+    // Remove from gossip mapping
+    this.gossip.delete(peer)
+    // Remove from control mapping
+    this.control.delete(peer)
+
     return peer
   }
 
   /**
    * Handles an rpc control message from a peer
-   *
    * @param {Peer} peer
    * @param {rpc.RPC} rpc
    * @returns {void}
@@ -139,6 +148,7 @@ class GossipSub extends BasicPubsub {
     if (!this._options.gossipIncoming) {
       return
     }
+
     // Emit to floodsub peers
     this.peers.forEach((peer) => {
       if (peer.info.protocols.has(constants.FloodSubID) &&
@@ -168,10 +178,8 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Handles IHAVE messages
-   *
    * @param {Peer} peer
    * @param {Array<rpc.RPC.ControlIHave>} ihave
-   *
    * @returns {rpc.RPC.ControlIWant}
    */
   _handleIHave (peer, ihave) {
@@ -204,10 +212,8 @@ class GossipSub extends BasicPubsub {
   /**
    * Handles IWANT messages
    * Returns messages to send back to peer
-   *
    * @param {Peer} peer
    * @param {Array<rpc.RPC.ControlIWant>} iwant
-   *
    * @returns {Array<rpc.RPC.Message>}
    */
   _handleIWant (peer, iwant) {
@@ -234,12 +240,9 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Handles Graft messages
-   *
    * @param {Peer} peer
    * @param {Array<rpc.RPC.ControlGraft>} graft
-   *
    * @return {Array<rpc.RPC.ControlPrune>}
-   *
    */
   _handleGraft (peer, graft) {
     const prune = []
@@ -271,12 +274,9 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Handles Prune messages
-   *
    * @param {Peer} peer
    * @param {Array<rpc.RPC.ControlPrune>} prune
-   *
    * @returns {void}
-   *
    */
   _handlePrune (peer, prune) {
     prune.forEach(({ topicID }) => {
@@ -292,36 +292,28 @@ class GossipSub extends BasicPubsub {
   /**
    * Mounts the gossipsub protocol onto the libp2p node and sends our
    * our subscriptions to every peer connected
-   *
    * @override
-   * @param {Function} callback
-   * @returns {void}
-   *
+   * @returns {Promise}
    */
-  start (callback) {
-    super.start((err) => {
-      if (err) return callback(err)
-      this.heartbeat.start(callback)
-    })
+  async start () {
+    await super.start()
+    this.heartbeat.start()
   }
 
   /**
    * Unmounts the gossipsub protocol and shuts down every connection
-   *
    * @override
-   * @param {Function} callback
-   * @returns {void}
+   * @returns {Promise}
    */
-  stop (callback) {
-    super.stop((err) => {
-      if (err) return callback(err)
-      this.mesh = new Map()
-      this.fanout = new Map()
-      this.lastpub = new Map()
-      this.gossip = new Map()
-      this.control = new Map()
-      this.heartbeat.stop(callback)
-    })
+  async stop () {
+    await super.stop()
+    this.heartbeat.stop()
+
+    this.mesh = new Map()
+    this.fanout = new Map()
+    this.lastpub = new Map()
+    this.gossip = new Map()
+    this.control = new Map()
   }
 
   /**
@@ -355,7 +347,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Leave topics
-   *
    * @param {Array<string>|string} topics
    * @returns {void}
    */
@@ -431,7 +422,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Sends a GRAFT message to a peer
-   *
    * @param {Peer} peer
    * @param {String} topic
    * @returns {void}
@@ -447,7 +437,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Sends a PRUNE message to a peer
-   *
    * @param {Peer} peer
    * @param {String} topic
    * @returns {void}
@@ -505,7 +494,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Send graft and prune messages
-   *
    * @param {Map<Peer, Array<String>>} tograft
    * @param {Map<Peer, Array<String>>} toprune
    */
@@ -532,7 +520,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Emits gossip to peers in a particular topic
-   *
    * @param {String} topic
    * @param {Set<Peer>} peers - peers to exclude
    * @returns {void}
@@ -575,7 +562,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Adds new IHAVE messages to pending gossip
-   *
    * @param {Peer} peer
    * @param {Array<rpc.RPC.ControlIHave>} controlIHaveMsgs
    * @returns {void}
@@ -588,7 +574,6 @@ class GossipSub extends BasicPubsub {
 
   /**
    * Returns the current time in milliseconds
-   *
    * @returns {number}
    */
   _now () {