import { BitField } from "@node-lightning/core" ; import { ILogger } from "@node-lightning/logger" ; import * as noise from "@node-lightning/noise" ; import { NoiseSocket } from "@node-lightning/noise" ; import assert from "assert" ; import { Readable } from "stream" ; import { InitFeatureFlags } from "./flags/InitFeatureFlags" ; import * as MessageFactory from "./MessageFactory" ; import { InitMessage } from "./messages/InitMessage" ; import { IWireMessage } from "./messages/IWireMessage" ; import { PeerState } from "./PeerState" ; import { PingPongState } from "./PingPongState" ; export interface IMessageSender { send ( buf ) : void ; sendMessage ( msg ) : void ; } export interface IMessageReceiver { read ( ) : IWireMessage ; on ( event , listener ) : this ; off ( event , listener ) : this ; } export type IMessageSenderReceiver = IMessageSender & IMessageReceiver ; export interface IPeer extends IMessageSenderReceiver { send ( buf ) : void ; sendMessage ( msg ) : void ; disconnect ( ) : void ; read ( ) : IWireMessage ; on ( event , listener ) : this ; on ( event , listener ) : this ; off ( event , listener ) : this ; off ( event , listener ) : this ; } emits emits emits emits emits emits emits export class Peer extends Readable implements IPeer { public static states = PeerState ; public state : PeerState = PeerState . Disconnected ; public socket : NoiseSocket ; public messageCounter : number = 0 ; public pingPongState : PingPongState ; public logger : ILogger ; public remoteFeatures : BitField < InitFeatureFlags > ; public remoteChains : Buffer [ ] ; public isInitiator : boolean = false ; public reconnectTimeoutMs = 15000 ; private _id : string ; private _rpk : Buffer ; private _host : string ; private _port : number ; private _reconnectHandle : NodeJS . Timeout ; constructor ( readonly ls , readonly localFeatures , readonly localChains , logger , readonly highWaterMark = 2048 ) { super ( { objectMode : true , highWaterMark } ) ; this . pingPongState = new PingPongState ( this ) ; this . logger = logger ; } public get id ( ) : string { return this . _id ; } public get pubkey ( ) : Buffer { return this . _rpk ; } public get pubkeyHex ( ) : string { return this . _rpk ? this . _rpk . toString ( "hex" ) : undefined ; } /**
     * Connect to the remote peer and binds socket events into the Peer.
     */ public connect ( rpk , host , port ) { this . _rpk = rpk ; this . _id = this . _rpk . slice ( 0 , 8 ) . toString ( "hex" ) ; this . logger = this . logger . sub ( "peer" , this . _id ) ; this . logger . info ( "connecting to peer" ) ; this . isInitiator = true ; this . _host = host ; this . _port = port ; this . socket = noise . connect ( { ls : this . ls , host : this . _host , port : this . _port , rpk : this . _rpk , logger : this . logger , } ) ; this . socket . on ( "readable" , this . _onSocketReadable . bind ( this ) ) ; this . socket . on ( "ready" , this . _onSocketReady . bind ( this ) ) ; this . socket . on ( "close" , this . _onSocketClose . bind ( this ) ) ; this . socket . on ( "error" , this . _onSocketError . bind ( this ) ) ; } param socket public attach ( socket ) { this . isInitiator = false ; this . socket = socket ; this . socket . on ( "readable" , this . _onSocketReadable . bind ( this ) ) ; this . socket . on ( "ready" , this . _onSocketReady . bind ( this ) ) ; this . socket . on ( "close" , this . _onSocketClose . bind ( this ) ) ; this . socket . on ( "error" , this . _onSocketError . bind ( this ) ) ; this . socket . once ( "ready" , ( ) => { this . _rpk = socket . rpk ; this . _id = this . _rpk . slice ( 0 , 8 ) . toString ( "hex" ) ; this . logger = this . logger . sub ( "peer" , this . _id ) ; this . logger . info ( "peer connected" ) ; } ) ; } param buf public send ( buf ) { assert . ok ( this . state === PeerState . Ready , new Error ( "Peer is not ready" ) ) ; this . emit ( "sending" , buf ) ; return this . socket . write ( buf ) ; } /**
     * Writes the message on the NoiseSocket using the default
     * serialization properties
     */ public sendMessage ( m ) { assert . ok ( this . state === PeerState . Ready , new Error ( "Peer is not ready" ) ) ; const buf = m . serialize ( ) as Buffer ; this . emit ( "sending" , buf ) ; return this . socket . write ( buf ) ; } /**
     * Closes the socket
     */ public disconnect ( ) { this . logger . info ( "disconnecting" ) ; this . state = PeerState . Disconnecting ; this . socket . end ( ) ; } /**
     * Reconnects the socket
     */ public reconnect ( ) { this . socket . end ( ) ; } private _onSocketReady ( ) { this . state = PeerState . AwaitingPeerInit ; this . logger . debug ( "state: awaiting_peer_init" ) ; this . emit ( "open" ) ; this . _sendInitMessage ( ) ; } private _onSocketClose ( ) { this . logger . debug ( "socket closed" ) ; clearTimeout ( this . _reconnectHandle ) ; if ( this . pingPongState ) this . pingPongState . onDisconnecting ( ) ; if ( this . state === PeerState . Disconnecting ) { this . emit ( "close" ) ; this . state = PeerState . Disconnected ; this . logger . debug ( "permanently disconnected" ) ; return ; } this . state = PeerState . Disconnected ; this . logger . debug ( "state: disconnected" ) ; if ( this . isInitiator ) { this . logger . debug ( `reconnecting in ${ ( this . reconnectTimeoutMs / 1000 ) . toFixed ( 1 ) }s` ) ; this . _reconnectHandle = setTimeout ( ( ) => { this . connect ( this . _rpk , this . _host , this . _port ) ; } , this . reconnectTimeoutMs ) ; } } private _onSocketError ( err ) { this . emit ( "error" , err ) ; } public _onSocketReadable ( ) { this . logger . trace ( "socket is readable" ) ; try { let cont = true ; while ( cont ) { const raw = this . socket . read ( ) as Buffer ; if ( ! raw ) return ; if ( this . state === PeerState . AwaitingPeerInit ) { this . _processPeerInitMessage ( raw ) ; } else { const m = this . _processMessage ( raw ) ; cont = this . push ( m ) ; } } } catch ( err ) { this . logger . error ( err ) ; this . socket . end ( ) ; this . emit ( "error" , err ) ; return ; } } public _read ( ) { setImmediate ( ( ) => this . _onSocketReadable ( ) ) ; } /**
     * Sends the initialization message to the peer. This message
     * does not matter if it is sent before or after the peer sends
     * there message.
     */ private _sendInitMessage ( ) { const msg = new InitMessage ( ) ; msg . features = this . localFeatures ; msg . chainHashes = this . localChains ; const payload = msg . serialize ( ) ; this . emit ( "sending" , payload ) ; this . socket . write ( payload ) ; this . logger . debug ( "init message sent" ) ; } /**
     * Processes the initialization message sent by the remote peer.
     * Once this is successfully completed, the state is transitioned
     * to `active`
     */ private _processPeerInitMessage ( raw ) { const m = MessageFactory . deserialize ( raw ) as InitMessage ; if ( this . logger ) { const features = m . features . flags ( ) ; this . logger . info ( "peer initialized with features" , features ) ; } assert . ok ( m instanceof InitMessage , new Error ( "Expecting InitMessage" ) ) ; this . remoteFeatures = m . features ; this . remoteChains = m . chainHashes ; let hasChain = false ; for ( const remoteChain of this . remoteChains ) { for ( const localChain of this . localChains ) { if ( remoteChain . equals ( localChain ) ) hasChain = true ; } } if ( ! hasChain && this . remoteChains . length && this . localChains . length ) { this . logger . trace ( "remote chains" , ... this . remoteChains ) ; this . logger . trace ( "local chains" , ... this . localChains ) ; this . logger . warn ( "remote node does not support any known chains, aborting" ) ; this . disconnect ( ) ; return ; } for ( const feature of this . localFeatures . flags ( ) ) { if ( feature % 2 === 1 ) continue ; if ( ! ( this . remoteFeatures . isSet ( feature ) || this . remoteFeatures . isSet ( feature + 1 ) ) ) { this . disconnect ( ) ; return ; } } this . pingPongState . start ( ) ; this . state = PeerState . Ready ; this . emit ( "ready" ) ; } /**
     * Process the raw message sent by the peer. These messages are
     * processed after the initialization message has been received.
     */ private _processMessage ( raw ) { this . messageCounter += 1 ; this . emit ( "rawmessage" , raw ) ; const m = MessageFactory . deserialize ( raw ) ; if ( m ) { this . pingPongState . onMessage ( m ) ; } return m ; } }