Testing refactor switch merge

package.json

@@ -50,7 +50,7 @@
     "libp2p-connection-manager": "^0.1.0",
     "libp2p-floodsub": "^0.16.1",
     "libp2p-ping": "^0.8.5",
-    "libp2p-switch": "^0.42.12",
+    "libp2p-switch": "./src/switch",
     "libp2p-websockets": "^0.12.2",
     "mafmt": "^6.0.7",
     "multiaddr": "^6.1.0",

src/switch/connection/base.js

@@ -0,0 +1,126 @@
+'use strict'
+
+const EventEmitter = require('events').EventEmitter
+const debug = require('debug')
+const withIs = require('class-is')
+
+class BaseConnection extends EventEmitter {
+  constructor ({ _switch, name }) {
+    super()
+
+    this.switch = _switch
+    this.ourPeerInfo = this.switch._peerInfo
+    this.log = debug(`libp2p:conn:${name}`)
+    this.log.error = debug(`libp2p:conn:${name}:error`)
+  }
+
+  /**
+   * Puts the state into its disconnecting flow
+   *
+   * @param {Error} err Will be emitted if provided
+   * @returns {void}
+   */
+  close (err) {
+    if (this._state._state === 'DISCONNECTING') return
+    this.log('closing connection to %s', this.theirB58Id)
+    if (err && this._events.error) {
+      this.emit('error', err)
+    }
+    this._state('disconnect')
+  }
+
+  emit (eventName, ...args) {
+    if (eventName === 'error' && !this._events.error) {
+      this.log.error(...args)
+    } else {
+      super.emit(eventName, ...args)
+    }
+  }
+
+  /**
+   * Gets the current state of the connection
+   *
+   * @returns {string} The current state of the connection
+   */
+  getState () {
+    return this._state._state
+  }
+
+  /**
+   * Puts the state into encrypting mode
+   *
+   * @returns {void}
+   */
+  encrypt () {
+    this._state('encrypt')
+  }
+
+  /**
+   * Puts the state into privatizing mode
+   *
+   * @returns {void}
+   */
+  protect () {
+    this._state('privatize')
+  }
+
+  /**
+   * Puts the state into muxing mode
+   *
+   * @returns {void}
+   */
+  upgrade () {
+    this._state('upgrade')
+  }
+
+  /**
+   * Event handler for disconnected.
+   *
+   * @fires BaseConnection#close
+   * @returns {void}
+   */
+  _onDisconnected () {
+    this.switch.connection.remove(this)
+    this.log('disconnected from %s', this.theirB58Id)
+    this.emit('close')
+    this.removeAllListeners()
+  }
+
+  /**
+   * Event handler for privatized
+   *
+   * @fires BaseConnection#private
+   * @returns {void}
+   */
+  _onPrivatized () {
+    this.emit('private', this.conn)
+  }
+
+  /**
+   * Wraps this.conn with the Switch.protector for private connections
+   *
+   * @private
+   * @fires ConnectionFSM#error
+   * @returns {void}
+   */
+  _onPrivatizing () {
+    if (!this.switch.protector) {
+      return this._state('done')
+    }
+
+    this.conn = this.switch.protector.protect(this.conn, (err) => {
+      if (err) {
+        return this.close(err)
+      }
+
+      this.log('successfully privatized conn to %s', this.theirB58Id)
+      this.conn.setPeerInfo(this.theirPeerInfo)
+      this._state('done')
+    })
+  }
+}
+
+module.exports = withIs(BaseConnection, {
+  className: 'BaseConnection',
+  symbolName: 'libp2p-switch/BaseConnection'
+})

src/switch/connection/handler.js

@@ -0,0 +1,47 @@
+'use strict'
+
+const debug = require('debug')
+const IncomingConnection = require('./incoming')
+const observeConn = require('../observe-connection')
+
+function listener (_switch) {
+  const log = debug(`libp2p:switch:listener`)
+
+  /**
+   * Takes a transport key and returns a connection handler function
+   *
+   * @param {string} transportKey The key of the transport to handle connections for
+   * @param {function} handler A custom handler to use
+   * @returns {function(Connection)} A connection handler function
+   */
+  return function (transportKey, handler) {
+    /**
+     * Takes a base connection and manages listening behavior
+     *
+     * @param {Connection} conn The connection to manage
+     * @returns {void}
+     */
+    return function (conn) {
+      log('received incoming connection for transport %s', transportKey)
+      conn.getPeerInfo((_, peerInfo) => {
+        // Add a transport level observer, if needed
+        const connection = transportKey ? observeConn(transportKey, null, conn, _switch.observer) : conn
+        const connFSM = new IncomingConnection({ connection, _switch, transportKey, peerInfo })
+
+        connFSM.once('error', (err) => log(err))
+        connFSM.once('private', (_conn) => {
+          // Use the custom handler, if it was provided
+          if (handler) {
+            return handler(_conn)
+          }
+          connFSM.encrypt()
+        })
+        connFSM.once('encrypted', () => connFSM.upgrade())
+
+        connFSM.protect()
+      })
+    }
+  }
+}
+
+module.exports = listener

