feat: use registrar

README.md

@@ -22,6 +22,7 @@ js-libp2p-floodsub
 - [Install](#install)
 - [Usage](#usage)
 - [API](#api)
+- [Events](#events)
 - [Contribute](#contribute)
 - [License](#license)
 
@@ -31,12 +32,22 @@ js-libp2p-floodsub
 > npm install libp2p-floodsub
 ```
 
-## Examples
+## Usage
 
 ```JavaScript
 const FloodSub = require('libp2p-floodsub')
 
-const fsub = new FloodSub(node)
+const registrar = {
+  register: (multicodec, handlers) => {
+    // register multicodec to libp2p
+    // handlers will be used to notify pubsub of peer connection establishment or closing
+  },
+  unregister: (multicodec) => {
+
+  }
+}
+
+const fsub = new FloodSub(peerInfo, registrar, options)
 
 await fsub.start()
 
@@ -48,6 +59,21 @@ fsub.subscribe('fruit')
 fsub.publish('fruit', new Buffer('banana'))
 ```
 
+## API
+
+### Create a floodsub implementation
+
+```js
+const options = {…}
+const floodsub = new Floodsub(peerInfo, registrar, options)
+```
+
+Options is an optional object with the following key-value pairs:
+
+* **`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
+
 ## Events
 
 Floodsub emits two kinds of events:
@@ -65,27 +91,11 @@ Floodsub emits two kinds of events:
   - `changes`: an array of `{ topicID: <topic>, subscribe: <boolean> }`
      eg `[ { topicID: 'fruit', subscribe: true }, { topicID: 'vegetables': false } ]`
 
-
-## API
-
-### Create a floodsub implementation
-
-```js
-const options = {…}
-const floodsub = new Floodsub(libp2pNode, options)
-```
-
-Options is an optional object with the following key-value pairs:
-
-* **`emitSelf`**: boolean identifying whether the node should emit to self on publish, in the event of the topic being subscribed (defaults to **false**).
-
-For more, see https://libp2p.github.io/js-libp2p-floodsub
-
 ## Contribute
 
-PRs are welcome!
+Feel free to join in. All welcome. Open an [issue](https://github.com/libp2p/js-libp2p-pubsub/issues)!
 
-Small note: If editing the Readme, please conform to the [standard-readme](https://github.com/RichardLitt/standard-readme) specification.
+This repository falls under the IPFS [Code of Conduct](https://github.com/ipfs/community/blob/master/code-of-conduct.md).
 
 ## License
 

benchmarks/index.js

@@ -2,81 +2,78 @@
 
 const Benchmark = require('benchmark')
 const crypto = require('crypto')
-const map = require('async/map')
-const parallel = require('async/parallel')
-const series = require('async/series')
 
-const PSG = require('../src')
-const utils = require('../test/utils')
+const DuplexPair = require('it-pair/duplex')
+
+const Floodsub = require('../src')
+const { multicodec } = require('../src')
+const { createPeerInfo } = require('../test/utils')
 
 const suite = new Benchmark.Suite('pubsub')
 
 // Simple benchmark, how many messages can we send from
 // one node to another.
 
-map([0, 1], (i, cb) => {
-  utils.createNode((err, node) => {
-    if (err) {
-      return cb(err)
+;(async () => {
+  const registrarRecordA = {}
+  const registrarRecordB = {}
+
+  const registrar = (registrarRecord) => ({
+    register: (multicodec, handlers) => {
+      registrarRecord[multicodec] = handlers
+    },
+    unregister: (multicodec) => {
+      delete registrarRecord[multicodec]
     }
+  })
 
-    const ps = new PSG(node)
+  const [peerInfoA, peerInfoB] = await Promise.all([
+    createPeerInfo(),
+    createPeerInfo()
+  ])
 
-    series([
-      (cb) => node.start(cb),
-      (cb) => ps.start(cb)
-    ], (err) => {
-      if (err) {
-        return cb(err)
-      }
+  const fsA = new Floodsub(peerInfoA, registrar(registrarRecordA))
+  const fsB = new Floodsub(peerInfoB, registrar(registrarRecordB))
 
-      cb(null, {
-        libp2p: node,
-        ps
-      })
-    })
-  })
-}, (err, peers) => {
-  if (err) {
-    throw err
-  }
-
-  parallel([
-    (cb) => peers[0].libp2p.dial(peers[1].libp2p.peerInfo, cb),
-    (cb) => setTimeout(() => {
-      peers[0].ps.subscribe('Z', () => {}, () => {})
-      peers[1].ps.subscribe('Z', () => {}, () => {})
-      cb(null, peers)
-    }, 200)
-  ], (err, res) => {
-    if (err) {
-      throw err
-    }
+  // Start pubsub
+  await Promise.all([
+    fsA.start(),
+    fsB.start()
+  ])
 
-    const peers = res[1]
+  // Connect floodsub nodes
+  const onConnectA = registrarRecordA[multicodec].onConnect
+  const onConnectB = registrarRecordB[multicodec].onConnect
 
-    suite.add('publish and receive', (deferred) => {
-      const onMsg = (msg) => {
-        deferred.resolve()
-        peers[1].ps.removeListener('Z', onMsg)
-      }
+  // Notice peers of connection
+  const [d0, d1] = DuplexPair()
+  onConnectA(peerInfoB, d0)
+  onConnectB(peerInfoA, d1)
 
-      peers[1].ps.on('Z', onMsg)
+  fsA.subscribe('Z')
+  fsB.subscribe('Z')
 
-      peers[0].ps.publish('Z', crypto.randomBytes(1024))
-    }, {
-      defer: true
-    })
+  suite.add('publish and receive', (deferred) => {
+    const onMsg = (msg) => {
+      deferred.resolve()
+      fsB.removeListener('Z', onMsg)
+    }
+
+    fsB.on('Z', onMsg)
 
-    suite
-      .on('cycle', (event) => {
-        console.log(String(event.target)) // eslint-disable-line
-      })
-      .on('complete', () => {
-        process.exit()
-      })
-      .run({
-        async: true
-      })
+    fsA.publish('Z', crypto.randomBytes(1024))
+  }, {
+    defer: true
   })
-})
+
+  suite
+    .on('cycle', (event) => {
+      console.log(String(event.target)) // eslint-disable-line
+    })
+    .on('complete', () => {
+      process.exit()
+    })
+    .run({
+      async: true
+    })
+})()

package.json

@@ -17,9 +17,6 @@
     "coverage": "aegir coverage",
     "coverage-publish": "aegir coverage --provider coveralls"
   },
-  "browser": {
-    "test/utils/nodejs-bundle": "./test/utils/browser-bundle.js"
-  },
   "files": [
     "src",
     "dist"
@@ -45,34 +42,28 @@
   },
   "homepage": "https://github.com/libp2p/js-libp2p-floodsub#readme",
   "devDependencies": {
-    "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-websocket-star": "~0.10.2",
-    "libp2p-websocket-star-rendezvous": "~0.4.1",
+    "it-pair": "^1.0.0",
     "lodash": "^4.17.15",
     "multiaddr": "^6.1.0",
-    "peer-id": "~0.12.5",
-    "peer-info": "~0.15.1",
+    "p-defer": "^3.0.0",
+    "peer-id": "~0.13.3",
+    "peer-info": "~0.17.0",
     "sinon": "^7.5.0"
   },
   "dependencies": {
-    "async": "^2.6.2",
     "debug": "^4.1.1",
-    "length-prefixed-stream": "^2.0.0",
+    "it-length-prefixed": "^2.0.0",
+    "it-pipe": "^1.0.1",
     "libp2p-pubsub": "libp2p/js-libp2p-pubsub#refactor/async",
     "p-map": "^3.0.0",
     "protons": "^1.0.1",
-    "pull-length-prefixed": "^1.3.3",
-    "pull-pushable": "^2.2.0",
-    "pull-stream": "^3.6.9"
+    "time-cache": "^0.3.0"
   },
   "contributors": [
     "Alan Shaw <alan.shaw@protocol.ai>",