chore: separate server and client rendezvous

package.json

@@ -44,26 +44,26 @@
     "it-buffer": "^0.1.2",
     "it-length-prefixed": "^3.1.0",
     "it-pipe": "^1.1.0",
+    "libp2p": "^0.29.0",
     "libp2p-interfaces": "^0.5.1",
+    "libp2p-mplex": "^0.10.0",
+    "libp2p-noise": "^2.0.1",
+    "libp2p-tcp": "^0.15.1",
+    "libp2p-websockets": "^0.14.0",
+    "menoetius": "0.0.2",
+    "minimist": "^1.2.5",
     "multiaddr": "^8.0.0",
     "peer-id": "^0.14.1",
     "protons": "^2.0.0",
     "streaming-iterables": "^5.0.2",
     "uint8arrays": "^1.1.0"
   },
-  "peerDependencies": {
-    "libp2p": "^0.29.0"
-  },
   "devDependencies": {
     "aegir": "^26.0.0",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
     "delay": "^4.4.0",
     "dirty-chai": "^2.0.1",
-    "libp2p": "^0.29.0",
-    "libp2p-mplex": "^0.10.0",
-    "libp2p-noise": "^2.0.1",
-    "libp2p-websockets": "^0.14.0",
     "p-defer": "^3.0.0",
     "p-times": "^3.0.0",
     "p-wait-for": "^3.1.0",

src/index.js

@@ -14,17 +14,11 @@ const toString = require('uint8arrays/to-string')
 
 const MulticodecTopology = require('libp2p-interfaces/src/topology/multicodec-topology')
 
-const Server = require('./server')
 const { codes: errCodes } = require('./errors')
 const { PROTOCOL_MULTICODEC } = require('./constants')
 const { Message } = require('./proto')
 const MESSAGE_TYPE = Message.MessageType
 
-const defaultServerOptions = {
-  enabled: true,
-  gcInterval: 3e5
-}
-
 /**
 * Rendezvous point contains the connection to a rendezvous server, as well as,
 * the cookies per namespace that the client received.
@@ -42,33 +36,17 @@ class Rendezvous {
    * @constructor
    * @param {object} params
    * @param {Libp2p} params.libp2p
-   * @param {object} [params.server]
-   * @param {boolean} [params.server.enabled = true]
-   * @param {number} [params.server.gcInterval = 3e5]
    */
-  constructor ({ libp2p, server = {} }) {
+  constructor ({ libp2p }) {
     this._libp2p = libp2p
     this._peerId = libp2p.peerId
     this._registrar = libp2p.registrar
 
-    this._serverOptions = {
-      ...defaultServerOptions,
-      ...server
-    }
-
     /**
      * @type {Map<string, RendezvousPoint>}
      */
     this._rendezvousPoints = new Map()
 
-    /**
-    * Client cookies per namespace for own server
-    * @type {Map<string, string>}
-    */
-    this._cookiesSelf = new Map()
-
-    this._server = undefined
-
     this._registrarId = undefined
     this._onPeerConnected = this._onPeerConnected.bind(this)
     this._onPeerDisconnected = this._onPeerDisconnected.bind(this)
@@ -85,12 +63,6 @@ class Rendezvous {
 
     log('starting')
 
-    // Create and start Rendezvous server if enabled
-    if (this._serverOptions.enabled) {
-      this._server = new Server(this._libp2p, this._serverOptions)
-      this._server.start()
-    }
-
     // register protocol with topology
     const topology = new MulticodecTopology({
       multicodecs: PROTOCOL_MULTICODEC,
@@ -105,7 +77,7 @@ class Rendezvous {
   }
 
   /**
-   * Unregister the rendezvous protocol and the streams with other peers will be closed.
+   * Unregister the rendezvous protocol and clear the state.
    * @returns {void}
    */
   stop () {
@@ -115,17 +87,11 @@ class Rendezvous {
 
     log('stopping')
 
-    clearInterval(this._interval)
-
     // unregister protocol and handlers
     this._registrar.unregister(this._registrarId)
-    if (this._serverOptions.enabled) {
-      this._server.stop()
-    }
 
     this._registrarId = undefined
     this._rendezvousPoints.clear()
-    this._cookiesSelf.clear()
 
     log('stopped')
   }
@@ -153,10 +119,6 @@ class Rendezvous {
     log('disconnected', idB58Str)
 
     this._rendezvousPoints.delete(idB58Str)
-
-    if (this._server) {
-      this._server.removePeerRegistrations(peerId)
-    }
   }
 
   /**
@@ -271,7 +233,7 @@ class Rendezvous {
    * Discover peers registered under a given namespace
    * @param {string} ns
    * @param {number} [limit]
-   * @returns {AsyncIterable<{ signedPeerRecord: Buffer, ns: string, ttl: number }>}
+   * @returns {AsyncIterable<{ signedPeerRecord: Uint8Array, ns: string, ttl: number }>}
    */
   async * discover (ns, limit) {
     // Are there available rendezvous servers?
@@ -285,24 +247,6 @@ class Rendezvous {
       ttl: r.ttl * 1e3 // convert to ms
     })
 
-    // Local search if Server enabled
-    if (this._server) {
-      const cookieSelf = this._cookiesSelf.get(ns)
-      const { cookie: cookieS, registrations: localRegistrations } = this._server.getRegistrations(ns, { limit, cookie: cookieSelf })
-
-      for (const r of localRegistrations) {
-        yield registrationTransformer(r)
-
-        limit--
-        if (limit === 0) {
-          return
-        }
-      }
-
-      // Store cookie self
-      this._cookiesSelf.set(ns, cookieS)
-    }
-
     // Iterate over all rendezvous points
     for (const [id, rp] of this._rendezvousPoints.entries()) {
       const rpCookies = rp.cookies || new Map()

src/server/bin.js

@@ -0,0 +1,93 @@
+#!/usr/bin/env node
+
+'use strict'
+
+// Usage: $0 [--peerId <jsonFilePath>] [--listenMultiaddrs <ma> ... <ma>] [--announceMultiaddrs <ma> ... <ma>] [--metricsMultiaddr <ma>] [--disableMetrics]
+
+/* eslint-disable no-console */
+
+const debug = require('debug')
+const log = debug('libp2p:rendezvous:bin')
+
+const fs = require('fs')
+const http = require('http')
+const menoetius = require('menoetius')
+const argv = require('minimist')(process.argv.slice(2))
+
+const TCP = require('libp2p-tcp')
+const Websockets = require('libp2p-websockets')
+const Muxer = require('libp2p-mplex')
+const { NOISE: Crypto } = require('libp2p-noise')
+
+const multiaddr = require('multiaddr')
+const PeerId = require('peer-id')
+
+const RendezvousServer = require('./index')
+const { getAnnounceAddresses, getListenAddresses } = require('./utils')
+
+async function main () {
+  // Metrics
+  let metricsServer
+  const metrics = !(argv.disableMetrics || process.env.DISABLE_METRICS)
+  const metricsMa = multiaddr(argv.metricsMultiaddr || argv.ma || process.env.METRICSMA || '/ip4/127.0.0.1/tcp/8003')
+  const metricsAddr = metricsMa.nodeAddress()
+
+  // Multiaddrs
+  const listenAddresses = getListenAddresses(argv)
+  const announceAddresses = getAnnounceAddresses(argv)
+
+  // PeerId
+  let peerId
+  if (argv.peerId) {
+    const peerData = fs.readFileSync(argv.peerId)
+    peerId = await PeerId.createFromJSON(JSON.parse(peerData))
+  } else {
+    peerId = await PeerId.create()
+    log('You are using an automatically generated peer.')
+    log('If you want to keep the same address for the server you should provide a peerId with --peerId <jsonFilePath>')
+  }
+
+  // Create Rendezvous server
+  const rendezvousServer = new RendezvousServer({
+    modules: {
+      transport: [Websockets, TCP],
+      streamMuxer: [Muxer],
+      connEncryption: [Crypto]
+    },
+    peerId,
+    addresses: {
+      listen: listenAddresses,
+      announce: announceAddresses
+    }
+  })
+
+  await rendezvousServer.start()
+
+  if (metrics) {
+    log('enabling metrics')
+    metricsServer = http.createServer((req, res) => {
+      if (req.url !== '/metrics') {
+        res.statusCode = 200
+        res.end()
+      }
+    })
+
+    menoetius.instrument(metricsServer)
+
+    metricsServer.listen(metricsAddr.port, metricsAddr.address, () => {
+      console.log(`metrics server listening on ${metricsAddr.port}`)
+    })
+  }
+
+  const stop = async () => {
+    console.log('Stopping...', rendezvousServer.multiaddrs)
+    await rendezvousServer.stop()
+    metricsServer && await metricsServer.close()
+    process.exit(0)
+  }
+
+  process.on('SIGTERM', stop)
+  process.on('SIGINT', stop)
+}
+
+main()

src/server/index.js

@@ -4,6 +4,7 @@ const debug = require('debug')
 const log = debug('libp2p:rendezvous-server')
 log.error = debug('libp2p:rendezvous-server:error')
 
+const Libp2p = require('libp2p')
 const PeerId = require('peer-id')
 
 const { PROTOCOL_MULTICODEC, MAX_LIMIT } = require('../constants')
@@ -27,16 +28,16 @@ const rpc = require('./rpc')
 /**
  * Libp2p rendezvous server.
  */
-class RendezvousServer {
+class RendezvousServer extends Libp2p {
   /**
      * @constructor
-     * @param {Libp2p} libp2p
+     * @param {Libp2pOptions} libp2pOptions
      * @param {object} [options]
      * @param {number} [options.gcInterval = 3e5]
      */
-  constructor (libp2p, { gcInterval = 3e5 } = {}) {
-    this._registrar = libp2p.registrar
-    this._peerStore = libp2p.peerStore
+  constructor (libp2pOptions, { gcInterval = 3e5 } = {}) {
+    super(libp2pOptions)
+
     this._gcInterval = gcInterval
 
     /**
@@ -59,6 +60,8 @@ class RendezvousServer {
    * @returns {void}
    */
   start () {
+    super.start()
+
     if (this._interval) {
       return
     }
@@ -69,7 +72,7 @@ class RendezvousServer {
     this._interval = setInterval(this._gc, this._gcInterval)
 
     // Incoming streams handling
-    this._registrar.handle(PROTOCOL_MULTICODEC, rpc(this))
+    this.registrar.handle(PROTOCOL_MULTICODEC, rpc(this))
 
     log('started')
   }
@@ -79,6 +82,8 @@ class RendezvousServer {
    * @returns {void}
    */
   stop () {
+    super.stop()
+
     clearInterval(this._interval)
     this._interval = undefined
 
@@ -143,7 +148,7 @@ class RendezvousServer {
     this.nsRegistrations.set(ns, nsReg)
 
     // Store envelope in the AddressBook
-    this._peerStore.addressBook.consumePeerRecord(envelope)
+    this.peerStore.addressBook.consumePeerRecord(envelope)
   }
 
   /**
@@ -213,7 +218,7 @@ class RendezvousServer {
       cRegistrations.add(nsReg.id)
       registrations.push({
         ns,
-        signedPeerRecord: this._peerStore.addressBook.getRawEnvelope(PeerId.createFromB58String(idStr)),
+        signedPeerRecord: this.peerStore.addressBook.getRawEnvelope(PeerId.createFromB58String(idStr)),
         ttl: Date.now() - nsReg.expiration
       })
 

src/server/utils.js

@@ -0,0 +1,41 @@
+'use strict'
+
+const multiaddr = require('multiaddr')
+
+function getAnnounceAddresses(argv) {
+  const announceAddr = argv.announceMultiaddrs || argv.am
+  const announceAddresses = announceAddr ? [multiaddr(announceAddr)] : []
+
+  if (argv.announceMultiaddrs || argv.am) {
+    const flagIndex = process.argv.findIndex((e) => e === '--announceMultiaddrs' || e === '--am')
+    const tmpEndIndex = process.argv.slice(flagIndex + 1).findIndex((e) => e.startsWith('--'))
+    const endIndex = tmpEndIndex !== -1 ? tmpEndIndex : process.argv.length - flagIndex - 1
+
+    for (let i = flagIndex + 1; i < flagIndex + endIndex; i++) {
+      announceAddresses.push(multiaddr(process.argv[i + 1]))
+    }
+  }
+
+  return announceAddresses
+}
+
+module.exports.getAnnounceAddresses = getAnnounceAddresses
+
+function getListenAddresses(argv) {
+  const listenAddr = argv.listenMultiaddrs || argv.lm || '/ip4/127.0.0.1/tcp/15002/ws'
+  const listenAddresses = [multiaddr(listenAddr)]
+
+  if (argv.listenMultiaddrs || argv.lm) {
+    const flagIndex = process.argv.findIndex((e) => e === '--listenMultiaddrs' || e === '--lm')
+    const tmpEndIndex = process.argv.slice(flagIndex + 1).findIndex((e) => e.startsWith('--'))
+    const endIndex = tmpEndIndex !== -1 ? tmpEndIndex : process.argv.length - flagIndex - 1
+
+    for (let i = flagIndex + 1; i < flagIndex + endIndex; i++) {
+      listenAddresses.push(multiaddr(process.argv[i + 1]))
+    }
+  }
+
+  return listenAddresses
+}
+
+module.exports.getListenAddresses = getListenAddresses