feat: integrate gossipsub by default

README.md

@@ -119,6 +119,7 @@ const MPLEX = require('libp2p-mplex')
 const SECIO = require('libp2p-secio')
 const MulticastDNS = require('libp2p-mdns')
 const DHT = require('libp2p-kad-dht')
+const GossipSub = require('libp2p-gossipsub')
 const defaultsDeep = require('@nodeutils/defaults-deep')
 const Protector = require('libp2p-pnet')
 const DelegatedPeerRouter = require('libp2p-delegated-peer-routing')
@@ -154,7 +155,8 @@ class Node extends Libp2p {
         peerDiscovery: [
           MulticastDNS
         ],
-        dht: DHT                      // DHT enables PeerRouting, ContentRouting and DHT itself components
+        dht: DHT,                      // DHT enables PeerRouting, ContentRouting and DHT itself components
+        pubsub: GossipSub
       },
 
       // libp2p config options (typically found on a config.json)
@@ -187,9 +189,8 @@ class Node extends Libp2p {
             timeout: 10e3
           }
         },
-        // Enable/Disable Experimental features
-        EXPERIMENTAL: {               // Experimental features ("behind a flag")
-          pubsub: false
+        pubsub: {
+          enabled: true
         }
       }
     }

package.json

@@ -48,7 +48,6 @@
     "err-code": "^1.1.2",
     "fsm-event": "^2.1.0",
     "libp2p-connection-manager": "^0.1.0",
-    "libp2p-floodsub": "^0.16.1",
     "libp2p-ping": "^0.8.5",
     "libp2p-switch": "^0.42.12",
     "libp2p-websockets": "^0.12.2",
@@ -74,6 +73,8 @@
     "libp2p-circuit": "^0.3.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-kad-dht": "^0.15.3",
     "libp2p-mdns": "^0.12.3",
     "libp2p-mplex": "^0.8.4",
@@ -84,6 +85,7 @@
     "libp2p-websocket-star": "~0.10.2",
     "libp2p-websocket-star-rendezvous": "~0.3.0",
     "lodash.times": "^4.3.2",
+    "merge-options": "^1.0.1",
     "nock": "^10.0.6",
     "pull-goodbye": "0.0.2",
     "pull-mplex": "^0.1.2",

src/config.js

@@ -19,6 +19,7 @@ const modulesSchema = s({
   connProtector: s.union(['undefined', s.interface({ protect: 'function' })]),
   contentRouting: optional(list(['object'])),
   dht: optional(s('null|function|object')),
+  pubsub: optional(s('null|function|object')),
   peerDiscovery: optional(list([s('object|function')])),
   peerRouting: optional(list(['object'])),
   streamMuxer: optional(list([s('object|function')])),
@@ -59,12 +60,10 @@ const configSchema = s({
       timeout: 10e3
     }
   }),
-  // Experimental config
-  EXPERIMENTAL: s({
-    pubsub: 'boolean'
-  }, {
-    // Experimental defaults
-    pubsub: false
+  // Pubsub config
+  pubsub: s('object?', {
+    // DHT defaults
+    enabled: false
   })
 }, {})
 

src/index.js

@@ -122,9 +122,9 @@ class Libp2p extends EventEmitter {
       })
     }
 
