import { EventEmitter } from 'events' ; import * as dgram from 'dgram' ; import ms from 'ms' ; import { debug as createDebugLogger } from 'debug' ; import LRUCache = require ( 'lru-cache' ) ; import { encode , decode } from './message' ; import { keccak256 , pk2id , createDeferred , formatLogId } from '../util' ; import { DPT , PeerInfo } from './dpt' ; import { Socket as DgramSocket , RemoteInfo } from 'dgram' ; const DEBUG_BASE_NAME = 'devp2p:dpt:server' ; const debug = createDebugLogger ( DEBUG_BASE_NAME ) ; const verbose = createDebugLogger ( 'verbose' ) . enabled ; const VERSION = 0x04 ; export interface DPTServerOptions { /**
     * Timeout for peer requests
     *
     * Default: 10s
     */ timeout ? : number ; /**
     * Network info to send a long a request
     *
     * Default: 0.0.0.0, no UDP or TCP port provided
     */ endpoint ? : PeerInfo ; /**
     * Function for socket creation
     *
     * Default: dgram-created socket
     */ createSocket ? : Function ; } export class Server extends EventEmitter { _dpt : DPT ; _privateKey : Buffer ; _timeout : number ; _endpoint : PeerInfo ; _requests : Map < string , any > ; _parityRequestMap : Map < string , string > ; _requestsCache : LRUCache < string , Promise < any > > ; _socket : DgramSocket | null ; private msgDebuggers : { [ key ] : ( debug : string ) => void ; } = { } ; constructor ( dpt , privateKey , options ) { super ( ) ; this . _dpt = dpt ; this . _privateKey = privateKey ; this . _timeout = options . timeout ?? ms ( '10s' ) ; this . _endpoint = options . endpoint ?? { address : '0.0.0.0' , udpPort : null , tcpPort : null } ; this . _requests = new Map ( ) ; this . _parityRequestMap = new Map ( ) ; this . _requestsCache = new LRUCache ( { max : 1000 , maxAge : ms ( '1s' ) , stale : false } ) ; this . initMsgDebuggers ( ) ; const createSocket = options . createSocket ?? dgram . createSocket . bind ( null , { type : 'udp4' } ) ; this . _socket = createSocket ( ) ; if ( this . _socket ) { this . _socket . once ( 'listening' , ( ) => this . emit ( 'listening' ) ) ; this . _socket . once ( 'close' , ( ) => this . emit ( 'close' ) ) ; this . _socket . on ( 'error' , ( err ) => this . emit ( 'error' , err ) ) ; this . _socket . on ( 'message' , ( msg , rinfo ) => { try { this . _handler ( msg , rinfo ) ; } catch ( err ) { this . emit ( 'error' , err ) ; } } ) ; } } bind ( ... args ) { this . _isAliveCheck ( ) ; debug ( 'call .bind' ) ; if ( this . _socket ) this . _socket . bind ( ... args ) ; } destroy ( ... args ) { this . _isAliveCheck ( ) ; debug ( 'call .destroy' ) ; if ( this . _socket ) { this . _socket . close ( ... args ) ; this . _socket = null ; } } async ping ( peer ) { this . _isAliveCheck ( ) ; const rckey = `${ peer . address }:${ peer . udpPort }` ; const promise = this . _requestsCache . get ( rckey ) ; if ( promise !== undefined ) return promise ; const hash = this . _send ( peer , 'ping' , { version : VERSION , from : this . _endpoint , to : peer , } ) ; const deferred = createDeferred ( ) ; const rkey = hash . toString ( 'hex' ) ; this . _requests . set ( rkey , { peer , deferred , timeoutId : setTimeout ( ( ) => { if ( this . _requests . get ( rkey ) !== undefined ) { debug ( `ping timeout: ${ peer . address }:${ peer . udpPort } ${ peer . id ? formatLogId ( peer . id . toString ( 'hex' ) , verbose ) : '-' }` ) ; this . _requests . delete ( rkey ) ; deferred . reject ( new Error ( `Timeout error: ping ${ peer . address }:${ peer . udpPort }` ) ) ; } else { return deferred . promise ; } } , this . _timeout ) , } ) ; this . _requestsCache . set ( rckey , deferred . promise ) ; return deferred . promise ; } findneighbours ( peer , id ) { this . _isAliveCheck ( ) ; this . _send ( peer , 'findneighbours' , { id } ) ; } _isAliveCheck ( ) { if ( this . _socket === null ) throw new Error ( 'Server already destroyed' ) ; } _send ( peer , typename , data ) { const debugMsg = `send ${ typename } to ${ peer . address }:${ peer . udpPort } (peerId: ${ peer . id ? formatLogId ( peer . id . toString ( 'hex' ) , verbose ) : '-' })` ; this . debug ( typename , debugMsg ) ; const msg = encode ( typename , data , this . _privateKey ) ; if ( typename === 'ping' ) { const rkeyParity = keccak256 ( msg . slice ( 98 ) ) . toString ( 'hex' ) ; this . _parityRequestMap . set ( rkeyParity , msg . slice ( 0 , 32 ) . toString ( 'hex' ) ) ; setTimeout ( ( ) => { if ( this . _parityRequestMap . get ( rkeyParity ) !== undefined ) { this . _parityRequestMap . delete ( rkeyParity ) ; } } , this . _timeout ) ; } if ( this . _socket && peer . udpPort ) this . _socket . send ( msg , 0 , msg . length , peer . udpPort , peer . address ) ; return msg . slice ( 0 , 32 ) ; } _handler ( msg , rinfo ) { const info = decode ( msg ) ; const peerId = pk2id ( info . publicKey ) ; const debugMsg = `received ${ info . typename } from ${ rinfo . address }:${ rinfo . port } (peerId: ${ formatLogId ( peerId . toString ( 'hex' ) , verbose ) })` ; this . debug ( info . typename . toString ( ) , debugMsg ) ; const peer = this . _dpt . getPeer ( peerId ) ; if ( peer === null && info . typename === 'ping' && info . data . from . udpPort !== null ) { setTimeout ( ( ) => this . emit ( 'peers' , [ info . data . from ] ) , ms ( '100ms' ) ) ; } switch ( info . typename ) { case 'ping' : { const remote = { id : peerId , udpPort : rinfo . port , address : rinfo . address , } ; this . _send ( remote , 'pong' , { to : { address : rinfo . address , udpPort : rinfo . port , tcpPort : info . data . from . tcpPort , } , hash : msg . slice ( 0 , 32 ) , } ) ; break ; } case 'pong' : { let rkey = info . data . hash . toString ( 'hex' ) ; const rkeyParity = this . _parityRequestMap . get ( rkey ) ; if ( rkeyParity ) { rkey = rkeyParity ; this . _parityRequestMap . delete ( rkeyParity ) ; } const request = this . _requests . get ( rkey ) ; if ( request ) { this . _requests . delete ( rkey ) ; request . deferred . resolve ( { id : peerId , address : request . peer . address , udpPort : request . peer . udpPort , tcpPort : request . peer . tcpPort , } ) ; } break ; } case 'findneighbours' : { const remote = { id : peerId , udpPort : rinfo . port , address : rinfo . address , } ; this . _send ( remote , 'neighbours' , { peers : this . _dpt . getClosestPeers ( info . data . id ) , } ) ; break ; } case 'neighbours' : { this . emit ( 'peers' , info . data . peers . map ( ( peer ) => peer . endpoint ) ) ; break ; } } } private initMsgDebuggers ( ) { const MESSAGE_NAMES = [ 'ping' , 'pong' , 'findneighbours' , 'neighbours' ] ; for ( const name of MESSAGE_NAMES ) { this . msgDebuggers [ name ] = createDebugLogger ( `${ DEBUG_BASE_NAME }:${ name }` ) ; } } param messageName param msg private debug ( messageName , msg ) { debug ( msg ) ; if ( this . msgDebuggers [ messageName ] ) { this . msgDebuggers [ messageName ] ( msg ) ; } } }