This repository has been archived by the owner on Aug 23, 2019. It is now read-only.
Make switch a state machine #278
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| '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}`) | ||
| } | ||
|
|
||
| /** | ||
| * 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.log(`disconnected from ${this.theirB58Id}`) | ||
| this.emit('close') | ||
| this.removeAllListeners() | ||
| } | ||
|
|
||
| /** | ||
| * Event handler for privatized | ||
| * | ||
| * @fires BaseConnection#private | ||
| * @returns {void} | ||
| */ | ||
| _onPrivatized () { | ||
| this.log(`successfully privatized incoming connection`) | ||
| 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) { | ||
| this.emit('error', err) | ||
| return this._state('disconnect') | ||
| } | ||
|
|
||
| this.log(`successfully privatized conn to ${this.theirB58Id}`) | ||
| this.conn.setPeerInfo(this.theirPeerInfo) | ||
| this._state('done') | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| module.exports = withIs(BaseConnection, { | ||
| className: 'BaseConnection', | ||
| symbolName: 'libp2p-switch/BaseConnection' | ||
| }) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| '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 (transportKey, handler) => { | ||
| /** | ||
| * Takes a base connection and manages listening behavior | ||
| * | ||
| * @param {Connection} conn The connection to manage | ||
| * @returns {void} | ||
| */ | ||
| return (conn) => { | ||
| // Add a transport level observer, if needed | ||
| const connection = transportKey ? observeConn(transportKey, null, conn, _switch.observer) : conn | ||
|
|
||
| log('received incoming connection') | ||
| const connFSM = new IncomingConnection({ connection, _switch, transportKey }) | ||
|
|
||
| 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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 }) { | ||
| super({ | ||
| _switch, | ||
| name: `inc:${_switch._peerInfo.id.toB58String().slice(0, 8)}` | ||
| }) | ||
| this.conn = connection | ||
| this.theirPeerInfo = 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: { }, | ||
| 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('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 ${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 ${this.theirB58Id || 'unknown peer'}`) | ||
| this.emit('muxed', this.conn) | ||
| }) | ||
| this._state.on('DISCONNECTING', () => { | ||
| if (this.theirPeerInfo) { | ||
| this.theirPeerInfo.disconnect() | ||
| } | ||
| 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 ${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) { | ||
| this.emit('error', err) | ||
| return this._state('disconnect') | ||
| } | ||
| 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' | ||
| }) |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
better just mimic js-libp2p and do dial and dialProtocol
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@diasdavid are you suggesting to add
dialProtocolin addition to the newdialFSMor as a naming replacement? I think havingdialhandle calls with and without a protocol is reasonable overloading. Replacing it wouldn't be good as it has a different callback footprint to give users access to the ConnectionFSM for finer control/handling.I've exposed the same
dialFSMmethod in the libp2p branchThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We don't need to block this PR for this. What I mean is that given that libp2p (the module) is just an extension to the libp2p-switch (the dialing machine) that adds it other features, it would be cool if they both expose the same dialing/hangup APIs.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That makes sense. Ideally I'd like to deprecate
dialProtocolin libp2p and just usedial. Having both feel unnecessary. I'll propose the change in an issue there and ifdialProtocolends up staying we can add that in here.