-    // enable/disable pubsub
-    if (this._config.EXPERIMENTAL.pubsub) {
-      this.pubsub = pubsub(this)
+    // start pubsub
+    if (this._modules.pubsub && this._config.pubsub.enabled !== false) {
+      this.pubsub = pubsub(this, this._modules.pubsub)
     }
 
     // Attach remaining APIs
@@ -403,8 +403,8 @@ class Libp2p extends EventEmitter {
         }
       },
       (cb) => {
-        if (this._floodSub) {
-          return this._floodSub.start(cb)
+        if (this.pubsub) {
+          return this.pubsub.start(cb)
         }
         cb()
       },
@@ -442,8 +442,8 @@ class Libp2p extends EventEmitter {
         )
       },
       (cb) => {
-        if (this._floodSub) {
-          return this._floodSub.stop(cb)
+        if (this.pubsub) {
+          return this.pubsub.stop(cb)
         }
         cb()
       },

src/pubsub.js

@@ -2,15 +2,12 @@
 
 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')
 
-module.exports = (node) => {
-  const floodSub = new FloodSub(node)
-
-  node._floodSub = floodSub
+module.exports = (node, Pubsub) => {
+  const pubsub = new Pubsub(node, { emitSelf: true })
 
   return {
     /**
@@ -41,16 +38,16 @@ module.exports = (node) => {
         options = {}
       }
 
-      if (!node.isStarted() && !floodSub.started) {
+      if (!node.isStarted() && !pubsub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
 
       function subscribe (cb) {
-        if (floodSub.listenerCount(topic) === 0) {
-          floodSub.subscribe(topic)
+        if (pubsub.listenerCount(topic) === 0) {
+          pubsub.subscribe(topic)
         }
 
-        floodSub.on(topic, handler)
+        pubsub.on(topic, handler)
         nextTick(cb)
       }
 
@@ -80,18 +77,18 @@ module.exports = (node) => {
      * libp2p.unsubscribe(topic, handler, callback)
      */
     unsubscribe: promisify((topic, handler, callback) => {
-      if (!node.isStarted() && !floodSub.started) {
+      if (!node.isStarted() && !pubsub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
 
       if (!handler) {
-        floodSub.removeAllListeners(topic)
+        pubsub.removeAllListeners(topic)
       } else {
-        floodSub.removeListener(topic, handler)
+        pubsub.removeListener(topic, handler)
       }
 
-      if (floodSub.listenerCount(topic) === 0) {
-        floodSub.unsubscribe(topic)
+      if (pubsub.listenerCount(topic) === 0) {
+        pubsub.unsubscribe(topic)
       }
 
       if (typeof callback === 'function') {
@@ -102,29 +99,31 @@ module.exports = (node) => {
     }),
 
     publish: promisify((topic, data, callback) => {
-      if (!node.isStarted() && !floodSub.started) {
+      if (!node.isStarted() && !pubsub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
 
-      if (!Buffer.isBuffer(data)) {
-        return nextTick(callback, errCode(new Error('data must be a Buffer'), 'ERR_DATA_IS_NOT_A_BUFFER'))
+      try {
+        data = Buffer.from(data)
+      } catch (err) {
+        return nextTick(callback, errCode(new Error('data must be convertible to a Buffer'), 'ERR_DATA_IS_NOT_VALID'))
       }
 
-      floodSub.publish(topic, data, callback)
+      pubsub.publish(topic, data, callback)
     }),
 
     ls: promisify((callback) => {
-      if (!node.isStarted() && !floodSub.started) {
+      if (!node.isStarted() && !pubsub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
 
-      const subscriptions = Array.from(floodSub.subscriptions)
+      const subscriptions = Array.from(pubsub.subscriptions)
 
       nextTick(() => callback(null, subscriptions))
     }),
 
     peers: promisify((topic, callback) => {
-      if (!node.isStarted() && !floodSub.started) {
+      if (!node.isStarted() && !pubsub.started) {
         return nextTick(callback, errCode(new Error(messages.NOT_STARTED_YET), codes.PUBSUB_NOT_STARTED))
       }
 
@@ -133,15 +132,19 @@ module.exports = (node) => {
         topic = null
       }
 
-      const peers = Array.from(floodSub.peers.values())
+      const peers = Array.from(pubsub.peers.values())
         .filter((peer) => topic ? peer.topics.has(topic) : true)
         .map((peer) => peer.info.id.toB58String())
 
       nextTick(() => callback(null, peers))
     }),
 
     setMaxListeners (n) {
-      return floodSub.setMaxListeners(n)
-    }
+      return pubsub.setMaxListeners(n)
+    },
+
+    start: (cb) => pubsub.start(cb),
+
+    stop: (cb) => pubsub.stop(cb)
   }
 }

test/config.spec.js

@@ -82,8 +82,8 @@ describe('configuration', () => {
         peerDiscovery: {
           autoDial: true
         },
-        EXPERIMENTAL: {
-          pubsub: false
+        pubsub: {
+          enabled: false
         },
         dht: {
           kBucketSize: 20,
@@ -144,8 +144,8 @@ describe('configuration', () => {
             enabled: true
           }
         },
-        EXPERIMENTAL: {
-          pubsub: false
+        pubsub: {
+          enabled: false
         },
         dht: {
           kBucketSize: 20,
@@ -269,8 +269,8 @@ describe('configuration', () => {
         dht: DHT
       },
       config: {
-        EXPERIMENTAL: {
-          pubsub: false
+        pubsub: {
+          enabled: false
         },
         peerDiscovery: {
           autoDial: true

test/create.spec.js

@@ -19,8 +19,8 @@ describe('libp2p creation', () => {
   it('should be able to start and stop successfully', (done) => {
     createNode([], {
       config: {
-        EXPERIMENTAL: {
-          pubsub: true
+        pubsub: {
+          enabled: true
         },
         dht: {
           enabled: true
@@ -32,7 +32,7 @@ describe('libp2p creation', () => {
       const sw = node._switch
       const cm = node.connectionManager
       const dht = node._dht
-      const pub = node._floodSub
+      const pub = node.pubsub
 
       sinon.spy(sw, 'start')
       sinon.spy(cm, 'start')
@@ -77,13 +77,13 @@ describe('libp2p creation', () => {
   it('should not create disabled modules', (done) => {
     createNode([], {
       config: {
-        EXPERIMENTAL: {
-          pubsub: false
+        pubsub: {
+          enabled: false
         }
       }
     }, (err, node) => {
       expect(err).to.not.exist()
-      expect(node._floodSub).to.not.exist()
+      expect(node._pubsub).to.not.exist()
       done()
     })
   })