// ĠVer i Block ĠNode Core Ċ // ĠCopyright Ġ 2 0 1 7 - 2 0 2 1 ĠX en ios ĠSE ZC Ċ // ĠAll Ġrights Ġreserved . Ċ // Ġhttps :// www . ver ib lock . org Ċ // ĠDistributed Ġunder Ġthe ĠMIT Ġsoftware Ġlicense , Ġsee Ġthe Ġaccompanying Ċ // Ġfile ĠLICENSE Ġor Ġhttp :// www . opensource . org / licenses / mit - license . php . Ċ package Ġnode core . p 2 p Ċ Ċ import Ġcom . google . protobuf . InvalidProtocolBufferException Ċ import Ġio . ktor . network . selector . Selector Manager Ċ import Ġio . ktor . network . sockets . Socket Ċ import Ġio . ktor . network . sockets . a Socket Ċ import Ġio . ktor . util . network . Network Address Ċ import Ġio . ktor . util . network . port Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . with Timeout Ċ import Ġnode core . api . grpc . Rpc Event Ċ import Ġnode core . p 2 p . event . Peer Mis behavior Event Ċ import Ġjava . lang . Exception Ċ import Ġjava . nio . channels . Cancelled Key Exception Ċ import Ġorg . ver ib lock . core . crypto . B loom Filter Ċ import Ġorg . ver ib lock . core . utilities . Utility Ċ import Ġorg . ver ib lock . core . utilities . create Logger Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ Ċ private Ġval Ġlogger Ġ= Ġcreate Logger Ġ{} Ċ Ċ class ĠPeer ( ĊĠĠĠ Ġprivate Ġval Ġselector Manager : ĠSelector Manager , ĊĠĠĠ Ġsocket : ĠSocket , ĊĠĠĠ Ġval Ġowns Connection : ĠBoolean Ċ ) Ġ{ ĊĠĠĠ Ġenum Ġclass ĠStatus Ġ{ ĊĠĠĠĠĠĠĠ ĠConnected , ĊĠĠĠĠĠĠĠ ĠEr rored , ĊĠĠĠĠĠĠĠ ĠClosed ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġvar Ġaddress : ĠString Ġ= Ġsocket . remote Address . address ĊĠĠĠ Ġvar Ġport : ĠInt Ġ= Ġsocket . remote Address . port ĊĊĠĠĠ Ġval Ġaddress Key : ĠString ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġ"$ address :$ port " ĊĊĠĠĠ Ġvar Ġreconnect Port : ĠInt ? Ġ= Ġnull ĊĊĠĠĠ Ġvar Ġmetadata Ġ= ĠNode Metadata () ĊĊĠĠĠ Ġval Ġprotocol Version Ġget () Ġ= Ġmetadata . protocol Version ĊĠĠĠ Ġval Ġcan Share Address Ġget () Ġ= Ġmetadata . can Share Address ĊĠĠĠ Ġval Ġcapabilities Ġget () Ġ= Ġmetadata . capabilities ĊĠĠĠ Ġval Ġid : ĠString Ġget () Ġ= Ġmetadata . id ĊĊĠĠĠ Ġvar Ġstatus : ĠStatus Ġ= ĠStatus . Connected ĊĊĠĠĠ Ġvar Ġstate : ĠPeer State Ġ= ĠPeer State () ĊĊĠĠĠ Ġvar Ġfilter : ĠB loom Filter ? Ġ= Ġnull ĊĊĠĠĠ Ġprivate Ġvar Ġsocket Handler Ġ= ĠPeer Socket Handler ( this , Ġsocket ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠExpected Ġresponses Ġfrom Ġmessage Ġrequests Ġsee Ġ[ request Message ] ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġexpected Responses : ĠMutable Map < String , ĠChannel < Rpc Event >> Ġ= ĠConcurrent HashMap () ĊĊĠĠĠ Ġvar Ġre connection Attempts Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġset ĊĊĠĠĠ Ġfun Ġdisconnect () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( status Ġ== ĠStatus . Closed Ġ|| Ġstatus Ġ== ĠStatus . Er rored ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġstatus Ġ= ĠStatus . Closed ĊĠĠĠĠĠĠĠ ĠP 2 p EventBus . peer Disconnected . trigger ( this ) ĊĠĠĠĠĠĠĠ Ġsocket Handler . stop () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġreconnect (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġre connection Attempts ++ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsocket Address Ġ= ĠNetwork Address ( address , Ġport ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠStop Ġthe Ġcurrent Ġconnection ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Handler . stop () ĊĠĠĠĠĠĠĠĠĠĠĠĠĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGet Ġa Ġnew Ġsocket ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsocket Ġ= Ġa Socket ( selector Manager ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. tcp () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. connect ( socket Address ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠReset Ġbasic Ġinfo ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaddress Ġ= Ġsocket . remote Address . address ĊĠĠĠĠĠĠĠĠĠĠĠ Ġport Ġ= Ġsocket . remote Address . port ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate Ġ= ĠPeer State () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstatus Ġ= ĠStatus . Connected ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Handler Ġ= ĠPeer Socket Handler ( this , Ġsocket ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug ( e ) Ġ{ Ġ" Unable Ġto Ġopen Ġconnection Ġto Ġ$ address :$ port !" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠĠĊĠĠĠ Ġvar Ġlast Successful Send Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠ Ġfun Ġsend ( message : ĠRpc Event ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( status Ġ== ĠStatus . Closed Ġ|| Ġstatus Ġ== ĠStatus . Er rored ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn Ġ{ Ġ" Peer Ġ$ address Ġis Ġnot Ġin Ġa Ġstate Ġfor Ġsending , Ġmessage Ġwill Ġnot Ġbe Ġsent " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ Ġ" At tempting Ġto Ġsend Ġ${ message . results Case . name } Ġto Ġpeer Ġ$ address " Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( System . currentTimeMillis () Ġ- Ġ 6 0 0 0 0 Ġ> Ġlast Successful Send ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Note : Ġpeer Ġ$ address Ġhas Ġnot Ġhad Ġa Ġsuccessful Ġwrite Ġin Ġ${ System . currentTimeMillis () Ġ/ Ġ 1 0 0 0 Ġ- Ġlast Successful Send Ġ/ Ġ 1 0 0 0 } Ġseconds " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( System . currentTimeMillis () Ġ- Ġ 1 0 0 0 Ġ* Ġ 5 Ġ* Ġ 6 0 Ġ> Ġlast Successful Send ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn Ġ{ Ġ" Peer Ġ$ address Ġhas Ġnot Ġhad Ġa Ġsuccessful Ġwrite Ġin Ġover Ġ 5 Ġminutes , Ġclosing Ġconnection ..." Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdisconnect () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ignored : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Handler . write ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠCancel led Key Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn Ġ{ Ġ" Key Ġhas Ġbeen Ġcancelled Ġfor Ġ$ address !" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdisconnect () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn ( e ) Ġ{ Ġ" Unable Ġto Ġsend Ġmessage Ġto Ġ$ address " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠĠĊĠĠĠ Ġfun Ġhandle Message ( raw Message : ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġevent : ĠRpc Event ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Ġ= ĠRpc Event . parseFrom ( raw Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ" Received Ġevent Ġof Ġtype Ġ${ event . results Case . name }" Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠInvalid ProtocolBufferException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠP 2 p EventBus . peer Mis behavior . trigger ( Peer Mis behavior Event ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpeer Ġ= Ġthis , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreason Ġ= ĠPeer Mis behavior Event . Reason . MAL FOR MED _ EVENT , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Ġ= Ġ" Malformed Ġevent : Ġ${ Utility . bytes To Hex ( raw Message )}" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġstate . last Message Received At Ġ= ĠUtility . getCurrent Time Seconds () ĊĊĠĠĠĠĠĠĠ Ġ// ĠHandle Ġas Ġan Ġexpected Ġresponse Ġif Ġpossible ĊĠĠĠĠĠĠĠ Ġval ĠrequestId : ĠString ? Ġ= Ġevent . requestId ĊĠĠĠĠĠĠĠ Ġif Ġ( requestId Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Responses [ event . requestId ]?. try Send ( event ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠBroadcast Ġto Ġevent Ġbus ĊĠĠĠĠĠĠĠ ĠP 2 p EventBus . new Event ( event , Ġthis ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠSends Ġa ĠP 2 P Ġrequest Ġand Ġwaits Ġfor Ġthe Ġpeer Ġto Ġrespond Ġit Ġduring Ġthe Ġgiven Ġtimeout Ġ( or Ġa Ġdefault Ġof Ġ 5 Ġseconds ). ĊĠĠĠĠ Ġ* ĠThis Ġapplies Ġto ĠRequest / Response Ġevent Ġtype Ġpairs . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġrequest Message ( ĊĠĠĠĠĠĠĠ Ġrequest : ĠRpc Event , ĊĠĠĠĠĠĠĠ Ġtimeout In Millis : ĠLong Ġ= Ġ 5 0 0 0 L ĊĠĠĠ Ġ): ĠRpc Event Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠCreate Ġcon fl ated Ġchannel ĊĠĠĠĠĠĠĠ Ġval Ġexpected Response Channel Ġ= ĠChannel < Rpc Event >( Channel . CON FL ATED ) ĊĠĠĠĠĠĠĠ Ġ// ĠSet Ġthis Ġchannel Ġas Ġthe Ġexpected Ġresponse Ġfor Ġthe Ġrequest Ġid ĊĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ" Expecting Ġa Ġresponse Ġto Ġ${ request . results Case . name } Ġfrom Ġ$ address " Ġ} ĊĠĠĠĠĠĠĠ Ġexpected Responses [ request . id ] Ġ= Ġexpected Response Channel ĊĠĠĠĠĠĠĠ Ġ// ĠSend Ġthe Ġrequest ĊĠĠĠĠĠĠĠ Ġsend ( request ) ĊĠĠĠĠĠĠĠ Ġ// ĠWait Ġuntil Ġthe Ġexpected Ġresponse Ġar rives Ġ( or Ġtimes Ġout ) ĊĠĠĠĠĠĠĠ Ġwith Timeout ( timeout In Millis ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Response Channel . receive () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠUn register Ġthe Ġchannel ĊĠĠĠĠĠĠĠ Ġexpected Responses . remove ( request . id ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ inline Ġfun ĠPeer . sendMessage ( cross inline Ġbuild Block : ĠRpc Event . Builder . () Ġ-> ĠUnit ) Ġ= Ġsend ( ĊĠĠĠ Ġbuild Message ( build Block Ġ= Ġbuild Block ) Ċ ) Ċ Ċ suspend Ġinline Ġfun ĠPeer . request Message ( ĊĠĠĠ Ġtimeout In Millis : ĠLong Ġ= Ġ 5 0 0 0 L , ĊĠĠĠ Ġcross inline Ġbuild Block : ĠRpc Event . Builder . () Ġ-> ĠUnit Ċ ) Ġ= Ġrequest Message ( ĊĠĠĠ Ġbuild Message ( build Block Ġ= Ġbuild Block ), ĊĠĠĠ Ġtimeout In Millis Ċ ) Ċ