package Ġnet . w ush il in . port for warder . udp Ċ Ċ import Ġnet . w ush il in . port for warder . common .* Ċ import Ġjava . net . SocketAddress Ċ import Ġjava . nio . ByteBuffer Ċ import Ġjava . nio . channels .* Ċ import Ġjava . time . Duration Ċ import Ġkotlin . system . exit Process Ċ Ċ // ĠStores Ġthe Ġincoming Ġsocket Ġand Ġremote Ġtargets , Ġfor Ġserver Ġsetup ! Ċ val Ġtarget Map Ġ= Ġmutable Map Of < Data gram Channel , ĠPair < String , ĠInt >>() Ċ Ċ // ĠStores Ġthe ĠU D PP ipe Ġto Ġavoid Ġmemory Ġallocation Ċ val Ġpipe Pool Ġ= ĠPool ( 1 , ĊĠĠĠ Ġsupplier Ġ= Ġ{ ĠU D PP ipe ( null , Ġnull , Ġnull , Ġnull , Ġnull ) Ġ}, ĊĠĠĠ Ġreset ter Ġ= Ġnull , ĊĠĠĠ Ġgarbage Collector Ġ= Ġnull ) Ċ Ċ // ĠStores Ġall Ġcurrent Ġbi - direction al Ġpipe Ġmapping Ġpairs Ċ // Ġfrom Ġclient address Ġ-> ĠU D PP ipe Ċ // Ġfrom Ġlocal address Ġ-> ĠUDP ipe Ċ late init Ġvar Ġpipes : ĠL RU Cache < SocketAddress , ĠU D PP ipe > Ċ Ċ var Ġchannel Pool Ġ= ĠPool ( 1 , Ġsupplier Ġ= Ġ{ new Channel ()}, Ġreset ter ={ Ġre cycle ( it ) Ġ}, ĊĠĠĠ Ġgarbage Collector Ġ= Ġ{ ĊĠĠĠĠĠĠĠ Ġdestroy ( it ) ĊĠĠĠ Ġ}) Ċ // ĠLink Ġu ptime , Ġby Ġusing Ġlocal Client Ġaddress Ġ-> Ġremote Ġserver . Ċ val Ġlink Up Ts Ġ= Ġmutable Map Of < SocketAddress , ĠLong >() Ċ Ċ // ĠRemember Ġstats Ġfor Ġthe Ġconnections Ċ val Ġstats Ġ= Ġmutable Map Of < SocketAddress , ĠLong >() Ċ Ċ // ĠTotal Ġnumber Ġof Ġbytes Ġtransferred Ċ var Ġtotal Bytes Ġ= Ġ 0 L Ċ Ċ // ĠNumber Ġof Ġtotal Ġrequests Ċ var Ġtotal Requests Ġ= Ġ 0 L Ċ Ċ // ĠNumber Ġof Ġactive Ġrequests Ċ var Ġactive Requests Ġ= Ġ 0 L Ċ Ċ // ĠReports Ġstats Ġevery Ġ 3 0 Ġseconds Ċ var Ġreport Interval Ġ= Ġ 3 0 0 0 0 L Ċ Ċ // ĠGlobal Ġbuffer Ġsize . ĠIt Ġis Ġused Ġfor Ġall Ġsockets ! Ċ var ĠbufferSize Ġ= Ġ 1 0 2 4 Ġ* Ġ 1 0 2 4 Ċ Ċ // Ġtime Ġof Ġlast Ġstats Ġreporting Ċ var Ġlast Report Ġ= Ġ 0 L Ċ Ċ // Ġtime Ġwhen Ġthis Ġprogram Ġwas Ġstarted Ċ var Ġstart TS Ġ= ĠSystem . currentTimeMillis () Ċ Ċ // ĠSet Ġof Ġlocal Ġlistener Ġaddresses Ċ var Ġlocal Listeners Ġ= Ġmutable Set Of < SocketAddress >() Ċ Ċ // ĠSet Ġtimeout Ġfor Ġidle Ċ var Ġidle Timeout Ġ= Ġ 3 6 0 0 0 0 0 L Ċ Ċ // ĠMax Ġconn Ġtracking Ċ var Ġcache Size Ġ= Ġ 0 Ċ Ċ // ĠCheck Ġe viction Ċ var Ġe viction Check Interval Ġ= Ġ 3 0 0 0 0 L Ċ Ċ fun Ġnew Channel (): Data gram Channel Ġ{ ĊĠĠĠ Ġval Ġnew Client : ĠData gram Channel Ġ= ĠData gram Channel . open () ĊĠĠĠ Ġ// Ġchoose Ġrandom Ġport ĊĠĠĠ Ġnew Client . bind ( null ) ĊĠĠĠ Ġnew Client . configure Blocking ( false ) ĊĠĠĠ Ġreturn Ġnew Client Ċ } Ċ Ċ fun Ġre cycle ( channel : Data gram Channel ) Ġ{ ĊĠĠĠ Ġif ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Disconnect ing Ġchannel Ġfor Ġreuse : Ġ${ channel . local Address }") ĊĠĠĠ Ġ} ĊĠĠĠ Ġchannel . disconnect () Ċ } Ċ Ċ fun Ġdestroy ( channel : Data gram Channel ) Ġ{ ĊĠĠĠ Ġif ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Closing Ġchannel Ġfor Ġgood : Ġ${ channel . local Address }") ĊĠĠĠ Ġ} ĊĠĠĠ Ġchannel . close () Ċ } Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠ ĠConfig . setUp () ĊĠĠĠ Ġif Ġ( args . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln (" Usage : Ġjava Ġ- jar Ġport for warder . jar Ġ< local bind >:: < remote _ target >") ĊĠĠĠĠĠĠĠ Ġprintln (" Where :") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ Ġ< local bind > ĠĠĠĠ Ġ= Ġhost : port Ġ( e . g . Ġ' localhost : 4 4 3 ' Ġor Ġ' 0 . 0 . 0 . 0 : 9 0 9 0 ' Ġwithout Ġthe Ġquotes ! )") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ Ġ< remote _ target > Ġ= Ġhost : port Ġ( e . g . Ġ' www . google . com : 4 4 3 ' Ġor Ġ' 1 7 2 . 2 0 . 1 1 . 4 2 : 9 0 9 2 ' Ġwithout Ġthe Ġquotes !") ĊĠĠĠĠĠĠĠ Ġprintln (" For Ġexample : Ġ") ĊĠĠĠĠĠĠĠ Ġprintln (" Build Ġthis Ġwith :") ĊĠĠĠĠĠĠĠ Ġprintln ("# Ġgradle Ġjar ") ĊĠĠĠĠĠĠĠ Ġprintln ("# Ġjava Ġ- jar Ġport - for warder - 1 . 0 . jar Ġ 1 2 7 . 0 . 0 . 1 : 2 2 :: ZM 0 9 . my company . com : 2 2 ") ĊĠĠĠĠĠĠĠ Ġprintln (" This Ġwill Ġallow Ġyou Ġto Ġssh , Ġor Ġsc p Ġto Ġremote Ġserver Ġusing Ġssh Ġuser @ localhost ") ĊĠĠĠĠĠĠĠ Ġprintln (" This Ġprogram Ġuses Ġasync ĠIO , Ġit Ġis Ġefficient , Ġbut Ġsingle Ġth readed .") ĊĠĠĠĠĠĠĠ Ġprintln (" Other Ġoptions :") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- D enable . timestamp . in . log = false Ġ- Ġdisables Ġlog Ġwith Ġtimestamp Ġ( default Ġtrue )") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- D buffer . size = 5 1 2 0 0 0 Ġ- Ġset Ġthe Ġglobal Ġbuffer Ġsize Ġin Ġbytes Ġ( default Ġ 1 M iB )") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- D log . level = 0 | 1 | 2 | 3 | 4 | 5 | 6 Ġ- Ġ 0 = debug +, Ġ 1 = info +, Ġ 2 = warn +, Ġ 3 = error +, Ġ 4 = nothing Ġ Ġ( default Ġ 1 )") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- D conn . track . max = 1 0 0 0 0 ĠThe ĠUDP Ġsource - destination Ġwill Ġbe Ġtracked Ġfor Ġup Ġto Ġthis Ġnumber . ĠL RU Ġwill Ġbe Ġev ict ed ") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- D stats . interval = 3 0 0 0 0 Ġ- Ġmin Ġduration Ġin Ġmilliseconds Ġbetween Ġstats Ġreporting Ġ( default Ġ 3 0 0 0 0 Ġ= Ġ 3 0 Ġseconds )") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- Did le . timeout = 3 6 0 0 0 0 0 Ġ- Ġis Ġno Ġtraffic Ġfor Ġthis Ġpipe Ġfor Ġmore Ġthan Ġthis , Ġit Ġwill Ġbe Ġclosed .") ĊĠĠĠĠĠĠĠ Ġprintln (" Ġ- Did le . check . interval = 5 0 0 0 Ġ- Ġthe Ġidle Ġtimeout Ġis Ġonly Ġchecked Ġevery Ġxxx Ġmilliseconds ") ĊĠĠĠĠĠĠĠ Ġexit Process ( 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ ĠLog . ENABLE _ TS _ IN _ LOG Ġ= ĠConfig . get (" enable . timestamp . in . log ", Ġtrue ) Ġ{ Ġit Ġ-> Ġit . to Boolean () Ġ} ĊĠĠĠ Ġidle Timeout Ġ= ĠConfig . get (" idle . timeout ", Ġ 3 6 0 0 0 0 0 L ) Ġ{ Ġit Ġ-> Ġit . to Long () Ġ} ĊĠĠĠ Ġe viction Check Interval Ġ= ĠConfig . get (" idle . check . interval ", Ġ 5 0 0 0 ) Ġ{ Ġit . to Long () Ġ} ĊĠĠĠ ĠbufferSize Ġ= ĠConfig . get (" buffer . size ", Ġ 1 0 4 8 5 7 6 ) Ġ{ Ġit Ġ-> Ġit . toInt () Ġ} ĊĠĠĠ Ġif Ġ( buffer Size Ġ< Ġ 1 0 0 0 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( Log . isError Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . error (" Require Ġbuffer . size Ġ>= Ġ 1 0 0 0 0 0 Ġ( max ĠUDP Ġis Ġ 6 5 5 3 5 ) Ġor Ġdata Ġmight Ġbe Ġlost .") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexit Process ( 1 ) ĊĠĠĠ Ġ} ĊĠĠĠ ĠLog . LOG _ LEVEL Ġ= ĠConfig . get (" log . level ", Ġ 1 ) Ġ{ Ġit Ġ-> Ġit . toInt () Ġ} ĊĠĠĠ Ġreport Interval Ġ= ĠConfig . get (" stats . interval ", Ġ 3 0 0 0 0 ) Ġ{ Ġit . to Long () Ġ} ĊĠĠĠ Ġcache Size Ġ= ĠConfig . get (" conn . track . max ", Ġ 1 0 0 0 0 ) Ġ{ Ġit . toInt () Ġ} ĊĠĠĠ Ġpipes Ġ= ĠL RU Cache ( cache Size Ġ* Ġ 2 ) ĊĊĠĠĠ Ġchannel Pool . resize ( cache Size ) ĊĠĠĠ Ġpipe Pool . resize ( cache Size ) ĊĠĠĠ Ġval Ġselector : ĠSelector Ġ= ĠSelector . open () ĊĠĠĠ Ġval Ġargs Parsed Ġ= ĠHost Utils . parse ( args ) ĊĠĠĠ Ġfor Ġ( next Arg Ġin Ġargs Parsed . entries ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġlisten Address Ġ= Ġnext Arg . key ĊĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= Ġnext Arg . value ĊĠĠĠĠĠĠĠ Ġval Ġserver Socket Ġ= ĠData gram Channel . open () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserver Socket . bind ( listen Address ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserver Socket . configure Blocking ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserver Socket . register ( selector , ĠSelection Key . OP _ READ ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlocal Listeners . add ( listen Address ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . isError Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . error (" Invalid Ġbinding Ġoption Ġ$ listen Address Ġ($ ex )") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexit Process ( 1 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Bound Ġto Ġ$ listen Address , Ġforwarding Ġto Ġ$ target ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġtarget Map [ server Socket ] Ġ= Ġtarget ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠ ĠLog . info (" Alloc ating Ġglobal Ġbuffer Ġof Ġ${ bytes ToString ( buffer Size . to Long ()) } Ġbytes ...") ĊĠĠĠ Ġ} ĊĠĠĠ Ġval Ġbuffer Ġ= ĠByteBuffer . allocate ( buffer Size ) ĊĊĊĠĠĠ ĠLog . print Logo () ĊĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠ ĠLog . info (" Server Ġstarted , Ġstats Ġwill Ġbe Ġreports Ġevery Ġ$ report Interval Ġmilliseconds ...") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġvar Ġlast Check Ġ= Ġ 0 L ĊĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġselect Count Ġ= Ġselector . select ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠ Ġif Ġ( select Count Ġ> Ġ 0 Ġ&& ĠLog . isDebugEnabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . debug ("$ select Count Ġkey ( s ) Ġselected .") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġnow Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġif Ġ( now Ġ- Ġlast Report Ġ> Ġreport Interval ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġu ptime Ġ= ĠDuration . of Millis ( System . currentTimeMillis () Ġ- Ġstart TS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Status ĠUpdate : ĠU ptime Ġ${ uptime }, Ġ${ selector . keys (). size } Ġkeys , Ġ$ active Requests Ġactive Ġrequests , Ġ$ total Requests Ġtotal Ġrequests , Ġ${ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbytes ToString ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Bytes ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġtransferred , ĠJVM ĠFREE =${ bytes ToString ( Runtime . get Runtime (). free Memory ()) }/ TOTAL =${ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbytes ToString ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRuntime . get Runtime (). total Memory () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĠUDP | PIPE Ġhit Rate (${ channel Pool . hit Rate ()} | ${ pipe Pool . hit Rate () })" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Object ĠCount : ĠP ip es (${ pipes . size () }) ĠStats (${ stats . size }) ĠLink Up (${ link Up Ts . size }) ĠUDP Channel (${ channel Pool . size () }) ĠPool (${ pipe Pool . size () }) ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Report Ġ= Ġnow ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġselected Keys Ġ= Ġselector . selected Keys () ĊĠĠĠĠĠĠĠ Ġval Ġiter Ġ= Ġselected Keys . iterator () ĊĠĠĠĠĠĠĠ Ġwhile Ġ( iter . hasNext ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġiter . next () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( key . isValid Ġ&& Ġkey . is Readable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġkey . channel () Ġas ĠData gram Channel ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( channel . isOpen ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandle Read ( channel , Ġkey , Ġbuffer , Ġselector ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġiter . remove () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġelapsed Ġ= Ġnow Ġ- Ġlast Check ĊĠĠĠĠĠĠĠ Ġif Ġ( elapsed Ġ>= Ġe viction Check Interval ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Check Ġ= Ġnow ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwater mark Ġ= Ġnow Ġ- Ġidle Timeout ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġev ict ed Ġ= Ġpipes . ev ict Before ( watermark ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ev ict ed . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Ev ict ed Ġ${ ev ict ed . size } Ġconnections Ġdue Ġto Ġidle Ġtimeout Ġ($ idle Timeout Ġms ). ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġev ict ed . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcleanup ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ @ Suppress (" UNUSED _ PARAMETER ") Ċ fun Ġhandle Read ( channel : ĠData gram Channel , Ġkey : ĠSelection Key , Ġbuffer : ĠByteBuffer , Ġselector : ĠSelector ) Ġ{ ĊĠĠĠ Ġbuffer . clear () ĊĠĠĠ Ġval Ġremote Address : ĠSocket Address ? ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġremote Address Ġ= Ġchannel . receive ( buffer ) ĊĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( Log . is Warn Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . warn (" Failed Ġto Ġreceive Ġfrom Ġlocal Ġ${ channel . local Address }: Ġ$ ex ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĠĠĠ Ġval Ġlocal Address Ġ= Ġchannel . local Address ĊĠĠĠ Ġbuffer . flip () ĊĠĠĠ Ġval Ġread Count Ġ= Ġbuffer . remaining () ĊĠĠĠ Ġval Ġis Listener Ġ= Ġlocal Listeners . contains ( local Address ) ĊĠĠĠ Ġvar Ġdest Address : ĠSocket Address ? ĊĠĠĠ Ġif Ġ( is Listener ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtarget Pair Ġ= Ġtarget Map [ channel ] !! ĊĠĠĠĠĠĠĠ Ġdest Address Ġ= ĠHost Utils . convertTo Address ( target Pair ) ĊĠĠĠĠĠĠĠ Ġvar Ġevent Pipe : ĠU D PP ipe ? ĊĠĠĠĠĠĠĠ Ġ// Ġclient Ġis Ġready Ġto Ġsend Ġto Ġme Ġand ĠI Ġdon 't Ġhave Ġthe Ġsession Ġup Ġyet ! ĊĠĠĠĠĠĠĠ Ġif Ġ( pipes . get ( remote Address ) Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Client Ġ= Ġchannel Pool . acquire () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnew Client . connect ( dest Address ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġselection Key Ġ= Ġnew Client . register ( selector , ĠSelection Key . OP _ READ ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Pipe Ġ= Ġpipe Pool . acquire () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Pipe . reuse ( remote Address , Ġchannel , Ġnew Client , Ġdest Address , Ġselection Key ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġev ict ed 1 Ġ= Ġpipes . put ( event Pipe . remote Client Address () !! , Ġevent Pipe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġev ict ed 2 Ġ= Ġpipes . put ( event Pipe . local Client Address () !! , Ġevent Pipe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġev ict edList Ġ= ĠlistOf NotNull ( ev ict ed 1 , Ġev ict ed 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ev ict edList . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Ev ict ed Ġ${ ev ict edList . size } Ġconnections Ġdue Ġto Ġmax Ġconn Ġtracking Ġ($ cache Size )") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġev ict edList . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcleanup ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactive Requests ++ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Requests ++ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Setting Ġup Ġnew Ġclient Ġfor Ġ${ remote Address } Ġto Ġ${ dest Address }, Ġvia Ġ${ new Client . local Address }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlink Up Ts [ event Pipe . local Client Address () !! ] Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( buffer . has Remaining ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew Client . write ( buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Warn Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . warn (" Failed Ġto Ġwrite Ġto Ġ$ dest Address , Ġdata Ġmight Ġbe Ġlost : Ġ$ ex ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthe Ġchannel Ġhad Ġbeen Ġsetup ! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Pipe Ġ= Ġpipes . get ( remote Address ) !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( buffer . has Remaining ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Pipe . local Client () !!. write ( buffer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Warn Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . warn (" Failed Ġto Ġwrite Ġto Ġ$ dest Address , Ġdata Ġmight Ġbe Ġlost : Ġ$ ex ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġadd Stats ( remote Address , Ġread Count ) ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġserver 's Ġresponse ĊĠĠĠĠĠĠĠ Ġval Ġevent Pipe Ġ= Ġpipes . get ( local Address ) ĊĠĠĠĠĠĠĠ Ġif Ġ( event Pipe Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Log . is Warn Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog . warn (" Unable Ġto Ġwrite Ġ$ read Count Ġbytes Ġfrom Ġ$ remote Address Ġback Ġto Ġclient . ĠClient Ġmight Ġbe Ġev ict ed .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġdest Address Ġ= Ġevent Pipe . remote Client Address () ĊĠĠĠĠĠĠĠ Ġwhile Ġ( buffer . has Remaining ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Pipe . local Listen ()?. send ( buffer , Ġevent Pipe . remote Client Address ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġadd Stats ( event Pipe . local Client Address () !! , Ġread Count ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġif Ġ( Log . isDebugEnabled ()) Ġ{ ĊĠĠĠĠĠĠĠ ĠLog . debug (" Copied Ġ${ read Count } Ġbytes Ġfrom Ġ$ remote Address Ġto Ġ$ dest Address ") ĊĠĠĠ Ġ} ĊĠĠĠ Ġtotal Bytes Ġ+= Ġread Count Ċ } Ċ Ċ fun Ġadd Stats ( from : ĠSocket Address , Ġdiff : ĠInt ) Ġ{ ĊĠĠĠ Ġval Ġcurrent Ġ= Ġstats [ from ] ĊĠĠĠ Ġif Ġ( current Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġstats [ from ] Ġ= Ġdiff . to Long () ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġstats [ from ] Ġ= Ġcurrent Ġ+ Ġdiff . to Long () ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun Ġcleanup ( session : ĠU D PP ipe ?) Ġ{ ĊĠĠĠ Ġif Ġ( session Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĠĠĠ Ġif Ġ( session . is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( Log . is Info Enabled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Pipe Ġclosed Ġfor Ġ${ session . remote Client Address ()} Ġ< == Ġ${ session . local Listen Address ()} Ġ=== Ġ${ session . local Client Address ()} Ġ==> Ġ${ session . target Address () }") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Ġ Ġ>> ĠTransfer Ġstats : Ġ${ session . remote Client Address ()} Ġ=> Ġ${ session . target Address ()} : Ġ${ bytes ToString ( stats [ session . remote Client Address ()]) }") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Ġ Ġ>> ĠTransfer Ġstats : Ġ${ session . remote Client Address ()} Ġ<= Ġ${ session . target Address ()} : Ġ${ bytes ToString ( stats [ session . local Client Address ()]) }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnow Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġup Ġ= Ġlink Up Ts [ session . local Client Address ()] Ġ?: Ġnow ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġduration Ġ= ĠDuration . of Millis ( now Ġ- Ġup ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . info (" Ġ Ġ>> ĠLink Ġup : Ġ$ duration ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠlistOf NotNull ( session . remote Client Address (), Ġsession . local Client Address ()). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpipes . remove ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstats . remove ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlink Up Ts . remove ( it ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsession . key ()?. cancel () ĊĠĠĠĠĠĠĠ Ġactive Requests -- ĊĠĠĠĠĠĠĠ Ġsession . close () ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠ Ġchannel Pool . release ( session . local Client () !! ) ĊĠĠĠĠĠĠĠ Ġ// Ġrelease Ġwill Ġreset Ġthe Ġsession Ġas Ġwell Ġusing Ġpool Ġconfig . ĊĠĠĠĠĠĠĠ Ġpipe Pool . release ( session ) ĊĠĠĠ Ġ} Ċ } ĊĊ