feat: initial implementation of circuit relaying

Author: dmitriy ryajov

package.json

@@ -1,45 +1,52 @@
 {
   "name": "libp2p-circuit",
-  "version": "0.9.2",
+  "version": "0.0.1",
   "description": "JavaScript implementation of circuit/switch relaying",
   "main": "src/index.js",
   "scripts": {
     "lint": "aegir-lint",
-    "test": "gulp test",
-    "test:node": "gulp test:node",
-    "test:browser": "gulp test:browser",
-    "build": "gulp build",
-    "release": "gulp release",
-    "release-minor": "gulp release --type minor",
-    "release-major": "gulp release --type major",
-    "coverage": "gulp coverage",
+    "build": "aegir-build",
+    "test": "aegir-test --env node",
+    "release": "aegir-release",
+    "release-minor": "aegir-release --type minor",
+    "release-major": "aegir-release --type major",
+    "coverage": "aegir-coverage",
     "coverage-publish": "aegir-coverage publish"
   },
-  "browser": {
-    "pull-ws/server": false
-  },
   "pre-commit": [
     "lint",
     "test"
   ],
   "repository": {
     "type": "git",
-    "url": "git+https://github.com/libp2p/js-libp2p-websockets.git"
+    "url": "git+https://github.com/libp2p/js-libp2p-circuit.git"
   },
   "keywords": [
     "IPFS"
   ],
-  "author": "David Dias <[email protected]>",
+  "author": "Dmitriy Ryajov <[email protected]>",
   "license": "MIT",
   "bugs": {
-    "url": "https://github.com/dryajov/js-libp2p-circuit/issues"
+    "url": "https://github.com/libp2p/js-libp2p-circuit/issues"
   },
-  "homepage": "https://github.com/dryajov/js-libp2p-circuit#readme",
+  "homepage": "https://github.com/libp2p/js-libp2p-circuit#readme",
   "devDependencies": {
     "aegir": "^10.0.0",
+    "chai": "^3.5.0",
+    "libp2p-ipfs-nodejs": "^0.19.0",
+    "peer-id": "^0.8.2",
+    "peer-info": "^0.8.3",
     "pre-commit": "^1.2.2"
   },
-  "contributors": [
-    "Dmitriy Ryajov <[email protected]>"
-  ]
+  "contributors": [],
+  "dependencies": {
+    "async": "^2.1.5",
+    "debug": "^2.6.1",
+    "interface-connection": "^0.3.1",
+    "lodash": "^4.17.4",
+    "multiaddr": "^2.2.1",
+    "multistream-select": "^0.13.4",
+    "pull-handshake": "^1.1.4",
+    "pull-stream": "^3.5.0"
+  }
 }

src/config.js

@@ -0,0 +1,11 @@
+'use strict'
+
+const debug = require('debug')
+
+const log = debug('libp2p:circuit')
+log.err = debug('libp2p:circuit:error')
+
+module.exports = {
+  log: log,
+  multicodec: '/ipfs/relay/circuit/1.0.0'
+}

src/index.js

@@ -0,0 +1,6 @@
+'use strict'
+
+exports.Dialer = require('./transport/dialer')
+exports.Listener = require('./transport/listener')
+exports.Peer = require('./peer')
+exports.Relay = require('./relay')

src/peer.js

@@ -0,0 +1,42 @@
+'use strict'
+
+/**
+ * The known state of a connected peer.
+ */
+class Peer {
+  /**
+   * @param {Connection} conn
+   * @param {PeerInfo} peerInfo
+   */
+  constructor (conn, peerInfo) {
+    /**
+     * @type {Connection}
+     */
+    this.conn = conn
+
+    /**
+     * @type {PeerInfo}
+     */
+    this.peerInfo = peerInfo
+  }
+
+  /**
+   * Attach a connection
+   * @param {Connection} conn
+   * @returns {void}
+   */
+  attachConnection (conn) {
+    this.conn = conn
+  }
+
+  /**
+   * Is the peer connected currently?
+   *
+   * @type {boolean}
+   */
+  get isConnected () {
+    return Boolean(this.conn)
+  }
+}
+
+module.exports = Peer

src/relay.js

