feat: removing multihop (onion) dialing

Author: dryajov

src/dialer.js renamed to src/circuit.js

@@ -3,7 +3,7 @@
 const mafmt = require('mafmt')
 const multiaddr = require('multiaddr')
 
-const OnionDialer = require('./circuit/onion-dialer')
+const CircuitDialer = require('./circuit/dialer')
 const utilsFactory = require('./circuit/utils')
 
 const debug = require('debug')
@@ -12,7 +12,7 @@ log.err = debug('libp2p:circuit:error:transportdialer')
 
 const createListener = require('./listener')
 
-class Dialer {
+class Circuit {
   /**
    * Creates an instance of Dialer.
    *
@@ -28,26 +28,22 @@ class Dialer {
     this.dialer = null
     this.utils = utilsFactory(swarm)
     this.peerInfo = this.swarm._peerInfo
-
-    // get all the relay addresses for this swarm
-    const relays = this.filter(this.peerInfo.multiaddrs.toArray())
+    this.relays = this.filter(this.peerInfo.multiaddrs.toArray())
 
     // if no explicit relays, add a default relay addr
-    if (relays.length === 0) {
+    if (this.relays.length === 0) {
       this.peerInfo
         .multiaddrs
         .add(`/p2p-circuit/ipfs/${this.peerInfo.id.toB58String()}`)
     }
 
     // TODO: add flag for other types of dealers, ie telescope
-    this.dialer = new OnionDialer(swarm, options)
+    this.dialer = new CircuitDialer(swarm, options)
 
     this.swarm.on('peer-mux-established', this.dialer.canHop.bind(this.dialer))
     this.swarm.on('peer-mux-closed', (peerInfo) => {
       this.dialer.relayPeers.delete(peerInfo.id.toB58String())
     })
-
-    this._dialSwarmRelays(relays)
   }
 
   /**
@@ -56,18 +52,16 @@ class Dialer {
    * @param {Array} relays
    * @return {void}
    */
-  _dialSwarmRelays (relays) {
+  _dialSwarmRelays () {
     // if we have relay addresses in swarm config, then dial those relays
-    this.swarm.on('listening', () => {
-      relays.forEach((relay) => {
-        let relaySegments = relay
-          .toString()
-          .split('/p2p-circuit')
-          .filter(segment => segment.length)
-
-        relaySegments.forEach((relaySegment) => {
-          this.dialer.dialRelay(this.utils.peerInfoFromMa(multiaddr(relaySegment)))
-        })
+    this.relays.forEach((relay) => {
+      let relaySegments = relay
+        .toString()
+        .split('/p2p-circuit')
+        .filter(segment => segment.length)
+
+      relaySegments.forEach((relaySegment) => {
+        this.dialer.dialRelay(this.utils.peerInfoFromMa(multiaddr(relaySegment)))
       })
     })
   }
@@ -99,7 +93,9 @@ class Dialer {
       options = this.options || {}
     }
 
-    return createListener(this.swarm, options, handler)
+    const listener = createListener(this.swarm, options, handler)
+    listener.on('listen', this._dialSwarmRelays.bind(this))
+    return listener
   }
 
   /**
@@ -121,4 +117,4 @@ class Dialer {
   }
 }
 
-module.exports = Dialer
+module.exports = Circuit

src/circuit/constants.js

@@ -34,14 +34,27 @@ class Dialer {
    * Dial a peer over a relay
    *
    * @param {multiaddr} ma - the multiaddr of the peer to dial
-   * @param {Object} options - dial options
    * @param {Function} cb - a callback called once dialed
    * @returns {Connection} - the connection
    *
    * @memberOf Dialer
    */
-  dial (ma, options, cb) {
-    throw new Error('abstract class, method not implemented')
+  dial (ma, cb) {
+    const addr = ma.toString().split('p2p-circuit')
+    const relay = addr[0] === '/' ? null : multiaddr(addr[0])
+    const peer = multiaddr(addr[1] || addr[0])
+
+    const dstConn = new Connection()
+    setImmediate(this.dialPeer.bind(this), peer, relay, (err, conn) => {
+      if (err) {
+        log.err(err)
+        return cb(err)
+      }
+      dstConn.setInnerConn(conn)
+      cb(null, dstConn)
+    })
+
+    return dstConn
   }
 
   /**
@@ -186,8 +199,22 @@ class Dialer {
             return cb(err)
           }
 
-          this.relayPeers.set(this.utils.getB58String(peer), peer)
-          cb(null)
+          streamHandler.read((err, msg) => {
+            if (err) {
+              log.err(err)
+              return cb(err)
+            }
+
+            const response = proto.CircuitRelay.decode(msg)
+
+            if (response.code !== proto.CircuitRelay.Status.SUCCESS) {
+              return log(`HOP not supported, skipping - ${this.utils.getB58String(peer)}`)
+            }
+
+            log(`HOP supported adding as relay - ${this.utils.getB58String(peer)}`)
+            this.relayPeers.set(this.utils.getB58String(peer), peer)
+            cb(null)
+          })
         })
       })
     }

src/circuit/dialer.js

@@ -8,7 +8,6 @@ const debug = require('debug')
 const PeerInfo = require('peer-info')
 const PeerId = require('peer-id')
 const EE = require('events').EventEmitter
-const constants = require('./constants')
 const once = require('once')
 const utilsFactory = require('./utils')
 const StreamHandler = require('./stream-handler')
@@ -97,7 +96,7 @@ class Hop extends EE {
     this._dialPeer(message.dstPeer, (err, dstConn) => {
       if (err) {
         const errStreamHandler = new StreamHandler(conn)
-        this._writeErr(errStreamHandler, constants.RESPONSE.CANT_DIAL_DST)
+        this.utils.writeResponse(errStreamHandler, proto.CircuitRelay.Status.HOP_CANT_DIAL_DST)
         pull(pull.empty(), errStreamHandler.rest())
         log.err(err)
         return cb(err)
@@ -119,7 +118,7 @@ class Hop extends EE {
         streamHandler.write(proto.CircuitRelay.encode(stopMsg), (err) => {
           if (err) {
             const errStreamHandler = new StreamHandler(conn)
-            this._writeErr(errStreamHandler, constants.RESPONSE.CANT_OPEN_DST_STREAM)
+            this.utils.writeResponse(errStreamHandler, proto.CircuitRelay.Status.HOP_CANT_OPEN_DST_STREAM)
             pull(pull.empty(), errStreamHandler.rest())
 
             log.err(err)

src/circuit/hop.js

@@ -1,8 +1,3 @@
 'use strict'
 
-module.exports = {
-  Hop: require('./circuit/hop'),
-  Dialer: require('./dialer'),
-  multicodec: require('./multicodec'),
-  tag: 'Circuit'
-}
+module.exports = require('./circuit')