import PeerMemoryRepository from 'core/repository/peer/peerMemory' ; import PeerNetworkRepository from 'core/repository/peer/peerNetwork' ; import { NetworkPeer } from 'shared/model/Peer/networkPeer' ; import SocketDriver from 'core/driver/socket/index' ; import SystemRepository from 'core/repository/system' ; import config from 'shared/config' ; import { logger } from 'shared/util/logger' ; import { PeerAddress , PeerHeadersReceived , RequestPeerInfo , ShortPeerInfo } from 'shared/model/types' ; import { Headers } from 'shared/model/Peer/headers' ; import { sortByKey } from 'shared/util/util' ; import VersionChecker from 'core/util/versionChecker' ; import { isArray } from 'util' ; import SwapTransactionQueue from 'core/service/swapTransactionQueue' ; import { messageON } from 'shared/util/bus' ; import { ActionTypes } from 'core/util/actionTypes' ; import { ResponseEntity } from 'shared/model/response' ; import { asyncTimeout } from 'shared/util/timer' ; import { diffArrayPeers , peerAddressToString } from 'core/util/peer' ; import { createBufferObject } from 'shared/util/byteSerializer' ; import { SchemaName } from 'shared/util/byteSerializer/config' ; const STEP_RECONNECT_DELAY = 600 ; const LOG_PREFIX = `[Service][Peer]` ; export const ERROR_NOT_ENOUGH_PEERS = 'ERROR_NOT_ENOUGH_PEERS' ; export class PeerService { add ( data ) { if ( config . CORE . PEERS . BLACKLIST . indexOf ( data . peerAddress . ip ) !== - 1 || ! VersionChecker . isAcceptable ( data . peerHeaders . version ) ) { logger . debug ( `[Service][peer][add] invalid version ${ data . peerHeaders . version }` ) ; data . socket . disconnect ( true ) ; return false ; } if ( PeerMemoryRepository . has ( data . peerAddress ) ) { return this . resolveConnectionConflict ( data ) ; } else { PeerMemoryRepository . add ( data . peerAddress , data . peerHeaders , data . type ) ; PeerNetworkRepository . add ( data . peerAddress , data . socket ) ; } return true ; } resolveConnectionConflict ( data ) { logger . debug ( `${ LOG_PREFIX }[add] sockets conflict detected on ${ data . peerAddress . ip }` ) ; const networkPeer = PeerNetworkRepository . get ( data . peerAddress ) ; const ids = [ data . socket , networkPeer ] . sort ( sortByKey ( 'id' ) ) ; if ( ids . indexOf ( networkPeer ) === 0 ) { logger . debug ( `${ LOG_PREFIX }[add] destroy new connection on ${ data . peerAddress . ip }, id ${ data . socket . id }` ) ; data . socket . disconnect ( true ) ; return false ; } else { logger . debug ( `${ LOG_PREFIX }[add] delete old connection ${ networkPeer . id }, create new peer ` + `${ data . peerAddress . ip }` ) ; this . remove ( data . peerAddress , false ) ; PeerMemoryRepository . add ( data . peerAddress , data . peerHeaders , data . type ) ; PeerNetworkRepository . add ( data . peerAddress , data . socket ) ; return true ; } } remove ( peerAddress , checkMinPeerCount = true ) { logger . trace ( `${ LOG_PREFIX }[remove] ${ peerAddressToString ( peerAddress ) }` ) ; PeerMemoryRepository . remove ( peerAddress ) ; PeerNetworkRepository . remove ( peerAddress ) ; if ( checkMinPeerCount && PeerMemoryRepository . count < config . CONSTANTS . PEERS_DISCOVER . MIN ) { messageON ( ActionTypes . EMIT_REQUEST_PEERS ) ; } } removeAll ( ) { PeerMemoryRepository . removeAll ( ) ; PeerNetworkRepository . removeAll ( ) ; } broadcast ( code , data , peerAddresses ? , checkQueue = true ) { if ( checkQueue && SwapTransactionQueue . size && PeerNetworkRepository . count ) { SwapTransactionQueue . process ( ) ; } let peers = [ ] ; if ( peerAddresses && isArray ( peerAddresses ) ) { peers = PeerNetworkRepository . getManyByAddress ( peerAddresses ) ; } else { peers = PeerNetworkRepository . getAll ( ) ; } const byteMessage = createBufferObject ( { code , data } , SchemaName . Request ) ; peers . forEach ( ( peer ) => peer . send ( byteMessage ) ) ; } update ( peerAddress , headers ) { if ( ! PeerMemoryRepository . has ( peerAddress ) ) { logger . debug ( `[${ LOG_PREFIX }][update] peer does not found ${ peerAddress . ip }` ) ; return ; } const peer = PeerMemoryRepository . get ( peerAddress ) ; peer . update ( headers ) ; if ( PeerNetworkRepository . isBanned ( peerAddress ) ) { if ( headers . height === SystemRepository . headers . height && headers . broadhash === SystemRepository . headers . broadhash ) { PeerNetworkRepository . unban ( peerAddress ) ; } } } filterPeers ( peers ) { return peers . filter ( ( peer ) => { return peer . peerCount < config . CONSTANTS . MAX_PEERS_CONNECTED && ! PeerNetworkRepository . isBanned ( peer ) && config . CORE . PEERS . BLACKLIST . indexOf ( peer . ip ) === - 1 ; } ) ; } async stepReconnect ( currentPeers , newPeers ) { const currentUnique = diffArrayPeers ( currentPeers , newPeers ) ; const newUnique = diffArrayPeers ( newPeers , currentPeers ) ; let i = 0 ; while ( i < currentUnique . length || i < newUnique . length ) { if ( currentUnique [ i ] ) { this . remove ( currentUnique [ i ] , false ) ; } if ( newUnique [ i ] ) { this . connectPeer ( newUnique [ i ] ) ; } i ++ ; await asyncTimeout ( STEP_RECONNECT_DELAY ) ; } } connectPeer ( peerAddress ) { const headers = SystemRepository . getFullHeaders ( ) ; if ( ! PeerMemoryRepository . has ( peerAddress ) ) { SocketDriver . connectPeer ( peerAddress , headers . serialize ( ) ) ; } } connectPeers ( peerList ) { peerList . forEach ( ( peerAddress ) => { this . connectPeer ( peerAddress ) ; } ) ; } ping ( ) { PeerNetworkRepository . getAll ( ) . forEach ( ( peer ) => { logger . trace ( `[Service][Peer][ping] ${ peer . peerAddress . ip }` ) ; peer . requestRPC ( ActionTypes . PING , createBufferObject ( { } , SchemaName . Empty ) ) . then ( ( response ) => { if ( ! response . success ) { logger . debug ( `[Service][Peer][ping] pong was failed ${ peer . peerAddress . ip }` ) ; messageON ( ActionTypes . REMOVE_PEER , peer . peerAddress ) ; } } ) ; } ) ; } pong ( requestPeerInfo ) { if ( ! PeerNetworkRepository . has ( requestPeerInfo . peerAddress ) ) { logger . trace ( `[Service][Peer][pong] peer is offline for response ${ requestPeerInfo . peerAddress . ip }` ) ; return ; } const networkPeer = PeerNetworkRepository . get ( requestPeerInfo . peerAddress ) ; networkPeer . responseRPC ( ActionTypes . PONG , createBufferObject ( { } , SchemaName . Empty ) , requestPeerInfo . requestId ) ; } } export default new PeerService ( ) ;