package Ġcom . jetbrains . rd . framework Ċ Ċ import Ġcom . jetbrains . rd . framework . base . Wire Base Ċ import Ġcom . jetbrains . rd . util .* Ċ import Ġcom . jetbrains . rd . util . lifetime . Lifetime Ċ import Ġcom . jetbrains . rd . util . lifetime . is Alive Ċ import Ġcom . jetbrains . rd . util . lifetime . on Termination Ċ import Ġcom . jetbrains . rd . util . lifetime . plus Assign Ċ import Ġcom . jetbrains . rd . util . reactive .* Ċ import Ġcom . jetbrains . rd . util . threading . ByteBuffer Async Processor Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġjava . io . EOF Exception Ċ import Ġjava . io . IOException Ċ import Ġjava . io . InputStream Ċ import Ġjava . io . OutputStream Ċ import Ġjava . net .* Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . ExecutorService Ċ import Ġjava . util . concurrent . Executors Ċ import Ġkotlin . concurrent . thread Ċ Ċ private Ġfun ĠInputStream . read ByteArray ( a Ġ: ĠByteArray ): ĠBoolean Ġ{ ĊĠĠĠ Ġvar Ġpos Ġ= Ġ 0 ĊĠĠĠ Ġwhile Ġ( pos Ġ< Ġa . size ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġb Ġ= Ġread ( a , Ġpos , Ġa . size Ġ- Ġpos ) ĊĠĠĠĠĠĠĠ Ġif Ġ( b Ġ== Ġ- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġpos Ġ+= Ġb ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn Ġtrue Ċ } ĊĊ Ċ private Ġfun ĠInputStream . readInt 3 2 () Ġ: ĠInt ? Ġ{ ĊĠĠĠ Ġval Ġb 1 Ġ= Ġread (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 2 Ġ= Ġread (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 3 Ġ= Ġread (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 4 Ġ= Ġread (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġres Ġ= Ġb 1 Ġor Ġ( b 2 Ġshl Ġ 8 ) Ġor Ġ( b 3 Ġshl Ġ 1 6 ) Ġor Ġ( b 4 Ġshl Ġ 2 4 ) ĊĊĠĠĠ Ġreturn Ġres Ċ } Ċ Ċ private Ġfun ĠInputStream . readInt 6 4 () Ġ: ĠLong ? Ġ{ ĊĠĠĠ Ġval Ġb 1 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 2 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 3 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 4 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 5 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 6 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 7 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġb 8 Ġ= Ġread (). to Long (). apply Ġ{ Ġif Ġ( this Ġ< Ġ 0 ) Ġreturn Ġnull Ġ} ĊĠĠĠ Ġval Ġres Ġ= Ġb 1 Ġor Ġ( b 2 Ġshl Ġ 8 ) Ġor Ġ( b 3 Ġshl Ġ 1 6 ) Ġor Ġ( b 4 Ġshl Ġ 2 4 ) Ġor ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( b 5 Ġshl Ġ 3 2 ) Ġor Ġ( b 6 Ġshl Ġ 4 0 ) Ġor Ġ( b 7 Ġshl Ġ 4 8 ) Ġor Ġ( b 8 Ġshl Ġ 5 6 ) ĊĊĊĠĠĠ Ġreturn Ġres Ċ } ĊĊ Ċ class ĠSocket Wire Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtimeout : ĠDuration Ġ= ĠDuration . of Millis ( 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġack _ msg _ len : ĠInt Ġ= Ġ- 1 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġping _ len : ĠInt Ġ= Ġ- 2 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġpkg _ header _ len Ġ= Ġ 1 2 ĊĠĠĠĠĠĠĠ Ġconst Ġval Ġdisconnected Pause Reason Ġ= Ġ" Socket Ġnot Ġconnected " ĊĠĠĠĠĠĠĠ Ġconst Ġval Ġmaximum Heartbeat Delay Ġ= Ġ 3 ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġconnection Est ab lished ( time Stamp Ġ: ĠInt , Ġnot ion Timestamp : ĠInt ) Ġ= Ġtime Stamp Ġ- Ġnot ion Timestamp Ġ<= Ġmaximum Heartbeat Delay ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġabstract Ġclass ĠBase Ġprotected Ġconstructor ( val Ġid : ĠString , Ġprivate Ġval Ġlifetime : ĠL ifetime , Ġscheduler : ĠIS cheduler ) Ġ: ĠWire Base ( scheduler ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġprotected Ġval Ġlogger : ĠLogger Ġ= Ġget Logger ( this :: class ) ĊĠĠĠĠĠĠĠ Ġval Ġsocket Provider Ġ= ĠOpt Property < Socket >() ĊĊĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġoutput Ġ: ĠOutputStream ĊĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġsocket Input Ġ: ĠInputStream ĊĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġpkg Input Ġ: ĠInputStream ĊĊĠĠĠĠĠĠĠ Ġprotected Ġval Ġsend Buffer Ġ= ĠByteBuffer Async Processor ("$ id / Sender ", Ġprocessor Ġ= Ġ:: send 0 ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġthread Local Buffer Array Ġ= ĠThread Local . with Initial Ġ{ ĠUnsafe Buffer ( ByteArray ( 1 6 3 8 4 )) Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġack tor : ĠExecutor Service Ġ= ĠExec utors . new Single Thread Executor () ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġsend Ack ( seq n : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcatch And Drop Ġ{ Ġack tor . execute Ġ{ Ġsend Ack 0 ( seq n ) Ġ} Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠ Ġprotected Ġval Ġlock Ġ= ĠObject () ĊĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġmax Received Seq n Ġ: ĠLong Ġ= Ġ 0 ĊĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . pause ( disconnected Pause Reason ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . start () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnected . adv ise ( lifetime ) Ġ{ Ġheartbeat Alive . value Ġ= Ġit Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Provider . adv ise ( lifetime ) Ġ{ Ġsocket Ġ-> ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id Ġ: Ġconnected " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput Ġ= Ġsocket . output Stream ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Input Ġ= Ġsocket . inputStream . buffered () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpkg Input Ġ= ĠP kg InputStream ( socket Input ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . re process Un ack nowled ged () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . resume ( disconnected Pause Reason ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġheartbeat Job Ġ= Ġstart Heartbeat () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler . queue Ġ{ Ġconnected . value Ġ= Ġtrue Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceiver Proc ( socket ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler . queue Ġ{ Ġconnected . value Ġ= Ġfalse Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheartbeat Job . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . pause ( disconnected Pause Reason ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch And Drop Ġ{ Ġsocket . close () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġstart Heartbeat () Ġ= ĠGlobal Scope . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( heartbeat Interval Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġping () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġping () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcatch And Drop Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! connection Est ab lished ( currentTime Stamp , Ġcounter part Not ion Timestamp )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( heartbeat Alive . value ) Ġ{ Ġ// Ġonly Ġon Ġchange ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Disconnect Ġdetected Ġwhile Ġsending ĠP ING Ġ$ id : Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" currentTime Stamp : Ġ$ currentTime Stamp , Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" counter part Timestamp : Ġ$ counter part Timestamp , Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" counter part Not ion Timestamp : Ġ$ counter part Not ion Timestamp " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheartbeat Alive . value Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( socket Send Lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ping Pkg Header . reset () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ping Pkg Header . writeInt ( ping _ len ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ping Pkg Header . writeInt ( currentTime Stamp ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ping Pkg Header . writeInt ( counter part Timestamp ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput . write ( send Ping Pkg Header . get Array ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ++ currentTime Stamp ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġreceiver Proc ( socket : ĠSocket ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! socket . isConnected ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ" Stop Ġreceive Ġmessages Ġbecause Ġsocket Ġdisconnected " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! read Msg ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id : ĠConnection Ġwas Ġgrace fully Ġshutdown " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( ex ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSocket Exception , Ġis ĠEOF Exception , Ġis ĠIOException Ġ-> Ġlogger . debug Ġ{" Exception Ġin ĠSocket Wire . Receive : Ġ Ġ$ id : Ġ$ ex " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġlogger . error ("$ id Ġcaught Ġprocessing ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġread Msg () Ġ: ĠBoolean Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġseq n At Start Ġ= Ġmax Received Seq n ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlen Ġ= Ġpkg Input . readInt 3 2 () Ġ?: Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( len Ġ> Ġ 0 ) Ġ{" len Ġ> Ġ 0 : Ġ$ len "} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire Ġ( len Ġ< Ġ 3 0 0 _ 0 0 0 _ 0 0 0 ) Ġ{ Ġ" Possible ĠO OM : Ġarray _ len =$ len ( 0 x ${ len . toString ( 1 6 )}) " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdata Ġ= ĠByteArray ( len ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! pkg Input . read ByteArray ( data )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( max Received Seq n Ġ> Ġseq n At Start ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresponse Seq n Ġ= Ġmax Received Seq n ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ack ( response Seq n ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġunsafe Buffer Ġ= ĠUnsafe Buffer ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġid Ġ= ĠR d Id . read ( unsafe Buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Broker . dispatch ( id , Ġunsafe Buffer ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠ Ġinner Ġclass ĠP kg InputStream ( private Ġval Ġstream : ĠInputStream ) Ġ: ĠInputStream () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġpkg : ĠByteArray Ġ= ĠByteArray ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġpos : ĠInt Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġread (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( pos Ġ< Ġpkg . size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġpkg [ pos ++ ]. toInt () Ġand Ġ 0 xff ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlen Ġ= Ġstream . readInt 3 2 () Ġ?: Ġreturn Ġ- 1 ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( len Ġ== Ġping _ len ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceived Timestamp Ġ= Ġstream . readInt 3 2 () Ġ?: Ġreturn Ġ- 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceived Counter part Timestamp Ġ= Ġstream . readInt 3 2 () Ġ?: Ġreturn Ġ- 1 ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcounter part Timestamp Ġ= Ġreceived Timestamp ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcounter part Not ion Timestamp Ġ= Ġreceived Counter part Timestamp ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( connection Est ab lished ( currentTime Stamp , Ġcounter part Not ion Timestamp )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! heartbeat Alive . value ) Ġ{ Ġ// Ġonly Ġon Ġchange ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Connection Ġis Ġalive Ġafter Ġreceiving ĠP ING Ġ$ id : Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" received Timestamp : Ġ$ received Timestamp , Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" received Counter part Timestamp : Ġ$ received Counter part Timestamp " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" currentTime Stamp : Ġ$ currentTime Stamp , Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" counter part Timestamp : Ġ$ counter part Timestamp , Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" counter part Not ion Timestamp : Ġ$ counter part Not ion Timestamp " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheartbeat Alive . value Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġseq n Ġ= Ġstream . readInt 6 4 () Ġ?: Ġreturn Ġ- 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( len Ġ== Ġack _ msg _ len ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . ack nowledge ( seq n ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( len Ġ> Ġ 0 ) Ġ{" len Ġ> Ġ 0 : Ġ$ len "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire Ġ( len Ġ< Ġ 3 0 0 _ 0 0 0 _ 0 0 0 ) Ġ{ Ġ" Possible ĠO OM : Ġarray _ len =$ len ( 0 x ${ len . toString ( 1 6 )}) " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpkg Ġ= ĠByteArray ( len ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpos Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream . read ByteArray ( pkg ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( seq n Ġ> Ġmax Received Seq n ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmax Received Seq n Ġ= Ġseq n ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġpkg [ pos ++ ]. toInt () Ġand Ġ 0 xff ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ack ( seq n ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġsend Ack 0 ( seq n : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġack Pkg Header . reset () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġack Pkg Header . writeInt ( ack _ msg _ len ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġack Pkg Header . write Long ( seq n ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( socket Send Lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput . write ( ack Pkg Header . get Array (), Ġ 0 , Ġpkg _ header _ len ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠSocket Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn Ġ{ Ġ"$ id : ĠException Ġraised Ġduring ĠACK , Ġseq n Ġ= Ġ$ seq n " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġsent Seq n Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsocket Send Lock Ġ= ĠAny () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsend Pkg Header Ġ= Ġcreate Abstract Buffer () ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġack Pkg Header Ġ= Ġcreate Abstract Buffer () ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠTimestamp Ġof Ġthis Ġwire Ġwhich Ġincreases Ġat Ġintervals Ġof Ġ[ heartbeat Interval ]. ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġcurrent TimeStamp Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠActual Ġnot ion Ġabout Ġcounter part 's Ġ[ currentTime Stamp ]. ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġcounter part Timestamp Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġlatest Ġreceived Ġcounter part 's Ġnot ion Ġof Ġthis Ġwire 's Ġ[ currentTime Stamp ]. ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġcounter part Not ion Timestamp Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsend Ping Pkg Header Ġ= Ġcreate Abstract Buffer () ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġsend 0 ( chunk : ĠByteBuffer Async Processor . Chunk ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( chunk . isNot Processed ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchunk . seq n Ġ= Ġ++ sent Seq n ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Pkg Header . reset () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Pkg Header . writeInt ( chunk . ptr ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Pkg Header . write Long ( chunk . seq n ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( socket Send Lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ Ġ" Send Ġpackage Ġwith Ġseq n Ġ${ chunk . seq n }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput . write ( send Pkg Header . get Array (), Ġ 0 , Ġpkg _ header _ len ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput . write ( chunk . data , Ġ 0 , Ġchunk . ptr ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠSocket Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . pause ( disconnected Pause Reason ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsend ( id : ĠR d Id , Ġwriter : Ġ( Abstract Buffer ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire (! id . isNull ) Ġ{ Ġ" id Ġmust n 't Ġbe Ġnull " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġunsafe Buffer Ġ= Ġthread Local Buffer Array . get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinitial Position Ġ= Ġunsafe Buffer . position ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Buffer . writeInt ( 0 ) Ġ// placeholder Ġfor Ġlength ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġid . write ( unsafe Buffer ) Ġ// write Ġid ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontexts . write Current Message Context ( unsafe Buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwriter ( unsafe Buffer ) Ġ// write Ġrest ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlen Ġ= Ġunsafe Buffer . position Ġ- Ġinitial Position ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Buffer . position Ġ= Ġinitial Position ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Buffer . writeInt ( len Ġ- Ġ 4 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbytes Ġ= Ġunsafe Buffer . get Array () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Buffer . put ( bytes , Ġinitial Position , Ġlen ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( initial Position Ġ== Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Buffer . reset () Ġ// Ġapply Ġshrink ing Ġlogic ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Buffer . position Ġ= Ġinitial Position ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġclass ĠClient ( lifetime Ġ: ĠL ifetime , Ġscheduler : ĠIS cheduler , Ġport Ġ: ĠInt , Ġopt Id : ĠString ? Ġ= Ġnull , Ġhost Address : ĠInet Address Ġ= ĠInet Address . get Loop back Address ()) Ġ: ĠBase ( opt Id Ġ? :" Client Socket ", Ġlifetime , Ġscheduler ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsocket Ġ: ĠSocket ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġthread Ġ= Ġthread ( name Ġ= Ġid , Ġis Daemon Ġ= Ġtrue ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġs Ġ= ĠSocket () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . tcp No Delay Ġ= Ġtrue ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠOn Ġwindows Ġconnect Ġwill Ġtry Ġto Ġsend ĠSYN Ġ 3 Ġtimes Ġwith Ġinterval Ġof Ġ 5 0 0 ms Ġ( total Ġtime Ġis Ġ 1 second ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠConnect Ġtimeout Ġdoesn 't Ġwork Ġif Ġit 's Ġmore Ġthan Ġ 1 Ġsecond . ĠBut Ġwe Ġdon 't Ġneed Ġit Ġbecause Ġwe Ġcan Ġclose Ġsocket Ġany Ġmoment . ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// https :// stackoverflow . com / questions / 2 2 4 1 7 2 2 8 / prevent - tcp - socket - connection - retries ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// HKLM \ SYSTEM \ Current Control Set \ Services \ Tcp ip \ Parameters \ Tcp Max Connect Re trans missions ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id Ġ: Ġconnecting Ġto Ġ$ host Address :$ port " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . connect ( Inet SocketAddress ( host Address , Ġport )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id Ġ: Ġconnected , Ġbut Ġlifetime Ġis Ġalready Ġcanceled , Ġclosing Ġsocket "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch And Drop Ġ{ s . close ()} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ thread ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Ġ= Ġs ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Provider . set ( s ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id : Ġreceiver Proc Ġfinished Ġ" Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠConnect Exception ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġshould Re connect Ġ= Ġsynchronized ( lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlock . wait ( timeout . to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlifetime . is Alive ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( should Re connect ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠSocket Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{" $ id : Ġclosed Ġwith Ġexception : Ġ$ ex "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlifetime Ġ+= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{" $ id : Ġstart Ġterminating Ġlifetime "} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{" $ id : Ġshut ting Ġdown Ġack Ġsending Ġexecutor "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġack tor . shutdown () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsend Buffer Stopped Ġ= Ġsend Buffer . stop ( timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug {" $ id : Ġsend Ġbuffer Ġstopped , Ġsuccess : Ġ$ send Buffer Stopped "} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug {" $ id : Ġclosing Ġsocket "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ{ socket ?. close ()} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlock . notify All () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id : Ġwaiting Ġfor Ġreceiver Ġthread " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ{ Ġthread . join ( timeout . to Millis ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ"$ id : Ġtermination Ġfinished " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġclass ĠServer Ġinternal Ġconstructor ( lifetime Ġ: ĠL ifetime , Ġscheduler : ĠIS cheduler , Ġss : ĠServer Socket , Ġopt Id : ĠString ? Ġ= Ġnull , Ġallow Re connect : ĠBoolean ) Ġ: ĠBase ( opt Id Ġ? :" Server Socket ", Ġlifetime , Ġscheduler ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġport Ġ: ĠInt Ġ= Ġss . local Port ĊĊĠĠĠĠĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinternal Ġfun Ġcreate Server Socket ( lifetime : ĠL ifetime , Ġport Ġ: ĠInt ?, Ġallow Remote Connections : ĠBoolean ) Ġ: ĠServer Socket Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaddress Ġ= Ġif Ġ( allow Remote Connections ) Ġnull Ġelse ĠInet Address . get ByName (" 1 2 7 . 0 . 0 . 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġres Ġ= ĠServer Socket ( port ?: 0 , Ġ 0 , Ġaddress ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlifetime . on Termination Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġres . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġres ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġconstructor Ġ( lifetime Ġ: ĠL ifetime , Ġscheduler : ĠIS cheduler , Ġport Ġ: ĠInt ?, Ġopt Id : ĠString ? Ġ= Ġnull , Ġallow Remote Connections : ĠBoolean Ġ= Ġfalse ) Ġ: Ġthis ( lifetime , Ġscheduler , Ġcreate Server Socket ( lifetime , Ġport , Ġallow Remote Connections ), Ġopt Id , Ġallow Re connect Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsocket Ġ: ĠSocket ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġthread Ġ= Ġthread ( name Ġ= Ġid , Ġis Daemon Ġ= Ġtrue ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id : Ġlistening Ġ${ ss . local SocketAddress }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġs Ġ= Ġss . accept () Ġ// could Ġbe Ġterminated Ġby Ġclose ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . tcp No Delay Ġ= Ġtrue ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! lifetime . is Alive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id Ġ: Ġconnected , Ġbut Ġlifetime Ġis Ġalready Ġcanceled , Ġclosing Ġsocket " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ{ Ġs . close () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ thread ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Ġ= Ġs ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket Provider . set ( s ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠSocket Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"$ id Ġclosed Ġwith Ġexception : Ġ$ ex " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ("$ id Ġclosed Ġwith Ġexception ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! allow Re connect ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlifetime . on Termination If Alive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{" $ id : Ġstart Ġterminating Ġlifetime " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{" $ id : Ġshut ting Ġdown Ġack Ġsending Ġexecutor "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġack tor . shutdown () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsend Buffer Stopped Ġ= Ġsend Buffer . stop ( timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{" $ id : Ġsend Ġbuffer Ġstopped , Ġsuccess : Ġ$ send Buffer Stopped "} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{" $ id : Ġclosing Ġsocket "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket ?. close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ{ Ġthread . join ( timeout . to Millis ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info {" $ id : Ġtermination Ġfinished "} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĊĠĠĠ Ġdata Ġclass ĠWire Parameters ( val Ġscheduler : ĠIS cheduler , Ġval Ġid : ĠString ?) ĊĠĠĠ Ġclass ĠServer Factory Ġprivate Ġconstructor ( lifetime Ġ: ĠL ifetime , Ġwire Parameters Factory : Ġ() Ġ-> ĠWire Parameters , Ġport Ġ: ĠInt ?, Ġallow Remote Connections : ĠBoolean , Ġset : ĠView able Set < Server >) Ġ: ĠI View able Set < Server > Ġby Ġset Ġ{ ĊĊĠĠĠĠĠĠĠ Ġconstructor ( lifetime Ġ: ĠL ifetime , Ġwire Parameters Factory : Ġ() Ġ-> ĠWire Parameters , Ġport Ġ: ĠInt ?, Ġallow Remote Connections : ĠBoolean Ġ= Ġfalse ) Ġ: ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis ( lifetime , Ġwire Parameters Factory , Ġport , Ġallow Remote Connections , ĠView able Set < Server >()) ĊĊĠĠĠĠĠĠĠ Ġconstructor ( lifetime Ġ: ĠL ifetime , Ġscheduler : ĠIS cheduler , Ġport Ġ: ĠInt ?, Ġallow Remote Connections : ĠBoolean Ġ= Ġfalse ) Ġ: ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis ( lifetime , Ġ{ ĠWire Parameters ( scheduler , Ġnull ) Ġ}, Ġport , Ġallow Remote Connections , ĠView able Set < Server >()) ĊĊĊĠĠĠĠĠĠĠ Ġval Ġlocal Port : ĠInt ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġss Ġ= ĠServer . create Server Socket ( lifetime , Ġport , Ġallow Remote Connections ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlocal Port Ġ= Ġss . local Port ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġfun Ġrec () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlifetime . execute If Alive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( scheduler , Ġopt Id ) Ġ= Ġwire Parameters Factory () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġs Ġ= ĠServer ( lifetime , Ġscheduler , Ġss , Ġopt Id , Ġallow Re connect Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . connected . when True ( lifetime ) Ġ{ Ġlt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset . add Unique ( lt , Ġs ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrec () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġrec () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ} Ċ Ċ } ĊĊ Ċ // todo Ġremove Ċ val ĠI Wire . server Port : ĠInt Ġget () Ġ{ ĊĠĠĠ Ġval Ġserver Socket Wire Ġ= Ġthis Ġas ? ĠSocket Wire . Server Ġ?: Ġthrow ĠIllegalArgumentException (" You Ġmust Ġuse ĠSocket Wire . Server Ġto Ġget Ġserver Ġport ") ĊĠĠĠ Ġreturn Ġserver Socket Wire . port Ċ } Ċ