src/switch/connection/incoming.js

@@ -0,0 +1,115 @@
+'use strict'
+
+const FSM = require('fsm-event')
+const multistream = require('multistream-select')
+const withIs = require('class-is')
+
+const BaseConnection = require('./base')
+
+class IncomingConnectionFSM extends BaseConnection {
+  constructor ({ connection, _switch, transportKey, peerInfo }) {
+    super({
+      _switch,
+      name: `inc:${_switch._peerInfo.id.toB58String().slice(0, 8)}`
+    })
+    this.conn = connection
+    this.theirPeerInfo = peerInfo || null
+    this.theirB58Id = this.theirPeerInfo ? this.theirPeerInfo.id.toB58String() : null
+    this.ourPeerInfo = this.switch._peerInfo
+    this.transportKey = transportKey
+    this.protocolMuxer = this.switch.protocolMuxer(this.transportKey)
+    this.msListener = new multistream.Listener()
+
+    this._state = FSM('DIALED', {
+      DISCONNECTED: {
+        disconnect: 'DISCONNECTED'
+      },
+      DIALED: { // Base connection to peer established
+        privatize: 'PRIVATIZING',
+        encrypt: 'ENCRYPTING'
+      },
+      PRIVATIZING: { // Protecting the base connection
+        done: 'PRIVATIZED',
+        disconnect: 'DISCONNECTING'
+      },
+      PRIVATIZED: { // Base connection is protected
+        encrypt: 'ENCRYPTING'
+      },
+      ENCRYPTING: { // Encrypting the base connection
+        done: 'ENCRYPTED',
+        disconnect: 'DISCONNECTING'
+      },
+      ENCRYPTED: { // Upgrading could not happen, the connection is encrypted and waiting
+        upgrade: 'UPGRADING',
+        disconnect: 'DISCONNECTING'
+      },
+      UPGRADING: { // Attempting to upgrade the connection with muxers
+        done: 'MUXED'
+      },
+      MUXED: {
+        disconnect: 'DISCONNECTING'
+      },
+      DISCONNECTING: { // Shutting down the connection
+        done: 'DISCONNECTED'
+      }
+    })
+
+    this._state.on('DISCONNECTED', () => this._onDisconnected())
+    this._state.on('PRIVATIZING', () => this._onPrivatizing())
+    this._state.on('PRIVATIZED', () => this._onPrivatized())
+    this._state.on('ENCRYPTING', () => this._onEncrypting())
+    this._state.on('ENCRYPTED', () => {
+      this.log('successfully encrypted connection to %s', this.theirB58Id || 'unknown peer')
+      this.emit('encrypted', this.conn)
+    })
+    this._state.on('UPGRADING', () => this._onUpgrading())
+    this._state.on('MUXED', () => {
+      this.log('successfully muxed connection to %s', this.theirB58Id || 'unknown peer')
+      this.emit('muxed', this.conn)
+    })
+    this._state.on('DISCONNECTING', () => {
+      this._state('done')
+    })
+  }
+
+  /**
+   * Attempts to encrypt `this.conn` with the Switch's crypto.
+   *
+   * @private
+   * @fires IncomingConnectionFSM#error
+   * @returns {void}
+   */
+  _onEncrypting () {
+    this.log('encrypting connection via %s', this.switch.crypto.tag)
+
+    this.msListener.addHandler(this.switch.crypto.tag, (protocol, _conn) => {
+      this.conn = this.switch.crypto.encrypt(this.ourPeerInfo.id, _conn, undefined, (err) => {
+        if (err) {
+          return this.close(err)
+        }
+        this.conn.getPeerInfo((_, peerInfo) => {
+          this.theirPeerInfo = peerInfo
+          this._state('done')
+        })
+      })
+    }, null)
+
+    // Start handling the connection
+    this.msListener.handle(this.conn, (err) => {
+      if (err) {
+        this.emit('crypto handshaking failed', err)
+      }
+    })
+  }
+
+  _onUpgrading () {
+    this.log('adding the protocol muxer to the connection')
+    this.protocolMuxer(this.conn, this.msListener)
+    this._state('done')
+  }
+}
+
+module.exports = withIs(IncomingConnectionFSM, {
+  className: 'IncomingConnectionFSM',
+  symbolName: 'libp2p-switch/IncomingConnectionFSM'
+})