import * as querystring from 'querystring' ; import * as socketClusterClient from 'socketcluster-client' ; import { ConnectionKind , DEFAULT_ACK_TIMEOUT , DEFAULT_CONNECT_TIMEOUT , DEFAULT_HTTP_PATH , INTENTIONAL_DISCONNECT_CODE , } from '../constants' ; import { EVENT_CLOSE_OUTBOUND , EVENT_CONNECT_ABORT_OUTBOUND , EVENT_CONNECT_OUTBOUND , EVENT_FAILED_TO_COLLECT_PEER_DETAILS_ON_CONNECT , EVENT_OUTBOUND_SOCKET_ERROR , REMOTE_EVENT_PING , REMOTE_EVENT_PONG , REMOTE_SC_EVENT_MESSAGE , REMOTE_SC_EVENT_RPC_REQUEST , } from '../events' ; import { P2PMessagePacketBufferData , PeerConfig , P2PPeerInfo , P2PNodeInfo , P2PRequestPacketBufferData , P2PResponsePacketBufferData , P2PRawRequestPacket , P2PRequestPacket , } from '../types' ; import { Peer , SCClientSocket , socketErrorStatusCodes , RATE_NORMALIZATION_FACTOR } from './base' ; import { P2PRequest } from '../p2p_request' ; interface ClientOptionsUpdated { readonly hostname : string ; readonly path : string ; readonly port : number ; readonly query : string ; readonly autoConnect : boolean ; readonly autoReconnect : boolean ; readonly multiplex : boolean ; readonly ackTimeout ? : number ; readonly connectTimeout ? : number ; readonly maxPayload ? : number ; } export class OutboundPeer extends Peer { protected _socket : SCClientSocket | undefined ; public constructor ( peerInfo , peerConfig ) { super ( peerInfo , peerConfig ) ; this . _peerInfo . internalState . connectionKind = ConnectionKind . OUTBOUND ; } public set socket ( scClientSocket ) { if ( this . _socket ) { this . _unbindHandlersFromOutboundSocket ( this . _socket ) ; } this . _socket = scClientSocket ; this . _bindHandlersToOutboundSocket ( this . _socket ) ; } public connect ( ) { if ( ! this . _socket ) { this . _socket = this . _createOutboundSocket ( ) ; } this . _socket . connect ( ) ; } public disconnect ( code = INTENTIONAL_DISCONNECT_CODE , reason ? ) { super . disconnect ( code , reason ) ; if ( this . _socket ) { this . _unbindHandlersFromOutboundSocket ( this . _socket ) ; } } public send ( packet ) { if ( ! this . _socket ) { this . _socket = this . _createOutboundSocket ( ) ; } super . send ( packet ) ; } public async request ( packet ) { if ( ! this . _socket ) { this . _socket = this . _createOutboundSocket ( ) ; } return super . request ( packet ) ; } private _createOutboundSocket ( ) { const connectTimeout = this . _peerConfig . connectTimeout ? this . _peerConfig . connectTimeout : DEFAULT_CONNECT_TIMEOUT ; const ackTimeout = this . _peerConfig . ackTimeout ? this . _peerConfig . ackTimeout : DEFAULT_ACK_TIMEOUT ; const { options , ... nodeInfo } = this . _serverNodeInfo as P2PNodeInfo ; const queryObject = { networkVersion : nodeInfo . networkVersion , networkIdentifier : nodeInfo . networkIdentifier , nonce : nodeInfo . nonce , advertiseAddress : nodeInfo . advertiseAddress , port : this . _peerConfig . hostPort , } ; const clientOptions = { hostname : this . ipAddress , path : DEFAULT_HTTP_PATH , port : this . port , query : querystring . stringify ( queryObject as querystring . ParsedUrlQueryInput ) , connectTimeout , ackTimeout , multiplex : false , autoConnect : true , autoReconnect : false , maxPayload : this . _peerConfig . wsMaxPayload , } ; const outboundSocket = socketClusterClient . create ( clientOptions ) ; this . _bindHandlersToOutboundSocket ( outboundSocket ) ; return outboundSocket ; } private _bindHandlersToOutboundSocket ( outboundSocket ) { outboundSocket . on ( 'error' , ( error ) => { this . emit ( EVENT_OUTBOUND_SOCKET_ERROR , error ) ; } ) ; outboundSocket . on ( 'connect' , async ( ) => { try { await this . fetchAndUpdateStatus ( ) ; } catch ( error ) { this . emit ( EVENT_FAILED_TO_COLLECT_PEER_DETAILS_ON_CONNECT , error ) ; return ; } try { await this . discoverPeers ( ) ; } catch ( error ) { this . emit ( EVENT_FAILED_TO_COLLECT_PEER_DETAILS_ON_CONNECT , error ) ; } this . emit ( EVENT_CONNECT_OUTBOUND , this . _peerInfo ) ; } ) ; outboundSocket . on ( 'connectAbort' , ( ) => { this . emit ( EVENT_CONNECT_ABORT_OUTBOUND , this . _peerInfo ) ; } ) ; outboundSocket . on ( 'close' , ( code , reasonMessage ) => { const reason = reasonMessage !== undefined && reasonMessage !== '' ? reasonMessage : socketErrorStatusCodes [ code ] ?? 'Unknown reason' ; this . emit ( EVENT_CLOSE_OUTBOUND , { peerInfo : this . _peerInfo , code , reason , } ) ; } ) ; outboundSocket . on ( 'message' , this . _handleWSMessage ) ; outboundSocket . on ( REMOTE_SC_EVENT_RPC_REQUEST , ( rawRequestPacket , respond ) => { if ( rawRequestPacket . procedure === REMOTE_EVENT_PING ) { this . _updateOutboundRPCCounter ( rawRequestPacket ) ; const rate = this . _getOutboundRPCRate ( rawRequestPacket ) ; const request = new P2PRequest ( { procedure : rawRequestPacket . procedure , data : rawRequestPacket . data , id : this . peerInfo . peerId , rate , productivity : this . internalState . productivity , } , respond ) ; request . end ( REMOTE_EVENT_PONG ) ; return ; } this . _handleRawRPC ( rawRequestPacket , respond ) ; } ) ; outboundSocket . on ( REMOTE_SC_EVENT_MESSAGE , this . _handleRawMessage ) ; const transportSocket = ( outboundSocket as any ) . transport ; if ( transportSocket ?. socket && transportSocket . socket . on ) { transportSocket . socket . on ( REMOTE_EVENT_PING , ( ) => { transportSocket . socket . terminate ( ) ; this . applyPenalty ( 100 ) ; } ) ; transportSocket . socket . on ( REMOTE_EVENT_PONG , ( ) => { transportSocket . socket . terminate ( ) ; this . applyPenalty ( 100 ) ; } ) ; } } private _unbindHandlersFromOutboundSocket ( outboundSocket ) { outboundSocket . off ( 'connect' ) ; outboundSocket . off ( 'connectAbort' ) ; outboundSocket . off ( 'close' ) ; outboundSocket . off ( 'message' , this . _handleWSMessage ) ; outboundSocket . off ( REMOTE_SC_EVENT_RPC_REQUEST , this . _handleRawRPC ) ; outboundSocket . off ( REMOTE_SC_EVENT_MESSAGE , this . _handleRawMessage ) ; outboundSocket . off ( REMOTE_EVENT_PING ) ; } private _updateOutboundRPCCounter ( packet ) { const key = packet . procedure ; const count = ( this . internalState . rpcCounter . get ( key ) ?? 0 ) + 1 ; this . peerInfo . internalState . rpcCounter . set ( key , count ) ; } private _getOutboundRPCRate ( packet ) { const rate = this . peerInfo . internalState . rpcRates . get ( packet . procedure ) ?? 0 ; return rate * RATE_NORMALIZATION_FACTOR ; } }