advanced peer-book (#206)

* fix: improve availableTransports filtering (was actually broken before)
* feat: new peer-info
* feat: now logs the used multiaddr

Author: daviddias

README.md

@@ -58,7 +58,7 @@ And use it in your Node.js code as:
 ```JavaScript
 const Swarm = require('libp2p-swarm')
 
-const sw = new Swarm(peerInfo)
+const sw = new Swarm(peerInfo [, peerBook])
 ```
 
 ## API

gulpfile.js

@@ -5,10 +5,10 @@ const PeerInfo = require('peer-info')
 const PeerId = require('peer-id')
 const WebSockets = require('libp2p-websockets')
 const pull = require('pull-stream')
+const PeerBook = require('peer-book')
 
 const Swarm = require('./src')
 const spdy = require('libp2p-spdy')
-const multiaddr = require('multiaddr')
 const fs = require('fs')
 const path = require('path')
 
@@ -27,19 +27,19 @@ gulp.task('test:browser:before', (done) => {
   function createListenerA (cb) {
     PeerId.createFromJSON(
       JSON.parse(
-        fs.readFileSync(
-          path.join(__dirname, './test/test-data/id-1.json'))
+        fs.readFileSync(path.join(__dirname, './test/test-data/id-1.json'))
       ),
       (err, id) => {
         if (err) {
           return cb(err)
         }
 
         const peerA = new PeerInfo(id)
-        const maA = multiaddr('/ip4/127.0.0.1/tcp/9100/ws')
+        const maA = '/ip4/127.0.0.1/tcp/9100/ws'
+
+        peerA.multiaddrs.add(maA)
+        swarmA = new Swarm(peerA, new PeerBook())
 
-        peerA.multiaddr.add(maA)
-        swarmA = new Swarm(peerA)
         swarmA.transport.add('ws', new WebSockets())
         swarmA.transport.listen('ws', {}, echo, cb)
       })
@@ -48,19 +48,18 @@ gulp.task('test:browser:before', (done) => {
   function createListenerB (cb) {
     PeerId.createFromJSON(
       JSON.parse(
-        fs.readFileSync(
-          path.join(__dirname, './test/test-data/id-2.json'))
+        fs.readFileSync(path.join(__dirname, './test/test-data/id-2.json'))
       ),
       (err, id) => {
         if (err) {
           return cb(err)
         }
 
         const peerB = new PeerInfo(id)
-        const maB = multiaddr('/ip4/127.0.0.1/tcp/9200/ws')
+        const maB = '/ip4/127.0.0.1/tcp/9200/ws'
 
-        peerB.multiaddr.add(maB)
-        swarmB = new Swarm(peerB)
+        peerB.multiaddrs.add(maB)
+        swarmB = new Swarm(peerB, new PeerBook())
 
         swarmB.transport.add('ws', new WebSockets())
         swarmB.connection.addStreamMuxer(spdy)

package.json

@@ -49,10 +49,11 @@
     "libp2p-secio": "~0.6.8",
     "libp2p-spdy": "~0.10.6",
     "libp2p-tcp": "~0.10.0",
-    "libp2p-webrtc-star": "~0.8.10",
+    "libp2p-webrtc-star": "~0.9.0",
     "libp2p-websockets": "~0.10.0",
     "pre-commit": "^1.2.2",
     "pull-goodbye": "0.0.1",
+    "peer-book": "~0.4.0",
     "pull-stream": "^3.5.0",
     "webrtcsupport": "^2.2.0"
   },
@@ -62,13 +63,13 @@
     "debug": "^2.6.3",
     "interface-connection": "~0.3.2",
     "ip-address": "^5.8.6",
-    "libp2p-identify": "~0.3.3",
+    "libp2p-identify": "~0.4.0",
     "lodash.includes": "^4.3.0",
     "multiaddr": "^2.3.0",
     "multistream-select": "~0.13.5",
     "once": "^1.4.0",
-    "peer-id": "~0.8.5",
-    "peer-info": "~0.8.5",
+    "peer-id": "~0.8.6",
+    "peer-info": "~0.9.2",
     "protocol-buffers": "^3.2.1"
   },
   "contributors": [
@@ -86,4 +87,4 @@
     "greenkeeper[bot] <greenkeeper[bot]@users.noreply.github.com>",
     "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <[email protected]>"
   ]
-}
+}

src/connection.js

@@ -41,7 +41,7 @@ module.exports = function connection (swarm) {
               (conn, cb) => identify.dialer(conn, cb),
               (peerInfo, observedAddrs, cb) => {
                 observedAddrs.forEach((oa) => {
-                  swarm._peerInfo.multiaddr.addSafe(oa)
+                  swarm._peerInfo.multiaddrs.addSafe(oa)
                 })
                 cb(null, peerInfo)
               }
@@ -52,13 +52,28 @@ module.exports = function connection (swarm) {
             if (err) {
               return log('Identify not successful')
             }
-            swarm.muxedConns[peerInfo.id.toB58String()] = {
-              muxer: muxedConn
+            const b58Str = peerInfo.id.toB58String()
+
+            swarm.muxedConns[b58Str] = { muxer: muxedConn }
+
+            if (peerInfo.multiaddrs.size > 0) {
+              // with incomming conn and through identify, going to pick one
+              // of the available multiaddrs from the other peer as the one
+              // I'm connected to as we really can't be sure at the moment
+              // TODO add this consideration to the connection abstraction!
+              peerInfo.connect(peerInfo.multiaddrs.toArray()[0])
+            } else {
+              // for the case of websockets in the browser, where peers have
+              // no addr, use just their IPFS id
+              peerInfo.connect(`/ipfs/${b58Str}`)
             }
+            peerInfo = swarm._peerBook.put(peerInfo)
 
             swarm.emit('peer-mux-established', peerInfo)
             muxedConn.on('close', () => {
-              delete swarm.muxedConns[peerInfo.id.toB58String()]
+              delete swarm.muxedConns[b58Str]
+              peerInfo.disconnect()
+              peerInfo = swarm._peerBook.put(peerInfo)
               swarm.emit('peer-mux-closed', peerInfo)
             })
           })

src/dial.js

@@ -14,9 +14,7 @@ module.exports = function dial (swarm) {
       protocol = null
     }
 
-    if (!callback) {
-      callback = function noop () {}
-    }
+    callback = callback || function noop () {}
 
     const proxyConn = new Connection()
 
@@ -160,7 +158,10 @@ module.exports = function dial (swarm) {
           swarm.emit('peer-mux-established', pi)
 
           muxedConn.once('close', () => {
-            delete swarm.muxedConns[pi.id.toB58String()]
+            const b58Str = pi.id.toB58String()
+            delete swarm.muxedConns[b58Str]
+            pi.disconnect()
+            swarm._peerBook.get(b58Str).disconnect()
             swarm.emit('peer-mux-closed', pi)
           })
 

src/index.js

@@ -4,28 +4,27 @@ const util = require('util')
 const EE = require('events').EventEmitter
 const each = require('async/each')
 const series = require('async/series')
-const includes = require('lodash.includes')
-
 const transport = require('./transport')
 const connection = require('./connection')
 const dial = require('./dial')
 const protocolMuxer = require('./protocol-muxer')
 const plaintext = require('./plaintext')
+const assert = require('assert')
 
 exports = module.exports = Swarm
 
 util.inherits(Swarm, EE)
 
-function Swarm (peerInfo) {
+function Swarm (peerInfo, peerBook) {
   if (!(this instanceof Swarm)) {
     return new Swarm(peerInfo)
   }
 
-  if (!peerInfo) {
-    throw new Error('You must provide a value for `peerInfo`')
-  }
+  assert(peerInfo, 'You must provide a `peerInfo`')
+  assert(peerBook, 'You must provide a `peerBook`')
 
   this._peerInfo = peerInfo
+  this._peerBook = peerBook
 
   // transports --
   // { key: transport }; e.g { tcp: <tcp> }
@@ -59,25 +58,11 @@ function Swarm (peerInfo) {
   this.connection = connection(this)
 
   this.availableTransports = (pi) => {
-    const addrs = pi.multiaddrs
+    const myAddrs = pi.multiaddrs.toArray()
+    const myTransports = Object.keys(this.transports)
 
     // Only listen on transports we actually have addresses for
-    return Object.keys(this.transports).filter((ts) => {
-      // ipfs multiaddrs are not dialable so we drop them here
-      let dialable = addrs.map((addr) => {
-        // webrtc-star needs the /ipfs/QmHash
-        if (addr.toString().indexOf('webrtc-star') > 0) {
-          return addr
-        }
-
-        if (includes(addr.protoNames(), 'ipfs')) {
-          return addr.decapsulate('ipfs')
-        }
-        return addr
-      })
-
-      return this.transports[ts].filter(dialable).length > 0
-    })
+    return myTransports.filter((ts) => this.transports[ts].filter(myAddrs).length > 0)
   }
 
   // higher level (public) API

src/limit-dialer/index.js

@@ -36,7 +36,8 @@ class LimitDialer {
   dialMany (peer, transport, addrs, callback) {
     log('dialMany:start')
     // we use a token to track if we want to cancel following dials
-    const token = {cancel: false}
+    const token = { cancel: false }
+
     map(addrs, (m, cb) => {
       this.dialSingle(peer, transport, m, token, cb)
     }, (err, results) => {
@@ -47,7 +48,7 @@ class LimitDialer {
       const success = results.filter((res) => res.conn)
       if (success.length > 0) {
         log('dialMany:success')
-        return callback(null, success[0].conn)
+        return callback(null, success[0])
       }
 
       log('dialMany:error')

src/limit-dialer/queue.js

@@ -38,34 +38,30 @@ class DialQueue {
    */
   _doWork (transport, addr, token, callback) {
     log('work')
-    this._dialWithTimeout(
-      transport,
-      addr,
-      (err, conn) => {
-        if (err) {
-          log('work:error')
-          return callback(null, {error: err})
-        }
+    this._dialWithTimeout(transport, addr, (err, conn) => {
+      if (err) {
+        log('work:error')
+        return callback(null, {error: err})
+      }
 
-        if (token.cancel) {
-          log('work:cancel')
-          // clean up already done dials
-          pull(pull.empty(), conn)
-          // TODO: proper cleanup once the connection interface supports it
-          // return conn.close(() => callback(new Error('Manual cancel'))
-          return callback(null, {cancel: true})
-        }
+      if (token.cancel) {
+        log('work:cancel')
+        // clean up already done dials
+        pull(pull.empty(), conn)
+        // TODO: proper cleanup once the connection interface supports it
+        // return conn.close(() => callback(new Error('Manual cancel'))
+        return callback(null, {cancel: true})
+      }
 
-        // one is enough
-        token.cancel = true
+      // one is enough
+      token.cancel = true
 
-        log('work:success')
+      log('work:success')
 
-        const proxyConn = new Connection()
-        proxyConn.setInnerConn(conn)
-        callback(null, {conn})
-      }
-    )
+      const proxyConn = new Connection()
+      proxyConn.setInnerConn(conn)
+      callback(null, { multiaddr: addr, conn: conn })
+    })
   }
 
   /**

src/transport.js

@@ -24,7 +24,8 @@ module.exports = function (swarm) {
         options = {}
       }
 
-      if (!callback) { callback = noop }
+      callback = callback || noop
+
       log('adding %s', key)
       if (swarm.transports[key]) {
         throw new Error('There is already a transport with this key')
@@ -39,7 +40,7 @@ module.exports = function (swarm) {
 
     dial (key, pi, callback) {
       const t = swarm.transports[key]
-      let multiaddrs = pi.multiaddrs.slice()
+      let multiaddrs = pi.multiaddrs.toArray()
 
       if (!Array.isArray(multiaddrs)) {
         multiaddrs = [multiaddrs]
@@ -48,7 +49,15 @@ module.exports = function (swarm) {
       // filter the multiaddrs that are actually valid for this transport (use a func from the transport itself) (maybe even make the transport do that)
       multiaddrs = dialables(t, multiaddrs)
 
-      dialer.dialMany(pi.id, t, multiaddrs, callback)
+      dialer.dialMany(pi.id, t, multiaddrs, (err, success) => {
+        if (err) {
+          return callback(err)
+        }
+
+        pi.connect(success.multiaddr)
+        swarm._peerBook.put(pi)
+        callback(null, success.conn)
+      })
     },
 
     listen (key, options, handler, callback) {
@@ -57,7 +66,7 @@ module.exports = function (swarm) {
         handler = protocolMuxer.bind(null, swarm.protocols)
       }
 
-      const multiaddrs = dialables(swarm.transports[key], swarm._peerInfo.distinctMultiaddr())
+      const multiaddrs = dialables(swarm.transports[key], swarm._peerInfo.multiaddrs.distinct())
 
       const transport = swarm.transports[key]
 
@@ -96,7 +105,7 @@ module.exports = function (swarm) {
         }
 
         // cause we can listen on port 0 or 0.0.0.0
-        swarm._peerInfo.multiaddr.replace(multiaddrs, freshMultiaddrs)
+        swarm._peerInfo.multiaddrs.replace(multiaddrs, freshMultiaddrs)
         callback()
       })
     },