Improve and Cleanup

.aegir.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const Utils = require('./test/utils.peer')
+
+let Server
+
+async function pre (done) {
+  Server = await Utils.createServer(require('./test/server.id.json'))
+  done()
+}
+
+function post (done) {
+  Server.stop()
+  Server.swarm.stop(done)
+}
+
+module.exports = {
+  hooks: {
+    pre,
+    post
+  }
+}

package.json

@@ -3,9 +3,15 @@
   "version": "0.0.0",
   "description": "A javascript implementation of the rendezvous protocol for libp2p",
   "leadMaintainer": "Vasco Santos <[email protected]>",
-  "main": "index.js",
+  "main": "src/index.js",
   "scripts": {
-    "test": "aegir test"
+    "test": "aegir test",
+    "test:node": "aegir test -t node",
+    "test:browser": "aegir test -t browser",
+    "test:webworker": "aegir test -t webworker"
+  },
+  "browser": {
+    "test/utils.js": "./test/utils.browser.js"
   },
   "keywords": [
     "libp2p",
@@ -18,16 +24,20 @@
   "dependencies": {
     "chai": "^4.1.2",
     "dirty-chai": "^2.0.1",
+    "immutable": "4.0.0-rc.9",
+    "promisify-es6": "^1.0.3",
     "protons": "^1.0.1",
-    "pull-protocol-buffers": "^0.1.2"
+    "pull-protocol-buffers": "^0.1.2",
+    "pull-through": "^1.0.18"
   },
   "devDependencies": {
-    "aegir": "^13.1.0",
-    "libp2p": "^0.20.2",
+    "aegir": "^14.0.0",
+    "libp2p": "^0.20.4",
     "libp2p-mplex": "^0.7.0",
     "libp2p-secio": "^0.10.0",
     "libp2p-spdy": "^0.12.1",
-    "libp2p-tcp": "^0.12.0"
+    "libp2p-tcp": "^0.12.0",
+    "libp2p-websockets": "^0.12.0"
   },
   "repository": {
     "type": "git",

src/client.js

@@ -0,0 +1,204 @@
+'use strict'
+
+const Sync = require('./sync')
+const RPC = require('./rpc')
+const pull = require('pull-stream')
+
+const debug = require('debug')
+const log = debug('libp2p:rendezvous:client')
+const {parallel, map} = require('async')
+
+class Client {
+  constructor (swarm) {
+    this.swarm = swarm
+    this.store = Sync.create()
+    this._dialLock = {}
+    this._failedCache = {}
+  }
+
+  dial (peer) {
+    const id = peer.id.toB58String()
+    this.sync()
+
+    // check if we need to dial
+    if (this._failedCache[id]) return log('not dialing %s because dial previously failed', id)
+    if (this._dialLock[id]) return log('not dialing %s because dial is already in progress', id)
+    if (Sync.getPoint(this.store, id)) return log('not dialing %s because peer is already connected', id)
+
+    this._dialLock[id] = true // prevent race
+    log('dialing %s', id)
+
+    const cb = (err) => {
+      delete this._dialLock[id]
+
+      if (err) {
+        log('dialing %s failed: %s', err)
+        this._failedCache[id] = true
+        return
+      }
+
+      log('dialing %s succeeded', id)
+      this.sync()
+    }
+
+    // do the actual dialing
+    this.swarm.dialProtocol(peer, '/p2p/rendezvous/1.0.0', (err, conn) => {
+      if (err) return cb(err)
+
+      conn.getPeerInfo((err, pi) => {
+        if (err) return cb(err)
+
+        let rpc = RPC(pi, this)
+
+        pull(
+          conn,
+          rpc,
+          conn
+        )
+
+        this.store = Sync.addPoint(this.store, id, rpc.rpc)
+        cb()
+      })
+    })
+  }
+
+  sync () {
+    if (this._syncLock) {
+      this._needResync = true
+      return
+    }
+    this._syncLock = true
+    log('syncing')
+    this.store = Sync.clearPoints(this.store)
+    let actions = [] // async rpc calls
+
+    // adds register / unregsiter actions to "actions" array
+    /*
+    pseudo-code:
+
+    for all store.points as point:
+      for all point.registrations as pReg:
+        if store.registrations does not contain pReg:
+          actions push "unregister pReg @ point"
+          delete point.registrations[pReg]
+      for all store.registrations as reg:
+        if point.registrations does not contain reg:
+          actions push "register reg @ point"
+          set point.registrations[reg]
+    */
+
+    let points = this.store.get('points')
+    let registrations = this.store.get('registrations')
+
+    this.store = this.store.set('points', points.reduce((points, point, id) => {
+      let regs = point.get('registrations')
+
+      regs = regs.reduce((regs, pReg, pRegId) => {
+        if (!registrations.get(pRegId)) {
+          log('sync: unregister@%s: %s', id, pRegId)
+          actions.push(cb => point.toJS().rpc().unregister(pRegId, pReg.peer.id.toBytes(), cb))
+          return regs.delete(pRegId)
+        }
+
+        return regs
+      }, regs)
+
+      regs = registrations.reduce((regs, reg, regId) => {
+        if (!regs.get(regId)) {
+          log('sync: register@%s: %s', id, regId)
+          actions.push(cb => point.toJS().rpc().register(regId, reg.peer, reg.ttl, cb))
+          return regs.set(regId, reg)
+        }
+
+        return regs
+      }, regs)
+
+      return points.set(id, point.set('registrations', regs))
+    }, points))
+
+    log('do sync')
+    parallel(actions, (err) => {
+      log('done sync')
+      delete this._syncLock
+
+      if (err) {
+        log(err) // ???
+      }
+
+      if (this._needResync) {
+        delete this._needResync
+        this.sync()
+      }
+    })
+  }
+
+  register (ns, peer, ttl, cb) {
+    if (typeof ttl === 'function') {
+      cb = ttl
+      ttl = 0
+    }
+    if (typeof peer === 'function') {
+      ttl = 0
+      cb = peer
+      peer = this.swarm.peerInfo
+    }
+
+    this.store = this.store.set('registrations', this.store.get('registrations').set(ns, {peer, ttl}))
+    this.sync()
+  }
+
+  _discover (peerID, ns, limit, cb) {
+    if (typeof limit === 'function') {
+      cb = limit
+      limit = 0
+    }
+    if (typeof ns === 'function') {
+      limit = 0
+      cb = ns
+      ns = null
+    }
+
+    log('discover@%s: %s limit=%s', peerID, ns || '<GLOBAL>', limit)
+
+    let point = this.store.get('points').get(peerID)
+    if (!point || !point.get('rpc')().online()) {
+      return cb(new Error('Point not connected!'))
+    }
+
+    point.get('rpc')().discover(ns, limit, point.get('cookies').get(ns) || Buffer.from(''), (err, res) => {
+      if (err) return cb(err)
+      this.store.set('points',
+        this.store.get('points').set(peerID,
+          this.store.get('points').get(peerID).set('cookies',
+            this.store.get('points').get(peerID).get('cookies').set(ns, res.cookie))))
+      return cb(null, res.peers)
+    })
+  }
+
+  discover (ns, cb) {
+    if (typeof ns === 'function') {
+      cb = ns
+    }
+
+    let ids = this.store.get('points').toArray().map(p => p[0])
+
+    map(ids,
+      (peerID, cb) => this._discover(peerID, ns, 0, (err, res) => err ? cb(null, []) : cb(null, res)),
+      (err, res) => err ? cb(err) : cb(null, res.reduce((a, b) => a.concat(b), [])))
+  }
+
+  unregister (ns, id) {
+    if (!ns) {
+      id = this.swarm.peerInfo.id.toBytes()
+      ns = null
+    }
+    if (!id) {
+      id = this.swarm.peerInfo.id.toBytes()
+    }
+
+    this.store = this.store.set('registrations', this.store.get('registrations').delete(ns))
+    this.sync()
+  }
+}
+
+module.exports = Client

src/index.js

@@ -1,77 +1,64 @@
 'use strict'
 
-const RPC = require('./rpc')
+const debug = require('debug')
+const log = debug('libp2p:rendezvous')
 const noop = () => {}
 
-class RendezvousDiscovery {
-  constructor (swarm) {
-    this.swarm = swarm
-    this.peers = []
-  }
-
-  _dial (pi, cb) {
-    if (!cb) cb = noop
-    this.swarm.dialProtocol(pi, '/rendezvous/1.0.0', (err, conn) => {
-      if (err) return cb(err)
-      const rpc = new RPC()
-      rpc.setup(conn, err => {
-        if (err) return cb(err)
-        this.peers.push(rpc)
-        cb()
-      })
-    })
-  }
-
-  _rpc (cmd, ...a) { // TODO: add. round-robin / multicast / anycast?
-    this.peers[0][cmd](...a)
-  }
+const Client = require('./client')
+const EE = require('events').EventEmitter
 
-  register (ns, peer, cb) {
-    this._rpc('register', ns, peer, 0, cb) // TODO: interface does not expose ttl option?!
+class RendezvousDiscovery extends EE {
+  constructor (swarm, opt) {
+    super()
+    this._client = new Client(swarm, opt)
+    this._discover = {}
+    this.swarm = swarm
+    this.tag = 'rendezvous'
   }
-
-  discover (ns, limit, cookie, cb) {
-    if (typeof cookie === 'function') {
-      cb = cookie
-      cookie = Buffer.from('')
-    }
-    if (typeof limit === 'function') {
-      cookie = Buffer.from('')
-      cb = limit
-      limit = 0
+  start (cb) {
+    log('start')
+    this._loop = setInterval(this._discoverLoop.bind(this), 10 * 1000)
+    this.swarm.on('peer:connect', (peer) => this._client.dial(peer))
+    this.swarm.on('peer:disconnect', () => this._client.sync())
+    if (cb) {
+      cb()
     }
-    if (typeof ns === 'function') {
-      cookie = Buffer.from('')
-      limit = 0
-      cb = ns
-      ns = null
+  }
+  stop (cb) {
+    log('stop')
+    clearInterval(this._loop)
+    this._client.stop()
+    if (cb) {
+      cb()
     }
-
-    this._rpc('discover', ns, limit, cookie, cb)
   }
-
-  unregister (ns, id) {
+  // TODO: https://github.com/libp2p/specs/issues/47
+  register (ns) {
     if (!ns) {
-      id = this.swarm.peerInfo.id.toBytes()
-      ns = null
-    }
-    if (!id) {
-      id = this.swarm.peerInfo.id.toBytes()
+      ns = null // need cannonical form of "empty"
     }
-
-    this._rpc('unregister', ns, id)
+    log('register', ns)
+    this._discover[ns] = true
+    this._client.register(ns, noop)
   }
-
-  start (cb) {
-    this.swarm.on('peer:connect', peer => {
-      this._dial(peer)
-    })
-    cb()
+  unregister (ns) {
+    if (!ns) {
+      ns = null // need cannonical form of "empty"
+    }
+    log('unregister', ns)
+    delete this._discover[ns]
+    this._client.unregister(ns, noop)
   }
-
-  stop (cb) {
-    // TODO: shutdown all conns
-    cb()
+  _discoverLoop() {
+    log('discover loop')
+    for (const ns in this._discover) {
+      this._client.discover(ns, (err, peers) => {
+        peers.forEach(peer => {
+          this.emit('peer', peer)
+          this.emit(ns ? 'ns:' + ns : 'global', peer)
+        })
+      })
+    }
   }
 }