@@ -0,0 +1,134 @@
+'use strict'
+
+const pull = require('pull-stream')
+const lp = require('pull-length-prefixed')
+const multiaddr = require('multiaddr')
+const config = require('./config')
+const Peer = require('./peer')
+const handshake = require('pull-handshake')
+const mss = require('multistream-select')
+const Connection = require('interface-connection').Connection
+
+const multicodec = require('./config').multicodec
+
+const log = config.log
+
+class Relay {
+  constructor (libp2p) {
+    this.libp2p = libp2p
+    this.peers = new Map()
+
+    this._onConnection = this._onConnection.bind(this)
+    this._dialPeer = this._dialPeer.bind(this)
+  }
+
+  start (cb) {
+    this.libp2p.handle(multicodec, this._onConnection)
+    cb()
+  }
+
+  stop (cb) {
+    this.libp2p.unhandle(multicodec)
+    cb()
+  }
+
+  _dialPeer (ma, callback) {
+    let idB58Str
+
+    try {
+      idB58Str = ma.peerId() // try to get the peerid from the multiaddr
+    } catch (err) {
+      log.err(err)
+    }
+
+    if (idB58Str) {
+      const peer = this.peers.get(idB58Str)
+      if (peer && peer.isConnected()) {
+        return
+      }
+    }
+
+    this.libp2p.dialByMultiaddr(ma, multicodec, (err, conn) => {
+      if (err) {
+        log.err(err)
+        return callback(err)
+      }
+
+      conn.getPeerInfo((err, peerInfo) => {
+        if (err) {
+          err(err)
+          return
+        }
+
+        const idB58Str = peerInfo.id.toB58String()
+        // If already had a dial to me, just add the conn
+        if (!this.peers.has(idB58Str)) {
+          this.peers.set(idB58Str, new Peer(conn, peerInfo))
+        }
+
+        const peer = this.peers.get(idB58Str)
+        callback(null, peer)
+      })
+    })
+  }
+
+  _onConnection (protocol, conn) {
+    conn.getPeerInfo((err, peerInfo) => {
+      if (err) {
+        log.err('Failed to identify incomming conn', err)
+        return pull(pull.empty(), conn)
+      }
+
+      const idB58Str = peerInfo.id.toB58String()
+      let srcPeer = this.peers.get(idB58Str)
+      if (!srcPeer) {
+        log('new peer', idB58Str)
+        srcPeer = new Peer(conn, peerInfo)
+        this.peers.set(idB58Str, srcPeer)
+      }
+      this._processConnection(srcPeer, conn)
+    })
+  }
+
+  _processConnection (srcPeer, conn) {
+    let stream = handshake({timeout: 1000 * 60})
+    let shake = stream.handshake
+
+    lp.decodeFromReader(shake, (err, msg) => {
+      if (err) {
+        log.err(err)
+        return pull(pull.empty(), conn)
+      }
+
+      let addr = multiaddr(msg.toString())
+      srcPeer.attachConnection(new Connection(shake.rest(), conn))
+      this._circuit(srcPeer, addr)
+    })
+
+    pull(stream, conn, stream)
+  }
+
+  _circuit (srcPeer, ma, callback) {
+    this._dialPeer(ma, (err, destPeer) => {
+      if (err) {
+        log.err(err)
+        return callback(err)
+      }
+
+      let srcAddrs = destPeer.peerInfo.distinctMultiaddr()
+
+      if (!(srcAddrs && srcAddrs.length > 0)) {
+        log.err(`No valid multiaddress for peer!`)
+      }
+
+      let stream = handshake({timeout: 1000 * 60}, callback)
+      let shake = stream.handshake
+
+      mss.util.writeEncoded(shake, `${srcAddrs[0].toString()}/ipfs/${srcPeer.peerInfo.id.toB58String()}`)
+      pull(stream, destPeer.conn, stream)
+      pull(srcPeer.conn, shake.rest(), srcPeer.conn)
+    })
+  }
+}
+
+module.exports = Relay

test/index.spec.js

@@ -0,0 +1,127 @@
+/* eslint-env mocha */
+'use strict'
+
+const Node = require('libp2p-ipfs-nodejs')
+const PeerInfo = require('peer-info')
+const series = require('async/series')
+const parallel = require('async/parallel')
+const pull = require('pull-stream')
+
+const Relay = require('../src').Relay
+const Dialer = require('../src').Dialer
+const Listener = require('../src').Listener
+
+const expect = require('chai').expect
+
+describe(`test circuit`, () => {
+  let srcNode
+  let dstNode
+  let relayNode
+
+  let srcPeer
+  let dstPeer
+  let relayPeer
+
+  let dialer
+  let relayCircuit
+  let listener
+
+  let portBase = 9000 // TODO: randomize or mock sockets
+  before((done) => {
+    series([
+      (cb) => {
+        PeerInfo.create((err, info) => {
+          srcPeer = info
+          srcPeer.multiaddr.add(`/ip4/0.0.0.0/tcp/${portBase++}`)
+          srcNode = new Node(srcPeer)
+          cb(err)
+        })
+      },
+      (cb) => {
+        PeerInfo.create((err, info) => {
+          dstPeer = info
+          dstPeer.multiaddr.add(`/ip4/0.0.0.0/tcp/${portBase++}`)
+          dstNode = new Node(dstPeer)
+          cb(err)
+        })
+      },
+      (cb) => {
+        PeerInfo.create((err, info) => {
+          relayPeer = info
+          relayPeer.multiaddr.add(`/ip4/0.0.0.0/tcp/${portBase++}`)
+          relayNode = new Node(relayPeer)
+          cb(err)
+        })
+      },
+      (cb) => {
+        let relays = new Map()
+        relays.set(relayPeer.id.toB58String(), relayPeer)
+        dialer = new Dialer(srcNode, relays)
+        cb()
+      },
+      (cb) => {
+        relayCircuit = new Relay(relayNode)
+        relayCircuit.start(cb)
+      }],
+      (err) => done(err)
+    )
+  })
+
+  beforeEach((done) => {
+    parallel([
+      (cb) => {
+        srcNode.start(cb)
+      },
+      (cb) => {
+        dstNode.start(cb)
+      },
+      (cb) => {
+        relayNode.start(cb)
+      }
+    ], (err) => done(err))
+  })
+
+  afterEach((done) => {
+    parallel([
+      (cb) => {
+        srcNode.stop(cb)
+      },
+      (cb) => {
+        dstNode.stop(cb)
+      },
+      (cb) => {
+        relayNode.stop(cb)
+      }
+    ], (err) => done(err))
+  })
+
+  it(`should connect to relay peer`, (done) => {
+    listener = new Listener(dstNode, (conn) => {
+      pull(
+        conn,
+        pull.map((data) => {
+          return data.toString().split('').reverse().join('')
+        }),
+        conn
+      )
+    })
+
+    listener.listen(() => {
+    })
+
+    dialer.dial(dstPeer, (err, conn) => {
+      if (err) {
+        done(err)
+      }
+
+      pull(
+        pull.values(['hello']),
+        conn,
+        pull.collect((err, data) => {
+          expect(data[0].toString()).to.equal('olleh')
+          done(err)
+        })
+      )
+    })
+  }).timeout(50000)
+})