refactor: pubsub (#467)

* feat: peer-store v0

* chore: apply suggestions from code review

Co-Authored-By: Jacob Heun <jacobheun@gmail.com>

* chore: address review

* refactor: pubsub subsystem

* chore: address review

* chore: use topology interface

* chore: address review

* chore: address review

* chore: simplify tests

README.md

@@ -211,22 +211,18 @@ class Node extends Libp2p {
 
 **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)`
+#### Create a Node - `Libp2p.create(options)`
 
 > Behaves exactly like `new Libp2p(options)`, but doesn't require a PeerInfo. One will be generated instead
 
 ```js
-const { createLibp2p } = require('libp2p')
-createLibp2p(options, (err, libp2p) => {
-  if (err) throw err
-  libp2p.start((err) => {
-    if (err) throw err
-  })
-})
+const { create } = require('libp2p')
+const libp2p = await create(options)
+
+await libp2p.start()
 ```
 
 - `options`: Object of libp2p configuration options
-- `callback`: Function with signature `function (Error, Libp2p) {}`
 
 #### Create a Node alternative - `new Libp2p(options)`
 

package.json

@@ -55,13 +55,14 @@
     "it-protocol-buffers": "^0.2.0",
     "latency-monitor": "~0.2.1",
     "libp2p-crypto": "^0.17.1",
-    "libp2p-interfaces": "^0.1.3",
+    "libp2p-interfaces": "^0.1.5",
     "mafmt": "^7.0.0",
     "merge-options": "^1.0.1",
     "moving-average": "^1.0.0",
     "multiaddr": "^7.1.0",
     "multistream-select": "^0.15.0",
     "once": "^1.4.0",
+    "p-map": "^3.0.0",
     "p-queue": "^6.1.1",
     "p-settle": "^3.1.0",
     "peer-id": "^0.13.3",
@@ -90,8 +91,8 @@
     "libp2p-bootstrap": "^0.9.7",
     "libp2p-delegated-content-routing": "^0.2.2",
     "libp2p-delegated-peer-routing": "^0.2.2",
-    "libp2p-floodsub": "~0.17.0",
-    "libp2p-gossipsub": "~0.0.4",
+    "libp2p-floodsub": "^0.19.0",
+    "libp2p-gossipsub": "ChainSafe/gossipsub-js#beta/async",
     "libp2p-kad-dht": "^0.15.3",
     "libp2p-mdns": "^0.12.3",
     "libp2p-mplex": "^0.9.1",
@@ -103,6 +104,7 @@
     "lodash.times": "^4.3.2",
     "nock": "^10.0.6",
     "p-defer": "^3.0.0",
+    "p-wait-for": "^3.1.0",
     "portfinder": "^1.0.20",
     "pull-goodbye": "0.0.2",
     "pull-length-prefixed": "^1.3.3",

src/index.js

@@ -1,15 +1,14 @@
 'use strict'
 
 const FSM = require('fsm-event')
-const EventEmitter = require('events').EventEmitter
+const { EventEmitter } = require('events')
 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 nextTick = require('async/nextTick')
 
 const PeerInfo = require('peer-info')
 const multiaddr = require('multiaddr')
@@ -66,6 +65,8 @@ class Libp2p extends EventEmitter {
     this._transport = [] // Transport instances/references
     this._discovery = [] // Discovery service instances/references
 
+    this.peerStore = new PeerStore()
+
     // create the switch, and listen for errors
     this._switch = new Switch(this.peerInfo, this.peerStore, this._options.switch)
 
@@ -147,7 +148,7 @@ class Libp2p extends EventEmitter {
     }
 
     // start pubsub
-    if (this._modules.pubsub && this._config.pubsub.enabled !== false) {
+    if (this._modules.pubsub) {
       this.pubsub = pubsub(this, this._modules.pubsub, this._config.pubsub)
     }
 
@@ -251,6 +252,7 @@ class Libp2p extends EventEmitter {
     this.state('stop')
 
     try {
+      this.pubsub && await this.pubsub.stop()
       await this.transportManager.close()
       await this._switch.stop()
     } catch (err) {
@@ -385,10 +387,16 @@ class Libp2p extends EventEmitter {
     const multiaddrs = this.peerInfo.multiaddrs.toArray()
 
     // Start parallel tasks
+    const tasks = [
+      this.transportManager.listen(multiaddrs)
+    ]
+
+    if (this._config.pubsub.enabled) {
+      this.pubsub && this.pubsub.start()
+    }
+
     try {
-      await Promise.all([
-        this.transportManager.listen(multiaddrs)
-      ])
+      await Promise.all(tasks)
     } catch (err) {
       log.error(err)
       this.emit('error', err)
@@ -483,16 +491,15 @@ module.exports = Libp2p
  * Like `new Libp2p(options)` except it will create a `PeerInfo`
  * instance if one is not provided in options.
  * @param {object} options Libp2p configuration options
- * @param {function(Error, Libp2p)} callback
- * @returns {void}
+ * @returns {Libp2p}
  */
-module.exports.createLibp2p = promisify((options, callback) => {
+module.exports.create = async (options = {}) => {
   if (options.peerInfo) {
-    return nextTick(callback, null, new Libp2p(options))
+    return new Libp2p(options)
   }
-  PeerInfo.create((err, peerInfo) => {
-    if (err) return callback(err)
-    options.peerInfo = peerInfo
-    callback(null, new Libp2p(options))
-  })
-})
+
+  const peerInfo = await PeerInfo.create()
+
+  options.peerInfo = peerInfo
+  return new Libp2p(options)
+}