feat: add topology interfaces (#7)

* feat: topology

* feat: multicodec-topology

* chore: address review

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

* chore: remove error from disconnect

* docs: topology

* chore: apply suggestions from code review

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

README.md

@@ -19,6 +19,7 @@
 - [Peer Discovery](./src/peer-discovery)
 - [Peer Routing](./src/peer-routing)
 - [Stream Muxer](./src/stream-muxer)
+- [Topology](./src/topology)
 - [Transport](./src/transport)
 
 ### Origin Repositories

src/topology/README.md

@@ -0,0 +1,141 @@
+interface-topology
+========================
+
+> Implementation of the topology interface used by the `js-libp2p` registrar.
+
+Topologies can be used in conjunction with `js-libp2p` to help shape its network and the overlays of its subsystems, such as pubsub and the DHT.
+
+## Table of Contents
+
+- [Implementations](#implementations)
+- [Install](#install)
+- [Modules using the interface](#modulesUsingTheInterface)
+- [Usage](#usage)
+- [Api](#api)
+
+## Implementations
+
+### Topology
+
+A libp2p topology with a group of common peers.
+
+### Multicodec Topology
+
+A libp2p topology with a group of peers that support the same protocol.
+
+## Install
+
+```sh
+$ npm install libp2p-interfaces
+```
+
+## Modules using the interface
+
+TBA
+
+## Usage
+
+###  Topology
+
+```js
+const Topology = require('libp2p-interfaces/src/topology')
+
+const toplogy = new Topology({
+  min: 0,
+  max: 50
+})
+```
+
+### Multicodec Topology
+
+```js
+const MulticodecTopology = require('libp2p-interfaces/src/topology/multicodec-topology')
+
+const toplogy = new MulticodecTopology({
+  min: 0,
+  max: 50,
+  multicodecs: ['/echo/1.0.0'],
+  handlers: {
+    onConnect: (peerInfo, conn) => {},
+    onDisconnect: (peerInfo) => {}
+  }
+})
+```
+
+## API
+
+The `MulticodecTopology` extends the `Topology`, which makes the `Topology` API a subset of the `MulticodecTopology` API.
+
+###  Topology
+
+- `Topology`
+  - `peers<Map<string, PeerInfo>>`: A Map of peers belonging to the topology.
+  - `disconnect<function(PeerInfo)>`: Called when a peer has been disconnected
+
+#### Constructor
+
+```js
+const toplogy = new Topology({
+  min: 0,
+  max: 50,
+  handlers: {
+    onConnect: (peerInfo, conn) => {},
+    onDisconnect: (peerInfo) => {}
+  }
+})
+```
+
+**Parameters**
+- `properties` is an `Object` containing the properties of the topology.
+  - `min` is a `number` with the minimum needed connections (default: 0)
+  - `max` is a `number` with the maximum needed connections (default: Infinity)
+  - `handlers` is an optional `Object` containing the handler called when a peer is connected or disconnected.
+    - `onConnect` is a `function` called everytime a peer is connected in the topology context.
+    - `onDisconnect` is a `function` called everytime a peer is disconnected in the topology context.
+
+#### Set a peer
+
+- `topology.peers.set(id, peerInfo)`
+
+Add a peer to the topology.
+
+**Parameters**
+- `id` is the `string` that identifies the peer to add.
+- `peerInfo` is the [PeerInfo][peer-info] of the peer to add.
+
+#### Notify about a peer disconnected event
+
+- `topology.disconnect(peerInfo)`
+
+**Parameters**
+- `peerInfo` is the [PeerInfo][peer-info] of the peer disconnected.
+
+###  Multicodec Topology
+
+- `MulticodecTopology`
+  - `registrar<Registrar>`: The `Registrar` of the topology. This is set by the `Registrar` during registration.
+  - `peers<Map<string, PeerInfo>>`: The Map of peers that belong to the topology
+  - `disconnect<function(PeerInfo)>`: Disconnects a peer from the topology.
+
+#### Constructor
+
+```js
+const toplogy = new MulticodecTopology({
+  min: 0,
+  max: 50,
+  multicodecs: ['/echo/1.0.0'],
+  handlers: {
+    onConnect: (peerInfo, conn) => {},
+    onDisconnect: (peerInfo) => {}
+  }
+})
+```
+
+**Parameters**
+- `properties` is an `Object` containing the properties of the topology.
+  - `min` is a `number` with the minimum needed connections (default: 0)
+  - `max` is a `number` with the maximum needed connections (default: Infinity)
+  - `multicodecs` is a `Array<String>` with the multicodecs associated with the topology.
+  - `handlers` is an optional `Object` containing the handler called when a peer is connected or disconnected.
+    - `onConnect` is a `function` called everytime a peer is connected in the topology context.
+    - `onDisconnect` is a `function` called everytime a peer is disconnected in the topology context.

src/topology/index.js

@@ -0,0 +1,44 @@
+'use strict'
+
+const noop = () => {}
+
+class Topology {
+  /**
+   * @param {Object} props
+   * @param {number} props.min minimum needed connections (default: 0)
+   * @param {number} props.max maximum needed connections (default: Infinity)
+   * @param {Object} [props.handlers]
+   * @param {function} [props.handlers.onConnect] protocol "onConnect" handler
+   * @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
+   * @constructor
+   */
+  constructor ({
+    min = 0,
+    max = Infinity,
+    handlers = {}
+  }) {
+    this.min = min
+    this.max = max
+
+    // Handlers
+    this._onConnect = handlers.onConnect || noop
+    this._onDisconnect = handlers.onDisconnect || noop
+
+    this.peers = new Map()
+  }
+
+  set registrar (registrar) {
+    this._registrar = registrar
+  }
+
+  /**
+   * Notify about peer disconnected event.
+   * @param {PeerInfo} peerInfo
+   * @returns {void}
+   */
+  disconnect (peerInfo) {
+    this._onDisconnect(peerInfo)
+  }
+}
+
+module.exports = Topology

src/topology/multicodec-topology.js

@@ -0,0 +1,93 @@
+'use strict'
+
+const assert = require('assert')
+const Topology = require('./index')
+
+class MulticodecTopology extends Topology {
+  /**
+   * @param {Object} props
+   * @param {number} props.min minimum needed connections (default: 0)
+   * @param {number} props.max maximum needed connections (default: Infinity)
+   * @param {Array<string>} props.multicodecs protocol multicodecs
+   * @param {Object} props.handlers
+   * @param {function} props.handlers.onConnect protocol "onConnect" handler
+   * @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
+   * @constructor
+   */
+  constructor ({
+    min,
+    max,
+    multicodecs,
+    handlers
+  }) {
+    super({ min, max, handlers })
+
+    assert(multicodecs, 'one or more multicodec should be provided')
+    assert(handlers, 'the handlers should be provided')
+    assert(handlers.onConnect && typeof handlers.onConnect === 'function',
+      'the \'onConnect\' handler must be provided')
+    assert(handlers.onDisconnect && typeof handlers.onDisconnect === 'function',
+      'the \'onDisconnect\' handler must be provided')
+
+    this.multicodecs = Array.isArray(multicodecs) ? multicodecs : [multicodecs]
+    this._registrar = undefined
+
+    this._onProtocolChange = this._onProtocolChange.bind(this)
+  }
+
+  set registrar (registrar) {
+    this._registrar = registrar
+    this._registrar.peerStore.on('change:protocols', this._onProtocolChange)
+
+    // Update topology peers
+    this._updatePeers(this._registrar.peerStore.peers.values())
+  }
+
+  /**
+   * Update topology.
+   * @param {Array<PeerInfo>} peerInfoIterable
+   * @returns {void}
+   */
+  _updatePeers (peerInfoIterable) {
+    for (const peerInfo of peerInfoIterable) {
+      if (this.multicodecs.filter(multicodec => peerInfo.protocols.has(multicodec))) {
+        // Add the peer regardless of whether or not there is currently a connection
+        this.peers.set(peerInfo.id.toB58String(), peerInfo)
+        // If there is a connection, call _onConnect
+        const connection = this._registrar.getConnection(peerInfo)
+        connection && this._onConnect(peerInfo, connection)
+      } else {
+        // Remove any peers we might be tracking that are no longer of value to us
+        this.peers.delete(peerInfo.id.toB58String())
+      }
+    }
+  }
+
+  /**
+   * Check if a new peer support the multicodecs for this topology.
+   * @param {Object} props
+   * @param {PeerInfo} props.peerInfo
+   * @param {Array<string>} props.protocols
+   */
+  _onProtocolChange ({ peerInfo, protocols }) {
+    const existingPeer = this.peers.get(peerInfo.id.toB58String())
+    const hasProtocol = protocols.filter(protocol => this.multicodecs.includes(protocol))
+
+    // Not supporting the protocol anymore?
+    if (existingPeer && hasProtocol.length === 0) {
+      this._onDisconnect({
+        peerInfo
+      })
+    }
+
+    // New to protocol support
+    for (const protocol of protocols) {
+      if (this.multicodecs.includes(protocol)) {
+        this._updatePeers([peerInfo])
+        return
+      }
+    }
+  }
+}
+
+module.exports = MulticodecTopology

src/topology/tests/multicodec-topology.js

@@ -0,0 +1,89 @@
+/* eslint-env mocha */
+
+'use strict'
+
+const chai = require('chai')
+const expect = chai.expect
+chai.use(require('dirty-chai'))
+const sinon = require('sinon')
+
+const PeerId = require('peer-id')
+const PeerInfo = require('peer-info')
+const peers = require('../../utils/peers')
+
+module.exports = (test) => {
+  describe('multicodec topology', () => {
+    let topology, peer
+
+    beforeEach(async () => {
+      topology = await test.setup()
+      if (!topology) throw new Error('missing multicodec topology')
+
+      const id = await PeerId.createFromJSON(peers[0])
+      peer = await PeerInfo.create(id)
+    })
+
+    afterEach(async () => {
+      sinon.restore()
+      await test.teardown()
+    })
+
+    it('should have properties set', () => {
+      expect(topology.multicodecs).to.exist()
+      expect(topology._onConnect).to.exist()
+      expect(topology._onDisconnect).to.exist()
+      expect(topology.peers).to.exist()
+      expect(topology._registrar).to.exist()
+    })
+
+    it('should trigger "onDisconnect" on peer disconnected', () => {
+      sinon.spy(topology, '_onDisconnect')
+      topology.disconnect(peer)
+
+      expect(topology._onDisconnect.callCount).to.equal(1)
+    })
+
+    it('should update peers on protocol change', async () => {
+      sinon.spy(topology, '_updatePeers')
+      expect(topology.peers.size).to.eql(0)
+
+      const id2 = await PeerId.createFromJSON(peers[1])
+      const peer2 = await PeerInfo.create(id2)
+
+      const peerStore = topology._registrar.peerStore
+      peerStore.emit('change:protocols', {
+        peerInfo: peer2,
+        protocols: Array.from(topology.multicodecs)
+      })
+
+      expect(topology._updatePeers.callCount).to.equal(1)
+      expect(topology.peers.size).to.eql(1)
+    })
+
+    it('should disconnect if peer no longer supports a protocol', async () => {
+      sinon.spy(topology, '_onDisconnect')
+      expect(topology.peers.size).to.eql(0)
+
+      const id2 = await PeerId.createFromJSON(peers[1])
+      const peer2 = await PeerInfo.create(id2)
+      const peerStore = topology._registrar.peerStore
+
+      // Peer with the protocol
+      peerStore.emit('change:protocols', {
+        peerInfo: peer2,
+        protocols: Array.from(topology.multicodecs)
+      })
+
+      expect(topology.peers.size).to.eql(1)
+
+      // Peer does not support the protocol anymore
+      peerStore.emit('change:protocols', {
+        peerInfo: peer2,
+        protocols: []
+      })
+
+      expect(topology.peers.size).to.eql(1)
+      expect(topology._onDisconnect.callCount).to.equal(1)
+    })
+  })
+}