feat: retrieve peers on connect (#64)

* feat: retrieve peers on connect

* fix: clean up connections in dial and disconnect tests

* fix: remove rendezvous server from discover test

* chore: merge from master

src/listener.js

@@ -103,7 +103,7 @@ class Listener extends EE {
 
     const maStr = this.ma.toString()
 
-    this.io.emit('ss-join', maStr, pubKeyStr, (err, sig) => {
+    this.io.emit('ss-join', maStr, pubKeyStr, (err, sig, peers) => {
       if (err) { return callback(err) }
 
       if (sig) {
@@ -127,7 +127,7 @@ class Listener extends EE {
           return callback(new Error('Tried to listen on a server with crypto challenge disabled!\n    This is prohibited by default and can lead to security issues!\n    Please set "allowJoinWithDisabledChallenge" to true in the constructor options (but only if you know what you are doing)!'))
         }
         this.signature = '_'
-        callback()
+        callback(null, null, peers)
       }
     })
   }
@@ -215,8 +215,8 @@ class Listener extends EE {
 
     series([
       (cb) => this._up(cb),
-      (cb) => this._crypto(cb)
-    ], (err) => {
+      (cb) => this._crypto((err, ignore, peers) => cb(err, peers))
+    ], (err, [ignore, peers]) => {
       if (err) {
         // Error connecting to WebSocket
         if (err.description && err.description.code === 'ENOTFOUND') {
@@ -241,15 +241,23 @@ class Listener extends EE {
       this.io.on('reconnect', () => {
         // force to get a new signature
         this.signature = null
-        this._crypto((err) => {
+        this._crypto((err, ignore, reconnectPeers) => {
           if (err) {
             this.log('reconnect error', err)
             this.emit('error', err)
-          } else this.log('reconnected')
+          } else {
+            this.log('reconnected')
+            for (const p of (reconnectPeers || [])) {
+              this.emit('peer', p)
+            }
+          }
         })
       })
 
       this.emit('listening')
+      for (const p of (peers || [])) {
+        this.emit('peer', p)
+      }
       callback()
     })
   }

test/dial.spec.js

@@ -100,16 +100,26 @@ describe('dial', () => {
     ws1.dial(ma2, (err, conn) => {
       expect(err).to.not.exist()
 
+      let endFn
+      let ended = false
       pull(
-        (end, cb) => {},
+        // Prevent end until test has completed
+        (end, cb) => {
+          endFn = cb
+        },
         conn,
         pull.drain(() => {
-          expect('Stream should never end').to.not.exist()
+          // Should not be called until test has completed
+          ended = true
         })
       )
 
       listeners[0].close(() => {})
-      listeners[0].listen(ma1, done)
+      listeners[0].listen(ma1, () => {
+        expect(ended).to.be.equal(false)
+        endFn(true)
+        done()
+      })
     })
   })
 

test/disconnect.spec.js

@@ -4,6 +4,7 @@
 
 const multiaddr = require('multiaddr')
 const series = require('async/series')
+const each = require('async/each')
 const pull = require('pull-stream')
 
 const WebSocketStar = require('../src')
@@ -17,6 +18,7 @@ describe('disconnect', () => {
 
   let conn
   let otherConn
+  const listeners = []
 
   before((done) => {
     series([first, second], dial)
@@ -26,29 +28,38 @@ describe('disconnect', () => {
 
       const listener = ws1.createListener((conn) => pull(conn, conn))
       listener.listen(ma1, next)
+      listeners.push(listener)
     }
 
     function second (next) {
       ws2 = new WebSocketStar({ allowJoinWithDisabledChallenge: true })
 
       const listener = ws2.createListener((conn) => (otherConn = conn))
       listener.listen(ma2, next)
+      listeners.push(listener)
     }
 
     function dial () {
       conn = ws1.dial(ma2, done)
     }
   })
 
+  after(done => each(listeners, (l, next) => l.close(next), done))
+
   it('all conns die when one peer quits', (done) => {
+    let endFn
     pull(
+      (end, cb) => {
+        endFn = cb
+      },
       conn,
       pull.collect(err => {
         if (err) return done(err)
         pull(
           otherConn,
           pull.collect(err => {
             if (err) return done(err)
+            endFn(true)
             done()
           })
         )

test/discovery.spec.js

@@ -14,10 +14,13 @@ const WebSocketStar = require('../src')
 describe('peer discovery', () => {
   let listeners = []
   let ws1
-  const ma1 = multiaddr('/ip4/127.0.0.1/tcp/15001/ws/p2p-websocket-star/ipfs/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSooo3A')
-
+  const ma1 = multiaddr('/ip4/127.0.0.1/tcp/15001/ws/p2p-websocket-star/ipfs/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSooo4A')
   let ws2
-  const ma2 = multiaddr('/ip4/127.0.0.1/tcp/15003/ws/p2p-websocket-star/ipfs/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSooo3B')
+  const ma2 = multiaddr('/ip4/127.0.0.1/tcp/15002/ws/p2p-websocket-star/ipfs/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSooo4B')
+  let ws3
+  const ma3 = multiaddr('/ip4/127.0.0.1/tcp/15002/ws/p2p-websocket-star/ipfs/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSooo4C')
+
+  after(done => each(listeners, (l, next) => l.close(next), done))
 
   it('listen on the first', (done) => {
     ws1 = new WebSocketStar({ allowJoinWithDisabledChallenge: true })
@@ -45,7 +48,34 @@ describe('peer discovery', () => {
     listener.listen(ma2, (err) => {
       expect(err).to.not.exist()
     })
-  }).timeout(5000)
+  })
+
+  it('new peer receives peer events for all other peers on connect', (done) => {
+    ws3 = new WebSocketStar({ allowJoinWithDisabledChallenge: true })
+
+    const discovered = []
+    ws3.discovery.on('peer', (peerInfo) => {
+      discovered.push(peerInfo.multiaddrs)
+      if (discovered.length === 2) {
+        gotAllPeerEvents()
+      }
+    })
 
-  after(done => each(listeners, (l, next) => l.close(next), done))
+    const gotAllPeerEvents = () => {
+      const allMas = new Set()
+      discovered.forEach(mas => {
+        mas.forEach(ma => allMas.add(ma.toString()))
+      })
+      expect(allMas.has(ma1.toString())).to.equal(true)
+      expect(allMas.has(ma2.toString())).to.equal(true)
+      done()
+    }
+
+    const listener = ws3.createListener((/* conn */) => {})
+
+    listeners.push(listener)
+    listener.listen(ma3, (err) => {
+      expect(err).to.not.exist()
+    })
+  })
 })