/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 Ġd ork box , Ġl lc Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġd ork box . network . connection Ċ Ċ import Ġd ork box . network . Client Ċ import Ġd ork box . network . Configuration Ċ import Ġd ork box . network . Server Ċ import Ġd ork box . network . Server Configuration Ċ import Ġd ork box . network . aer on . A eron Driver Ċ import Ġd ork box . network . aer on . Coroutine Idle Strategy Ċ import Ġd ork box . network . coroutines . Suspend Wait er Ċ import Ġd ork box . network . exceptions . ClientException Ċ import Ġd ork box . network . exceptions . Server Exception Ċ import Ġd ork box . network . handshake . Handshake Message Ċ import Ġd ork box . network . ip Filter . Ip Filter Rule Ċ import Ġd ork box . network . ping . Ping Ċ import Ġd ork box . network . ping . Ping Manager Ċ import Ġd ork box . network . rmi . Response Manager Ċ import Ġd ork box . network . rmi . R mi Manager Connections Ċ import Ġd ork box . network . rmi . R mi Manager Global Ċ import Ġd ork box . network . rmi . messages . Method Response Ċ import Ġd ork box . network . rmi . messages . R mi Message Ċ import Ġd ork box . network . serialization . K ry o Extra Ċ import Ġd ork box . network . serialization . Serialization Ċ import Ġd ork box . network . serialization . Settings Store Ċ import Ġd ork box . util . exceptions . Security Exception Ċ import Ġio . aer on . Publication Ċ import Ġio . aer on . driver . Media Driver Ċ import Ġio . aer on . log buffer . Header Ċ import Ġkotlinx . atomic fu . atomic Ċ import Ġkotlinx . coroutines .* Ċ import Ġmu . K Logger Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . ag ron a . Direct Buffer Ċ import Ġorg . ag ron a . concurrent . Idle Strategy Ċ Ċ fun ĠCor outine Scope . event Loop ( block : Ġsuspend ĠCor outine Scope . () Ġ-> ĠUnit ): ĠJob Ġ{ ĊĠĠĠ Ġ// ĠUN DISPATCH ED Ġmeans Ġthat Ġthis Ġcoroutine Ġwill Ġstart Ġas Ġan Ġevent Ġloop , Ġinstead Ġof Ġconcurrent ly Ġin Ġa Ġdifferent Ġthread ĊĠĠĠ Ġ// ĠĠ Ġwe Ġwant Ġthis Ġbehavior Ġto Ġprevent Ġ" stack Ġoverflow " Ġin Ġcase Ġthere Ġare Ġnested Ġcalls ĊĠĠĠ Ġreturn Ġlaunch ( start Ġ= ĠCor outine Start . UN DISPATCH ED , Ġblock Ġ= Ġblock ) Ċ } ĊĊ Ċ // ĠIf ĠTCP Ġand ĠUDP Ġboth Ġfill Ġthe Ġpipe , ĠT HERE ĠWILL ĠBE ĠFR AG MENT ATION Ġand Ġdropped ĠUDP Ġpackets ! Ċ // Ġit Ġresults Ġin Ġsever e ĠUDP Ġpacket Ġloss Ġand Ġcontent ion . Ċ // Ċ // Ġhttp :// www . is oc . org / INET 9 7 / proceedings / F 3 / F 3 _ 1 . HT M Ċ // Ġalso , Ġa Ġgoogle Ġsearch Ġon Ġjust Ġ" INET 9 7 / proceedings / F 3 / F 3 _ 1 . HT M " Ġturns Ġup Ġinteresting Ġproblems . Ċ // ĠUsually Ġit 's Ġwith ĠI SP s . Ċ /** Ċ Ġ* Ġrepresents Ġthe Ġbase Ġof Ġa Ġclient / server Ġend Ġpoint Ġfor Ġinteracting Ġwith Ġa eron Ċ Ġ* Ċ Ġ* Ġ@ param Ġtype Ġthis Ġis Ġeither Ġ" Client " Ġor Ġ" Server ", Ġdepending Ġon Ġwho Ġis Ġcreating Ġthis Ġendpoint . Ċ Ġ* Ġ@ param Ġconfig Ġthese Ġare Ġthe Ġspecific Ġconnection Ġoptions Ċ Ġ* Ġ@ param Ġconnection Func Ġallows Ġfor Ġcustom Ġconnection Ġimplementations Ġdefined Ġas Ġa Ġunit Ġfunction Ċ Ġ* Ċ Ġ* Ġ Ġ@ throws ĠSecurity Exception Ġif Ġunable Ġto Ġinitialize / generate ĠE CC Ġkeys Ċ */ Ċ abstract Ġclass ĠEnd Point < CONNECTION Ġ: ĠConnection > Ċ internal Ġconstructor ( val Ġtype : ĠClass <* >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinternal Ġval Ġconfig : ĠConfiguration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinternal Ġval Ġconnection Func : Ġ( connection Parameters : ĠConnection Params < CONNECTION >) Ġ-> ĠCONNECTION ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ: ĠAuto Closeable Ġ{ ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprotected Ġconstructor ( config : ĠConfiguration , Ġconnection Func : Ġ( connection Parameters : ĠConnection Params < CONNECTION >) Ġ-> ĠCONNECTION ) Ġ: Ġthis ( Client :: class . java , Ġconfig , Ġconnection Func ) ĊĠĠĠ Ġprotected Ġconstructor ( config : ĠServer Configuration , Ġconnection Func : Ġ( connection Parameters : ĠConnection Params < CONNECTION >) Ġ-> ĠCONNECTION ) Ġ: Ġthis ( Server :: class . java , Ġconfig , Ġconnection Func ) ĊĊĠĠĠ Ġval Ġlogger : ĠK Logger Ġ= ĠKotlin Logging . logger ( type . simple Name ) ĊĊĠĠĠ Ġinternal Ġval Ġaction Dispatch Ġ= ĠCor outine Scope ( Dispatch ers . Default ) ĊĊĠĠĠ Ġinternal Ġval Ġlistener Manager Ġ= ĠListener Manager < CONNECTION >( logger ) ĊĠĠĠ Ġinternal Ġval Ġconnections Ġ= ĠConnection Manager < CONNECTION >() ĊĊĠĠĠ Ġinternal Ġval Ġa eron Driver : ĠA eron Driver ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠReturns Ġthe Ġserialization Ġwrapper Ġif Ġthere Ġis Ġan Ġobject Ġtype Ġthat Ġneeds Ġto Ġbe Ġadded Ġoutside Ġof Ġthe Ġbasic Ġtypes . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġserialization : ĠSerialization < CONNECTION > ĊĊĠĠĠ Ġprivate Ġval Ġhandshake K ry o : ĠK ry o Extra < CONNECTION > ĊĊĠĠĠ Ġprivate Ġval Ġsend Idle Strategy : ĠCor outine Idle Strategy ĊĠĠĠ Ġprivate Ġval Ġsend Idle Strategy Hand Sh ake : ĠId le Strategy ĊĊĠĠĠ Ġprivate Ġval Ġpoll Idle Strategy : ĠCor outine Idle Strategy ĊĠĠĠ Ġinternal Ġval Ġpoll Idle Strategy Hand Sh ake : ĠId le Strategy ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCrypto Ġand Ġsignature Ġmanagement ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġval Ġcrypto : ĠCrypto Management ĊĊĠĠĠ Ġprivate Ġval Ġshutdown Ġ= Ġatomic ( false ) ĊĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġshutdown Wait er : ĠS uspend Wait er Ġ= ĠS uspend Wait er () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠReturns Ġthe Ġstorage Ġused Ġby Ġthis Ġendpoint . ĠThis Ġis Ġthe Ġbacking Ġdata Ġstructure Ġfor Ġkey / value Ġpairs , Ġand Ġcan Ġbe Ġa Ġdatabase , Ġfile , Ġetc ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠOnly Ġone Ġinstance Ġof Ġthese Ġis Ġcreated Ġfor Ġan Ġendpoint . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġstorage : ĠSettings Store ĊĊĠĠĠ Ġinternal Ġval Ġresponse Manager Ġ= ĠResponse Manager ( logger , Ġaction Dispatch ) ĊĠĠĠ Ġinternal Ġval Ġr mi Global Support Ġ= ĠR mi Manager Global < CONNECTION >( logger ) ĊĠĠĠ Ġinternal Ġval Ġr mi Connection Support : ĠR mi Manager Connections < CONNECTION > ĊĊĠĠĠ Ġinternal Ġval Ġping Manager Ġ= ĠPing Manager < CONNECTION >() ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire (! config . previous ly Used ) Ġ{ Ġ"${ type . simple Name } Ġconfiguration Ġcannot Ġbe Ġreused !" Ġ} ĊĠĠĠĠĠĠĠ Ġconfig . validate () Ġ// Ġthis Ġhappens Ġmore Ġthan Ġonce ! Ġ( this Ġis Ġok ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġserialization Ġstuff ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġserialization Ġ= Ġconfig . serialization Ġas ĠSerialization < CONNECTION > ĊĠĠĠĠĠĠĠ Ġsend Idle Strategy Ġ= Ġconfig . send Idle Strategy ĊĠĠĠĠĠĠĠ Ġpoll Idle Strategy Ġ= Ġconfig . poll Idle Strategy ĊĊĠĠĠĠĠĠĠ Ġsend Idle Strategy Hand Sh ake Ġ= Ġsend Idle Strategy . clone To Normal () ĊĠĠĠĠĠĠĠ Ġpoll Idle Strategy Hand Sh ake Ġ= Ġpoll Idle Strategy . clone To Normal () ĊĊĠĠĠĠĠĠĠ Ġhandshake K ry o Ġ= Ġserialization . init Handshake K ry o () ĊĊĠĠĠĠĠĠĠ Ġ// Ġwe Ġhave Ġto Ġbe Ġable Ġto Ġspecify Ġthe Ġproperty Ġstore ĊĠĠĠĠĠĠĠ Ġstorage Ġ= ĠSettings Store ( config . settings Store . logger ( logger ), Ġlogger ) ĊĊĠĠĠĠĠĠĠ Ġcrypto Ġ= ĠCrypto Management ( logger , Ġstorage , Ġtype , Ġconfig . enable Remote Signature Validation ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠOnly Ġstarts Ġthe Ġmedia Ġdriver Ġif Ġwe Ġare ĠNOT Ġalready Ġrunning ! ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġa eron Driver Ġ= ĠA eron Driver ( config , Ġtype , Ġlogger , Ġlistener Manager . notify Error ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Error Ġinitialize Ġendpoint ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( type . java Class Ġ== ĠServer :: class . java ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġserver Ġcannot Ġ" get " Ġglobal ĠR MI Ġobjects , Ġonly Ġthe Ġclient Ġcan ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġr mi Connection Support Ġ= ĠR mi Manager Connections ( logger , Ġresponse Manager , Ġlistener Manager , Ġserialization ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġ_, Ġ_, Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegal AccessException (" Global ĠR MI Ġaccess Ġis Ġonly Ġpossible Ġfrom Ġa ĠClient Ġconnection !") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġr mi Connection Support Ġ= ĠR mi Manager Connections ( logger , Ġresponse Manager , Ġlistener Manager , Ġserialization ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġconnection , Ġobject Id , Ġinterface Class Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ R mi Manager Connections Ġr mi Global Support . getGlobal Remote Object ( connection , Ġobject Id , Ġinterface Class ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ throws ĠException Ġif Ġthere Ġis Ġa Ġproblem Ġstarting Ġthe Ġmedia Ġdriver ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġinit Endpoint State () Ġ{ ĊĠĠĠĠĠĠĠ Ġshutdown . get AndSet ( false ) ĊĠĠĠĠĠĠĠ Ġshutdown Wait er Ġ= ĠS uspend Wait er () ĊĊĠĠĠĠĠĠĠ Ġ// ĠOnly Ġstarts Ġthe Ġmedia Ġdriver Ġif Ġwe Ġare ĠNOT Ġalready Ġrunning ! ĊĠĠĠĠĠĠĠ Ġa eron Driver . start () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġabstract Ġfun Ġnew Exception ( message : ĠString , Ġcause : ĠThrowable ? Ġ= Ġnull ): ĠThrowable ĊĊĠĠĠ Ġ// Ġused Ġinternally Ġto Ġremove Ġa Ġconnection . ĠWill Ġalso Ġremove Ġall Ġproxy Ġobjects ĊĠĠĠ Ġinternal Ġfun Ġremove Connection ( connection : ĠConnection ) Ġ{ ĊĠĠĠĠĠĠĠ Ġr mi Connection Support . close () ĊĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġremove Connection ( connection Ġas ĠCONNECTION ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠAdds Ġa Ġcustom Ġconnection Ġto Ġthe Ġserver . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġshould Ġonly Ġbe Ġused Ġin Ġsituations Ġwhere Ġthere Ġcan Ġbe ĠDI FFER ENT Ġtypes Ġof Ġconnections Ġ( such Ġas Ġa Ġ' web - based ' Ġconnection ) Ġand ĊĠĠĠĠ Ġ* Ġyou Ġwant Ġ* this * Ġendpoint Ġto Ġmanage Ġlisteners Ġ+ Ġmessage Ġdispatch ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġconnection Ġthe Ġconnection Ġto Ġadd ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġadd Connection ( connection : ĠCONNECTION ) Ġ{ ĊĠĠĠĠĠĠĠ Ġconnections . add ( connection ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠRemoves Ġa Ġcustom Ġconnection Ġto Ġthe Ġserver . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġshould Ġonly Ġbe Ġused Ġin Ġsituations Ġwhere Ġthere Ġcan Ġbe ĠDI FFER ENT Ġtypes Ġof Ġconnections Ġ( such Ġas Ġa Ġ' web - based ' Ġconnection ) Ġand ĊĠĠĠĠ Ġ* Ġyou Ġwant Ġ* this * Ġendpoint Ġto Ġmanage Ġlisteners Ġ+ Ġmessage Ġdispatch ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġconnection Ġthe Ġconnection Ġto Ġremove ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġremove Connection ( connection : ĠCONNECTION ) Ġ{ ĊĠĠĠĠĠĠĠ Ġconnections . remove ( connection ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠAdds Ġan ĠIP + subnet Ġrule Ġthat Ġdefines Ġif Ġthat ĠIP + subnet Ġis Ġallowed / den ied Ġconnectivity Ġto Ġthis Ġserver . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠBy Ġdefault , Ġif Ġthere Ġare Ġno Ġfilter Ġrules , Ġthen Ġall Ġconnections Ġare Ġallowed Ġto Ġconnect ĊĠĠĠĠ Ġ* ĠIf Ġthere Ġare Ġfilter Ġrules Ġ- Ġthen ĠONLY Ġconnections Ġfor Ġthe Ġfilter Ġthat Ġreturns Ġtrue Ġare Ġallowed Ġto Ġconnect Ġ( all Ġelse Ġare Ġdenied ) ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġfunction Ġwill Ġbe Ġcalled Ġfor Ġ** only ** Ġnetwork Ġclients Ġ( IPC Ġclient Ġare Ġexcluded ) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġfilter ( ip Filter Rule : ĠIp Filter Rule ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . filter ( ip Filter Rule ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠAdds Ġa Ġfunction Ġthat Ġwill Ġbe Ġcalled ĠBE FORE Ġa Ġclient / server Ġ" connect s " Ġwith Ġeach Ġother , Ġand Ġused Ġto Ġdetermine Ġif Ġa Ġconnection ĊĠĠĠĠ Ġ* Ġshould Ġbe Ġallowed ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠBy Ġdefault , Ġif Ġthere Ġare Ġno Ġfilter Ġrules , Ġthen Ġall Ġconnections Ġare Ġallowed Ġto Ġconnect ĊĠĠĠĠ Ġ* ĠIf Ġthere Ġare Ġfilter Ġrules Ġ- Ġthen ĠONLY Ġconnections Ġfor Ġthe Ġfilter Ġthat Ġreturns Ġtrue Ġare Ġallowed Ġto Ġconnect Ġ( all Ġelse Ġare Ġdenied ) ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠIt Ġis Ġthe Ġresponsibility Ġof Ġthe Ġcustom Ġfilter Ġto Ġwrite Ġthe Ġerror , Ġif Ġthere Ġis Ġone ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠIf Ġthe Ġfunction Ġreturns ĠTRUE , Ġthen Ġthe Ġconnection Ġwill Ġcontinue Ġto Ġconnect . ĊĠĠĠĠ Ġ* ĠIf Ġthe Ġfunction Ġreturns ĠFALSE , Ġthen Ġthe Ġother Ġend Ġof Ġthe Ġconnection Ġwill ĊĠĠĠĠ Ġ* ĠĠ Ġreceive Ġa Ġconnection Ġerror ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġfunction Ġwill Ġbe Ġcalled Ġfor Ġ** only ** Ġnetwork Ġclients Ġ( IPC Ġclient Ġare Ġexcluded ) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġfilter ( function : ĠCONNECTION . () Ġ-> ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . filter ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠAdds Ġa Ġfunction Ġthat Ġwill Ġbe Ġcalled Ġwhen Ġa Ġclient / server Ġ" connect s " Ġwith Ġeach Ġother ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġon Connect ( function : Ġsuspend ĠCONNECTION . () Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . on Connect ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalled Ġwhen Ġthe Ġremote Ġend Ġis Ġno Ġlonger Ġconnected . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠDo Ġnot Ġtry Ġto Ġsend Ġmessages ! ĠThe Ġconnection Ġwill Ġalready Ġbe Ġclosed , Ġresulting Ġin Ġan Ġerror Ġif Ġyou Ġattempt Ġto Ġdo Ġso . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġon Disconnect ( function : Ġsuspend ĠCONNECTION . () Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . on Disconnect ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalled Ġwhen Ġthere Ġis Ġan Ġerror Ġfor Ġa Ġspecific Ġconnection ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThe Ġerror Ġis Ġalso Ġsent Ġto Ġan Ġerror Ġlog Ġbefore Ġthis Ġmethod Ġis Ġcalled . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun ĠonError ( function : ĠCONNECTION .( Throwable ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . onError ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalled Ġwhen Ġthere Ġis Ġa Ġglobal Ġerror Ġ( and Ġerror Ġthat Ġis Ġnot Ġspecific Ġto Ġa Ġconnection ) ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThe Ġerror Ġis Ġalso Ġsent Ġto Ġan Ġerror Ġlog Ġbefore Ġthis Ġmethod Ġis Ġcalled . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun ĠonError Global ( function : Ġ( Throwable ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . onError ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalled Ġwhen Ġan Ġobject Ġhas Ġbeen Ġreceived Ġfrom Ġthe Ġremote Ġend Ġof Ġthe Ġconnection . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġmethod Ġshould Ġnot Ġblock Ġfor Ġlong Ġperiods Ġas Ġother Ġnetwork Ġactivity Ġwill Ġnot Ġbe Ġprocessed Ġuntil Ġit Ġreturns . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġ< Message Ġ: ĠAny > Ġon Message ( function : Ġsuspend ĠCONNECTION .( Message ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . on Message ( function ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠSends Ġa Ġ" ping " Ġpacket Ġto Ġmeasure Ġ** ROUND ĠTR IP ** Ġtime Ġto Ġthe Ġremote Ġconnection . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġmessage Ġwas Ġsuccessfully Ġsent Ġby Ġa eron ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġsuspend Ġfun Ġping ( connection : ĠConnection , Ġping Timeout Ms : ĠInt , Ġfunction : Ġsuspend ĠPing . () Ġ-> ĠUnit ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġping Manager . ping ( connection , Ġping Timeout Ms , Ġaction Dispatch , Ġresponse Manager , Ġlogger , Ġfunction ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠNOTE : Ġthis Ġ** MUST ** Ġstay Ġon Ġthe Ġsame Ġco - routine Ġthat Ġcalls Ġ" send ". ĠThis Ġcannot Ġbe Ġre - dis patched Ġonto Ġa Ġdifferent Ġcoroutine ! ĊĠĠĠĠ Ġ* ĠĠĠĠĠĠ ĠCAN NOT Ġbe Ġcalled Ġin Ġaction Ġdispatch . ĠAL WAYS ĠON ĠS AME ĠTHREAD ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġmessage Ġwas Ġsuccessfully Ġsent Ġby Ġa eron ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Suppress (" D uplicated Code ") ĊĠĠĠ Ġinternal Ġfun Ġwrite Handshake Message ( publication : ĠPublic ation , Ġmessage : ĠHand shake Message ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠThe Ġhandshake ĠsessionId ĠIS ĠNOT Ġglobally Ġunique ĊĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"[ ${ publication . sessionId ()} ] Ġsend ĠH S : Ġ$ message " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġare Ġnot Ġthread - safe ! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= Ġhandshake K ry o . write ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġobject Size Ġ= Ġbuffer . position () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinternal Buffer Ġ= Ġbuffer . internal Buffer ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult : ĠLong ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġpublication . offer ( internal Buffer , Ġ 0 , Ġobject Size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ>= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsuccess ! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġpublication Ġis Ġnot Ġconnected Ġto Ġa Ġsubscriber , Ġthis Ġcan Ġbe Ġan Ġinter mitt ent Ġstate Ġas Ġsubscribers Ġcome Ġand Ġgo . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠNOT _ CONNECTED : ĠLong Ġ= Ġ- 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġoffer Ġfailed Ġdue Ġto Ġback Ġpressure Ġfrom Ġthe Ġsubscribers Ġprevent ing Ġfurther Ġtransmission . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠBACK _ PRESS URE D : ĠLong Ġ= Ġ- 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġoffer Ġfailed Ġdue Ġto Ġan Ġadministration Ġaction Ġand Ġshould Ġbe Ġretried . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġaction Ġis Ġan Ġoperation Ġsuch Ġas Ġlog Ġrotation Ġwhich Ġis Ġlikely Ġto Ġhave Ġsucceeded Ġby Ġthe Ġnext Ġretry Ġattempt . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠADMIN _ ACTION : ĠLong Ġ= Ġ- 3 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ>= ĠPublic ation . ADMIN _ ACTION ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġshould Ġretry . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Idle Strategy Hand Sh ake . idle () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġmore Ġcritical Ġerror Ġsending Ġthe Ġmessage . Ġwe Ġshouldn 't Ġretry Ġor Ġanything . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġexception Ġwill Ġbe Ġa ĠClient Exception Ġor Ġa ĠServer Exception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexception Ġ= Ġnew Exception ("[ ${ publication . sessionId ()} ] ĠError Ġsending Ġhandshake Ġmessage . Ġ$ message Ġ(${ errorCode Name ( result )}) ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠListener Manager . clean StackTrace Internal ( exception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( exception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( e Ġis ĠClient Exception Ġ|| Ġe Ġis ĠServer Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexception Ġ= Ġnew Exception ("[ ${ publication . sessionId ()} ] ĠError Ġserial izing Ġhandshake Ġmessage Ġ$ message ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠListener Manager . clean StackTrace ( exception , Ġ 2 ) Ġ// Ġ 2 Ġbecause Ġwe Ġdo Ġnot Ġwant Ġto Ġsee Ġthe Ġstack Ġfor Ġthe Ġabstract Ġ` new Exception ` ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( exception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Idle Strategy Hand Sh ake . reset () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ param Ġbuffer ĠThe Ġbuffer ĊĠĠĠĠ Ġ* Ġ@ param Ġoffset ĠThe Ġoffset Ġfrom Ġthe Ġstart Ġof Ġthe Ġbuffer ĊĠĠĠĠ Ġ* Ġ@ param Ġlength ĠThe Ġnumber Ġof Ġbytes Ġto Ġextract ĊĠĠĠĠ Ġ* Ġ@ param Ġheader ĠThe Ġa eron Ġheader Ġinformation ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġthe Ġmessage ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ// Ġnote : ĠCAN NOT Ġbe Ġcalled Ġin Ġaction Ġdispatch . ĠAL WAYS ĠON ĠS AME ĠTHREAD ĊĠĠĠ Ġinternal Ġfun Ġread Handshake Message ( buffer : ĠDirect Buffer , Ġoffset : ĠInt , Ġlength : ĠInt , Ġheader : ĠHeader ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNOTE : ĠThis ĠA BS OL UT ELY ĠMUST Ġbe Ġdone Ġon Ġthe Ġsame Ġthread ! ĠThis Ġcannot Ġbe Ġdone Ġon Ġa Ġnew Ġone , Ġbecause Ġthe Ġbuffer Ġcould Ġchange ! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġhandshake K ry o . read ( buffer , Ġoffset , Ġlength ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"[ ${ header . sessionId ()} ] Ġreceived ĠH S : Ġ$ message " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġhandshake ĠsessionId ĠIS ĠNOT Ġglobally Ġunique ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Error Ġde - serializing Ġmessage Ġon Ġconnection Ġ${ header . sessionId ()} !", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( e ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġread Ġthe Ġmessage Ġfrom Ġthe Ġa eron Ġbuffer ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġbuffer ĠThe Ġbuffer ĊĠĠĠĠ Ġ* Ġ@ param Ġoffset ĠThe Ġoffset Ġfrom Ġthe Ġstart Ġof Ġthe Ġbuffer ĊĠĠĠĠ Ġ* Ġ@ param Ġlength ĠThe Ġnumber Ġof Ġbytes Ġto Ġextract ĊĠĠĠĠ Ġ* Ġ@ param Ġheader ĠThe Ġa eron Ġheader Ġinformation ĊĠĠĠĠ Ġ* Ġ@ param Ġconnection ĠThe Ġconnection Ġthis Ġmessage Ġhappened Ġon ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġprocess Message ( buffer : ĠDirect Buffer , Ġoffset : ĠInt , Ġlength : ĠInt , Ġheader : ĠHeader , Ġconnection : ĠConnection ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġthis Ġis Ġprocessed Ġon Ġthe Ġthread Ġthat Ġcalls Ġ" poll ". ĠSub scriptions Ġare ĠNOT Ġmulti - thread Ġsafe ! ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġconnection Ġas ĠCONNECTION ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNOTE : ĠThis ĠA BS OL UT ELY ĠMUST Ġbe Ġdone Ġon Ġthe Ġsame Ġthread ! ĠThis Ġcannot Ġbe Ġdone Ġon Ġa Ġnew Ġone , Ġbecause Ġthe Ġbuffer Ġcould Ġchange ! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġserialization . read Message ( buffer , Ġoffset , Ġlength , Ġconnection ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ Ġ"[ ${ header . sessionId ()} ] Ġreceived : Ġ$ message " Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNOTE : ĠThis ĠMUST Ġbe Ġon Ġa Ġnew Ġco - routine ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Dispatch . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Message ( message , Ġconnection ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Error Ġprocessing Ġmessage ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġhandshake ĠsessionId ĠIS ĠNOT Ġglobally Ġunique ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ("[ ${ header . sessionId ()} ] ĠError Ġde - serializing Ġmessage ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠActually Ġprocess Ġthe Ġmessage . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġprocess Message ( message : ĠAny ?, Ġconnection : ĠCONNECTION ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( message ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPing Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġping Manager . manage ( connection , Ġresponse Manager , Ġmessage , Ġlogger ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsmall Ġproblem ... ĠIf Ġwe Ġexpect ĠIN ĠORDER Ġmessages Ġ( ie : Ġsetting Ġa Ġvalue , Ġthen Ġlater Ġreading Ġthe Ġvalue ), Ġmultiple Ġthreads Ġdon 't Ġwork . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġis Ġworked Ġaround Ġby Ġhaving ĠR MI Ġalways Ġreturn Ġ( unless Ġasync ), Ġeven Ġwith Ġa Ġnull Ġvalue , Ġso Ġthe ĠCALL ING Ġside Ġof ĠR MI Ġwill Ġalways ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġgo Ġin Ġ" lock Ġstep " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠR mi Message Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġif Ġwe Ġare Ġan ĠR MI Ġmessage / registration , Ġwe Ġhave Ġvery Ġspecific , Ġdefined Ġbehavior . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġdo Ġnot Ġuse Ġthe Ġ" normal " Ġlistener Ġcallback Ġpattern Ġbecause Ġthis Ġrequire Ġspecial Ġfunctionality ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġr mi Global Support . manage ( serialization , Ġconnection , Ġmessage , Ġr mi Connection Support , Ġresponse Manager , Ġlogger ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAny Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġhas Listeners Ġ= Ġlistener Manager . notify On Message ( connection , Ġmessage ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġeach Ġconnection Ġregisters , Ġand Ġis Ġpol led ĠIN DEPEND ENT LY Ġfor Ġmessages . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhas Listeners Ġ= Ġhas Listeners Ġor Ġconnection . notify On Message ( message ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! has Listeners ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" No Ġmessage Ġcallbacks Ġfound Ġfor Ġ${ message :: class . java . simple Name }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Error Ġprocessing Ġmessage ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdo Ġnothing , Ġthere Ġwere Ġproblems Ġwith Ġthe Ġmessage ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( message Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" No Ġmessage Ġcallbacks Ġfound Ġfor Ġ${ message :: class . java . simple Name }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Unknown Ġmessage Ġreceived !! ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠNOTE : Ġthis Ġ** MUST ** Ġstay Ġon Ġthe Ġsame Ġco - routine Ġthat Ġcalls Ġ" send ". ĠThis Ġcannot Ġbe Ġre - dis patched Ġonto Ġa Ġdifferent Ġcoroutine ! ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġmessage Ġwas Ġsuccessfully Ġsent Ġby Ġa eron ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Suppress (" D uplicated Code ", Ġ" UN CHECK ED _ CAST ") ĊĠĠĠ Ġinternal Ġsuspend Ġfun Ġsend ( message : ĠAny , Ġpublication : ĠPublic ation , Ġconnection : ĠConnection ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠThe Ġhandshake ĠsessionId ĠIS ĠNOT Ġglobally Ġunique ĊĠĠĠĠĠĠĠ Ġlogger . trace Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"[ ${ publication . sessionId ()} ] Ġsend : Ġ$ message " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġconnection Ġas ĠCONNECTION ĊĊĠĠĠĠĠĠĠ Ġ// Ġsince ĠANY Ġthread Ġcan Ġcall Ġ' send ', Ġwe Ġhave Ġto Ġtake Ġk ry o Ġinstances Ġin Ġa Ġsafe Ġway ĊĠĠĠĠĠĠĠ Ġval Ġk ry o : ĠK ry o Extra < CONNECTION > Ġ= Ġserialization . take K ry o () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= Ġk ry o . write ( connection , Ġmessage ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġobject Size Ġ= Ġbuffer . position () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinternal Buffer Ġ= Ġbuffer . internal Buffer ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult : ĠLong ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġpublication . offer ( internal Buffer , Ġ 0 , Ġobject Size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ>= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsuccess ! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġpublication Ġis Ġnot Ġconnected Ġto Ġa Ġsubscriber , Ġthis Ġcan Ġbe Ġan Ġinter mitt ent Ġstate Ġas Ġsubscribers Ġcome Ġand Ġgo . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠNOT _ CONNECTED : ĠLong Ġ= Ġ- 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġoffer Ġfailed Ġdue Ġto Ġback Ġpressure Ġfrom Ġthe Ġsubscribers Ġprevent ing Ġfurther Ġtransmission . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠBACK _ PRESS URE D : ĠLong Ġ= Ġ- 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġoffer Ġfailed Ġdue Ġto Ġan Ġadministration Ġaction Ġand Ġshould Ġbe Ġretried . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġaction Ġis Ġan Ġoperation Ġsuch Ġas Ġlog Ġrotation Ġwhich Ġis Ġlikely Ġto Ġhave Ġsucceeded Ġby Ġthe Ġnext Ġretry Ġattempt . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* Ġ Ġval ĠADMIN _ ACTION : ĠLong Ġ= Ġ- 3 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ>= ĠPublic ation . ADMIN _ ACTION ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġshould Ġretry , ĠBUT Ġwe Ġwant Ġsuspend ĠANY ONE ĠELSE Ġtrying Ġto Ġwrite Ġat Ġthe Ġsame Ġtime ! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Idle Strategy . idle () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ== ĠPublic ation . CLOSED Ġ&& Ġconnection . is Closed Via A eron ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġcan Ġhappen Ġwhen Ġwe Ġuse ĠR MI Ġto Ġclose Ġa Ġconnection . ĠR MI Ġwill Ġ( in Ġmost Ġcases ) ĠAL WAYS Ġsend Ġa Ġresponse Ġwhen Ġit 's ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdone Ġexecuting . ĠIf Ġthe Ġconnection Ġis Ġ* closed * Ġfirst Ġ( because Ġan ĠR MI Ġmethod Ġclosed Ġit ), Ġthen Ġwe Ġwill Ġnot Ġbe Ġable Ġto ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsend Ġthe Ġmessage . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNOTE : Ġwe Ġalready Ġknow Ġthe Ġconnection Ġis Ġclosed . Ġwe Ġclosed Ġit Ġ( so Ġit Ġdoesn 't Ġmake Ġsense Ġto Ġemit Ġan Ġerror Ġabout Ġthis ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġmore Ġcritical Ġerror Ġsending Ġthe Ġmessage . Ġwe Ġshouldn 't Ġretry Ġor Ġanything . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠerrorMessage Ġ= Ġ"[ ${ publication . sessionId ()} ] ĠError Ġsending Ġmessage . Ġ$ message Ġ(${ errorCode Name ( result )}) " ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġeither Ġclient Ġor Ġserver . ĠNo Ġother Ġchoices . ĠWe Ġcreate Ġan Ġexception , Ġbecause Ġit 's Ġmore Ġuseful ! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexception Ġ= Ġnew Exception ( errorMessage ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ+ 2 Ġbecause Ġwe Ġdo Ġnot Ġwant Ġto Ġsee Ġthe Ġstack Ġfor Ġthe Ġabstract Ġ` new Exception ` ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ+ 2 Ġmore Ġbecause Ġwe Ġdo Ġnot Ġneed Ġto Ġsee Ġthe Ġ" internals " Ġfor Ġsending Ġmessages . ĠThe Ġimportant Ġpart Ġof Ġthe Ġstack Ġtrace Ġis ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwhere Ġwe Ġsee Ġwho Ġis Ġcalling Ġ" send ()" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠListener Manager . clean StackTrace ( exception , Ġ 4 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" A eron Ġerror !", Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠMethod Response Ġ&& Ġmessage . result Ġis ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġmessage . result Ġas ĠException ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ("[ ${ publication . sessionId ()} ] ĠError Ġserial izing Ġmessage Ġ$ message ", Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ("[ ${ publication . sessionId ()} ] ĠError Ġserial izing Ġmessage Ġ$ message ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Manager . notify Error ( connection , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Idle Strategy . reset () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserialization . return K ry o ( k ry o ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ return Ġthe Ġerror Ġcode Ġtext Ġfor Ġthe Ġspecified Ġnumber ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun ĠerrorCode Name ( result : ĠLong ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( result ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġpublication Ġis Ġnot Ġconnected Ġto Ġa Ġsubscriber , Ġthis Ġcan Ġbe Ġan Ġinter mitt ent Ġstate Ġas Ġsubscribers Ġcome Ġand Ġgo . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPublic ation . NOT _ CONNECTED Ġ-> Ġ" Not Ġconnected " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġoffer Ġfailed Ġdue Ġto Ġback Ġpressure Ġfrom Ġthe Ġsubscribers Ġprevent ing Ġfurther Ġtransmission . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPublic ation . BACK _ PRESS URE D Ġ-> Ġ" Back Ġpress ured " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġaction Ġis Ġan Ġoperation Ġsuch Ġas Ġlog Ġrotation Ġwhich Ġis Ġlikely Ġto Ġhave Ġsucceeded Ġby Ġthe Ġnext Ġretry Ġattempt . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPublic ation . ADMIN _ ACTION Ġ-> Ġ" Administr ative Ġaction " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe ĠPublic ation Ġhas Ġbeen Ġclosed Ġand Ġshould Ġno Ġlonger Ġbe Ġused . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPublic ation . CLOSED Ġ-> Ġ" Publication Ġis Ġclosed " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġthis Ġhappens Ġthen Ġthe Ġpublication Ġshould Ġbe Ġclosed Ġand Ġa Ġnew Ġone Ġadded . ĠTo Ġmake Ġit Ġless Ġlikely Ġto Ġhappen Ġthen Ġincrease Ġthe Ġterm Ġbuffer Ġlength . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPublic ation . MAX _ POSITION _ EX CE EDED Ġ-> Ġ" Maximum Ġterm Ġposition Ġexceeded " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalStateException (" Unknown Ġerror Ġcode : Ġ$ result ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġ" EndPoint Ġ[ ${ type . simple Name }] " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprime Ġ= Ġ 3 1 ĊĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġ 1 ĊĠĠĠĠĠĠĠ Ġresult Ġ= Ġprime Ġ* Ġresult Ġ+ Ġ( crypto . hashCode ()) ĊĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( other Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother . java Class ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġother Ġas ĠEnd Point < *> ĊĠĠĠĠĠĠĠ Ġreturn Ġcrypto Ġ== Ġother . crypto ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthis Ġendpoint Ġhas Ġbeen Ġclosed ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġis Shutdown (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġshutdown . value ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠW aits Ġfor Ġthis Ġendpoint Ġto Ġbe Ġclosed ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun ĠwaitFor Close () Ġ{ ĊĠĠĠĠĠĠĠ Ġshutdown Wait er . do Wait () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠChecks Ġto Ġsee Ġif Ġan Ġendpoint Ġ( using Ġthe Ġcurrent Ġconfiguration ) Ġis Ġrunning . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġmedia Ġdriver Ġis Ġactive Ġand Ġrunning ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġis Running (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġa eron Driver . isRunning () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠChecks Ġto Ġsee Ġif Ġan Ġendpoint Ġ( using Ġthe Ġspecified Ġconfiguration ) Ġis Ġrunning . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġmedia Ġdriver Ġis Ġactive Ġand Ġrunning ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġis Running ( context : ĠMedia Driver . Context ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġif Ġthe Ġmedia Ġdriver Ġis Ġrunning , Ġit Ġwill Ġbe Ġa Ġquick Ġconnection . ĠUsually Ġ 1 0 0 ms Ġor Ġso ĊĠĠĠĠĠĠĠ Ġreturn Ġcontext . is Driver Active ( 1 _ 0 0 0 ) Ġ{ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfinal Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( shutdown . compare AndSet ( expect Ġ= Ġfalse , Ġupdate Ġ= Ġtrue )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Shut ting Ġdown ..." Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġa eron Driver . close () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnections . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠConnection s Ġare Ġclosed Ġfirst , Ġbecause Ġwe Ġwant Ġto Ġmake Ġsure Ġthat Ġno ĠR MI Ġmessages Ġcan Ġbe Ġreceived ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwhen Ġwe Ġclose Ġthe ĠR MI Ġsupport Ġobjects Ġ( in Ġwhich Ġcase , Ġweird Ġ- Ġbut Ġharm less Ġ- Ġerrors Ġshow Ġup ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġwill Ġwait Ġfor ĠR MI Ġtimeouts Ġif Ġthere Ġare ĠR MI Ġin - progress . Ġ( this Ġhappens Ġif Ġwe Ġclose Ġvia Ġand ĠR MI Ġmethod ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse Manager . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthe Ġstorage Ġis Ġclosed Ġvia Ġthis Ġas Ġwell . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstorage . close () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose 0 () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġif Ġwe Ġare Ġwaiting Ġfor Ġshutdown , Ġcancel Ġthe Ġwaiting Ġthread Ġ( since Ġwe Ġhave Ġshutdown Ġnow ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshutdown Wait er . cancel () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġopen Ġfun Ġclose 0 () Ġ{} Ċ } Ċ