refactor: switch to async iterators

README.md

@@ -22,6 +22,7 @@ js-libp2p-floodsub
 - [Install](#install)
 - [Usage](#usage)
 - [API](#api)
+- [Events](#events)
 - [Contribute](#contribute)
 - [License](#license)
 
@@ -31,26 +32,48 @@ 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) => {
 
-fsub.start((err) => {
-  if (err) {
-    console.log('Upsy', err)
   }
-  fsub.on('fruit', (data) => {
-    console.log(data)
-  })
-  fsub.subscribe('fruit')
+}
 
-  fsub.publish('fruit', new Buffer('banana'))
+const fsub = new FloodSub(peerInfo, registrar, options)
+
+await fsub.start()
+
+fsub.on('fruit', (data) => {
+  console.log(data)
 })
+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:
@@ -68,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
+    })
+})()