package Ġnet . cord a . client . rpc . internal Ċ Ċ import Ġnet . cord a . client . rpc . Connection FailureException Ċ import Ġnet . cord a . client . rpc . C ord a RPC Client Ċ import Ġnet . cord a . client . rpc . C ord a RPC Client Configuration Ċ import Ġnet . cord a . client . rpc . C ord a RPC Connection Ċ import Ġnet . cord a . client . rpc . Grace ful Re connect Ċ import Ġnet . cord a . client . rpc . Max Rpc Retry Exception Ċ import Ġnet . cord a . client . rpc . Permission Exception Ċ import Ġnet . cord a . client . rpc . RPC Connection Ċ import Ġnet . cord a . client . rpc . RPC Exception Ċ import Ġnet . cord a . client . rpc . internal . Re connecting C ord a RP CO ps . Re connecting RPC Connection . Current State . CLOSED Ċ import Ġnet . cord a . client . rpc . internal . Re connecting C ord a RP CO ps . Re connecting RPC Connection . Current State . CONNECTED Ċ import Ġnet . cord a . client . rpc . internal . Re connecting C ord a RP CO ps . Re connecting RPC Connection . Current State . CONNECT ING Ċ import Ġnet . cord a . client . rpc . internal . Re connecting C ord a RP CO ps . Re connecting RPC Connection . Current State . DI ED Ċ import Ġnet . cord a . client . rpc . internal . Re connecting C ord a RP CO ps . Re connecting RPC Connection . Current State . UN CONNECTED Ċ import Ġnet . cord a . client . rpc . reconnect . Could Not Start Flow Exception Ċ import Ġnet . cord a . core . flows . StateMachine Run Id Ċ import Ġnet . cord a . core . internal . messaging . Internal C ord a RP CO ps Ċ import Ġnet . cord a . core . internal . min Ċ import Ġnet . cord a . core . internal . times Ċ import Ġnet . cord a . core . internal . unchecked Cast Ċ import Ġnet . cord a . core . messaging . Client Rpc Ssl Options Ċ import Ġnet . cord a . core . messaging . C ord a RP CO ps Ċ import Ġnet . cord a . core . messaging . Data Feed Ċ import Ġnet . cord a . core . messaging . Flow Handle Ċ import Ġnet . cord a . core . utilities . Network Host And Port Ċ import Ġnet . cord a . core . utilities . context Logger Ċ import Ġnet . cord a . core . utilities . debug Ċ import Ġnet . cord a . core . utilities . seconds Ċ import Ġnet . cord a . node api . exceptions . Rejected Command Exception Ċ import Ġorg . apache . activ emq . art emis . api . core . Active MQ Connection Timed Out Exception Ċ import Ġorg . apache . activ emq . art emis . api . core . Active MQ Security Exception Ċ import Ġorg . apache . activ emq . art emis . api . core . Active MQ Un Blocked Exception Ċ import Ġjava . io . Not Serializable Exception Ċ import Ġjava . lang . reflect . Invocation Handler Ċ import Ġjava . lang . reflect . Invocation TargetException Ċ import Ġjava . lang . reflect . Method Ċ import Ġjava . lang . reflect . Proxy Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . ExecutorService Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ /** Ċ Ġ* ĠWrapper Ġover Ġ[ C ord a RP CO ps ] Ġthat Ġhandles Ġexceptions Ġwhen Ġthe Ġnode Ġor Ġthe Ġconnection Ġto Ġthe Ġnode Ġfail . Ċ Ġ* Ċ Ġ* ĠAll Ġoperations Ġare Ġretried Ġon Ġfailure , Ġexcept Ġflow Ġstart Ġoperations Ġthat Ġdie Ġbefore Ġreceiving Ġa Ġvalid Ġ[ Flow Handle ], Ġin Ġwhich Ġcase Ġa Ċ Ġ* Ġ[ Could Not Start Flow Exception ] Ġis Ġthrown . Ċ Ġ* Ċ Ġ* ĠWhen Ġcalling Ġmethods Ġthat Ġreturn Ġa Ġ[ Data Feed ] Ġlike Ġ[ C ord a RP CO ps . vault Track By ], Ġthe Ġreturned Ġ[ Data Feed . updates ] Ġobject Ġwill Ġno Ġlonger Ċ Ġ* Ġbe Ġa Ġusable Ġ[ rx . Observable ] Ġbut Ġan Ġinstance Ġof Ġ[ Re connecting Observable ]. Ċ Ġ* ĠThe Ġcaller Ġhas Ġto Ġexplicitly Ġcast Ġto Ġ[ Re connecting Observable ] Ġand Ġcall Ġ[ Re connecting Observable . subscribe ]. ĠIf Ġused Ġas Ġan Ġ[ rx . Observable ] Ċ Ġ* Ġit Ġwill Ġjust Ġfail . Ċ Ġ* ĠThe Ġreturned Ġ[ Data Feed . snapshot ] Ġis Ġthe Ġsnapshot Ġas Ġit Ġwas Ġwhen Ġthe Ġfeed Ġwas Ġfirst Ġretrieved . Ċ Ġ* Ċ Ġ* ĠNote : ĠThere Ġis Ġno Ġguarantee Ġthat Ġobservations Ġwill Ġnot Ġbe Ġlost . Ċ Ġ* Ċ Ġ* Ġ* This Ġclass Ġis Ġnot Ġa Ġstable ĠAPI . ĠAny Ġproject Ġthat Ġwants Ġto Ġuse Ġit , Ġmust Ġcopy Ġand Ġpaste Ġit .* Ċ Ġ*/ Ċ // ĠTODO ĠThe Ġexecutor Ġservice Ġis Ġnot Ġneeded . ĠAll Ġwe Ġneed Ġis Ġa Ġsingle Ġthread Ġthat Ġdeals Ġwith Ġreconnect ing Ġand Ġonto Ġwhich Ċ // Ġ ĠRe connecting Observable s Ġand Ġother Ġthings Ġcan Ġattach Ġthemselves Ġas Ġlisteners Ġfor Ġreconnect Ġevents . Ċ class ĠRe connecting C ord a RP CO ps Ġprivate Ġconstructor ( ĊĠĠĠĠĠĠĠ Ġval Ġreconnect ing RPC Connection : ĠRe connecting RPC Connection Ċ ) Ġ: ĠInternal C ord a RP CO ps Ġby Ġproxy ( reconnect ing RPC Connection ) Ġ{ ĊĠĠĠ Ġconstructor ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnode Host And Ports : ĠList < Network Host And Port >, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġusername : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpassword : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrpc Configuration : ĠC ord a RPC Client Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgrace ful Re connect : ĠGr ace ful Re connect Ġ= ĠGr ace ful Re connect (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġssl Configuration : ĠClient Rpc Ssl Options ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclass Loader : ĠClassLoader ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġob servers Pool : ĠExecutor Service ĊĠĠĠ Ġ) Ġ: Ġthis ( Re connecting RPC Connection ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnode Host And Ports , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġusername , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpassword , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrpc Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġssl Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclass Loader , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgrace ful Re connect , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġob servers Pool )) ĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= Ġcontext Logger () ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġproxy ( reconnect ing RPC Connection : ĠRe connecting RPC Connection ): ĠInternal C ord a RP CO ps Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠProxy . new Proxy Instance ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis :: class . java . class Loader , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠarrayOf ( Internal C ord a RP CO ps :: class . java ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠError Intercept ing Handler ( reconnect ing RPC Connection )) Ġas ĠInternal C ord a RP CO ps ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġval Ġretry Flows Pool Ġ= ĠExec utors . new Scheduled ThreadPool ( 1 ) ĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThis Ġfunction Ġruns Ġa Ġflow Ġand Ġretries Ġuntil Ġit Ġcompletes Ġsuccessfully . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ[ run Flow ] Ġis Ġa Ġfunction Ġthat Ġstarts Ġa Ġflow . ĊĠĠĠĠ Ġ* Ġ[ has Flow Started ] Ġis Ġa Ġfunction Ġthat Ġchecks Ġif Ġthe Ġflow Ġhas Ġactually Ġcompleted Ġby Ġchecking Ġsome Ġside - effect , Ġfor Ġexample Ġthe Ġvault . ĊĠĠĠĠ Ġ* Ġ[ on Flow Confirmed ] ĠCallback Ġwhen Ġthe Ġflow Ġis Ġconfirmed . ĊĠĠĠĠ Ġ* Ġ[ timeout ] ĠIndic ative Ġtimeout Ġto Ġwait Ġuntil Ġthe Ġflow Ġwould Ġcreate Ġthe Ġside - effect . ĠShould Ġbe Ġincreased Ġif Ġthe Ġflow Ġis Ġslow . ĠNote Ġthat ĊĠĠĠĠ Ġ* Ġthis Ġtimeout Ġis Ġcalculated Ġafter Ġthe Ġrpc Ġclient Ġhas Ġre connected Ġto Ġthe Ġnode . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠNote Ġthat Ġthis Ġmethod Ġdoes Ġnot Ġguarantee Ġ 1 0 0 % Ġthat Ġthe Ġflow Ġwill Ġnot Ġbe Ġstarted Ġtwice . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġrun Flow With Logical Retry ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Flow : Ġ( C ord a RP CO ps ) Ġ-> ĠState Machine Run Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhas Flow Started : Ġ( C ord a RP CO ps ) Ġ-> ĠBoolean , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Flow Confirmed : Ġ() Ġ-> ĠUnit Ġ= Ġ{}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtimeout : ĠDuration Ġ= Ġ 4 . seconds ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Flow ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Flow Confirmed () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠCould Not Start Flow Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Couldn 't Ġstart Ġflow : Ġ${ e . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretry Flows Pool . schedule ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! has Flow Started ( this )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Flow With Logical Retry ( run Flow , Ġhas Flow Started , Ġon Flow Confirmed , Ġtimeout ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Flow Confirmed () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtimeout . seconds , ĠTimeUnit . SECONDS ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠHelper Ġclass Ġuseful Ġfor Ġreconnect ing Ġto Ġa ĠNode . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġdata Ġclass ĠRe connecting RPC Connection ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnode Host And Ports : ĠList < Network Host And Port >, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġusername : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpassword : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrpc Configuration : ĠC ord a RPC Client Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġssl Configuration : ĠClient Rpc Ssl Options ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclass Loader : ĠClassLoader ?, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġgrace ful Re connect : ĠGr ace ful Re connect Ġ= ĠGr ace ful Re connect (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġob servers Pool : ĠExecutor Service ĊĠĠĠ Ġ) Ġ: ĠRPC Connection < C ord a RP CO ps > Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġcurrent RPC Connection : ĠC ord a RPC Connection ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġenum Ġclass ĠCurrent State Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUN CONNECTED , ĠCONNECT ED , ĠCONNECT ING , ĠCLO SED , ĠDI ED ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ@ Volatile ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġcurrent State Ġ= ĠUN CONNECTED ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġcurrent : ĠC ord a RPC Connection ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Synchron ized Ġget () Ġ= Ġwhen Ġ( currentState ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġfirst Ġattempt Ġto Ġestablish Ġa Ġconnection Ġwill Ġtry Ġevery Ġaddress Ġonly Ġonce . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠUN CONNECTED Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnect ( node Host And Ports . size ) Ġ?: Ġthrow ĠIllegalArgumentException (" The ĠRe connecting RPC Connection Ġhas Ġbeen Ġclosed .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCONNECT ED Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent RPC Connection !! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCLO SED Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalArgumentException (" The ĠRe connecting RPC Connection Ġhas Ġbeen Ġclosed .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCONNECT ING , ĠDI ED Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalArgumentException (" Illegal Ġstate : Ġ$ currentState Ġ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ@ Synchron ized ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġdo Re connect ( e : ĠThrowable , Ġprevious Connection : ĠC ord a RPC Connection ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġdon 't Ġwant Ġto Ġreconnect Ġif Ġwe Ġpurpose ly Ġclosed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( previous Connection Ġ!= Ġcurrent RPC Connection ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe 've Ġalready Ġdone Ġthis , Ġskip ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFirst Ġone Ġto Ġget Ġhere Ġgets Ġto Ġdo Ġall Ġthe Ġreconnect Ġlogic , Ġincluding Ġcalling Ġon Disconnect Ġand Ġon Re connect . ĠThis Ġmakes Ġsure ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthat Ġthey 're Ġonly Ġcalled Ġonce Ġper Ġreconnect . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent State Ġ= ĠDI ED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgrace ful Re connect . on Disconnect . invoke () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// TODO Ġ- Ġhandle Ġerror Ġcases ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Re connecting Ġto Ġ${ this . node Host And Ports } Ġdue Ġto Ġerror : Ġ${ e . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug ("", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnect ( rpc Configuration . max Re connect Attempts ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprevious Connection ?. force Close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgrace ful Re connect . on Re connect . invoke () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠCalled Ġon Ġexternal Ġerror . ĊĠĠĠĠĠĠĠĠ Ġ* ĠWill Ġblock Ġuntil Ġthe Ġconnection Ġis Ġestablished Ġagain . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġfun Ġreconnect OnError ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġprevious Connection Ġ= Ġcurrent RPC Connection ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdo Re connect ( e , Ġprevious Connection ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġconnect ( max Connect Attempts : ĠInt ): ĠC ord a RPC Connection ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent State Ġ= ĠCONNECT ING ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent RPC Connection Ġ= Ġestablish Connection With Retry ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrpc Configuration . connection Retry Interval , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretries Ġ= Ġmax Connect Attempts ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIt 's Ġpossible Ġwe Ġcould Ġget Ġclosed Ġwhile Ġwaiting Ġfor Ġthe Ġconnection Ġto Ġestablish . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent State Ġ= ĠCONNECT ED ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcurrent RPC Connection ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠEst abl ishes Ġa Ġconnection Ġby Ġautomatically Ġretry ing Ġif Ġthe Ġattempt Ġto Ġestablish Ġa Ġconnection Ġfails . ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ param Ġretry Interval Ġthe Ġinterval Ġbetween Ġretries . ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ param Ġround Rob in Index Ġindex Ġof Ġthe Ġaddress Ġthat Ġwill Ġbe Ġused Ġfor Ġthe Ġconnection . ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ param Ġretries Ġthe Ġnumber Ġof Ġretries Ġremaining . ĠA Ġnegative Ġvalue Ġimplies Ġinfinite Ġretries . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġtail rec Ġfun Ġestablish Connection With Retry ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretry Interval : ĠDuration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġround Rob in Index : ĠInt Ġ= Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretries : ĠInt Ġ= Ġ- 1 ĊĠĠĠĠĠĠĠ Ġ): ĠC ord a RPC Connection ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe 've Ġdecided Ġto Ġexit Ġfor Ġsome Ġreason Ġ( maybe Ġthe Ġclient Ġis Ġbeing Ġshutdown ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattempted Address Ġ= Ġnode Host And Ports [ round Rob in Index ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Connecting Ġto : Ġ$ at tempted Address ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠC ord a RPC Client ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġattempted Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrpc Configuration . copy ( connection Max Retry Interval Ġ= Ġretry Interval , Ġmax Re connect Attempts Ġ= Ġ 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġssl Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclass Loader ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ). start ( username , Ġpassword ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCheck Ġconnection Ġis Ġtruly Ġoperational Ġbefore Ġreturning Ġit . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( it . proxy . node Info (). legal Id entities And Certs . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Could Ġnot Ġestablish Ġconnection Ġto Ġ$ at tempted Address ." ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug Ġ{ Ġ" Connection Ġsuccessfully Ġestablished Ġwith : Ġ$ at tempted Address " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( ex ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠActive MQ Security Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġlogin Ġto Ġnode .", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRPC Exception , Ġis ĠActive MQ Connection Timed Out Exception , Ġis ĠActive MQ Un Blocked Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDel iber ately Ġnot Ġlogging Ġfull Ġstack Ġtrace Ġas Ġit Ġwill Ġbe Ġfull Ġof Ġinternal Ġstack traces . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug Ġ{ Ġ" Exception Ġupon Ġestablish ing Ġconnection : Ġ${ ex . message }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPermission Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDel iber ately Ġnot Ġlogging Ġfull Ġstack Ġtrace Ġas Ġit Ġwill Ġbe Ġfull Ġof Ġinternal Ġstack traces . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug Ġ{ Ġ" Permission ĠException Ġestablish ing Ġconnection : Ġ${ ex . message }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Unknown Ġexception Ġ[ ${ ex . java Class . name }] Ġupon Ġestablish ing Ġconnection .", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġremaining Retries Ġ= Ġif Ġ( retries Ġ< Ġ 0 ) Ġretries Ġelse Ġ( retries Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( remaining Retries Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRPC Exception (" Cannot Ġconnect Ġto Ġserver ( s ). ĠT ried Ġwith Ġall Ġavailable Ġservers .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCould Ġnot Ġconnect Ġthis Ġtime Ġround Ġ- Ġpause Ġbefore Ġgiving Ġanother Ġtry . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( retry Interval . to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnext Round Rob in Index Ġ= Ġ( round Rob in Index Ġ+ Ġ 1 ) Ġ% Ġnode Host And Ports . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnext Interval Ġ= Ġmin ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrpc Configuration . connection Max Retry Interval , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretry Interval Ġ* Ġrpc Configuration . connection Retry Interval Multiplier ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Could Ġnot Ġestablish Ġconnection . Ġ ĠNext Ġretry Ġinterval Ġ$ next Interval ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġestablish Connection With Retry ( next Interval , Ġnext Round Rob in Index , Ġremaining Retries ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġproxy : ĠC ord a RP CO ps ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġcurrent . proxy ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġserver Protocol Version ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġcurrent . server Protocol Version ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġnotify Server And Close () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent State Ġ= ĠCLO SED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent RPC Connection ?. notify Server And Close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġforce Close () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent State Ġ= ĠCLO SED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent RPC Connection ?. force Close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġfun Ġis Closed (): ĠBoolean Ġ= Ġcurrent State Ġ== ĠCLO SED ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġclass ĠError Intercept ing Handler ( val Ġreconnect ing RPC Connection : ĠRe connecting RPC Connection ) Ġ: ĠInvocation Handler Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġfun ĠMethod . is Start Flow () Ġ= Ġname . startsWith (" start Flow ") Ġ|| Ġname . startsWith (" start Tracked Flow ") ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġcheck If Is Start Flow ( method : ĠMethod , Ġe : ĠInvocation TargetException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( method . is Start Flow ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDon 't Ġretry Ġflows ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠCould Not Start Flow Exception ( e . target Exception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThis Ġmethod Ġretries Ġthe Ġinvoked Ġoperation Ġin Ġa Ġloop Ġby Ġre - est ablish ing Ġthe Ġconnection Ġwhen Ġthere Ġis Ġa Ġproblem ĊĠĠĠĠĠĠĠĠ Ġ* Ġand Ġchecking Ġif Ġthe Ġ[ max NumberOf Attempts ] Ġhas Ġbeen Ġex hausted . ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* ĠA Ġnegative Ġnumber Ġfor Ġ[ max NumberOf Attempts ] Ġmeans Ġan Ġun limited Ġnumber Ġof Ġretries Ġwill Ġbe Ġperformed . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" Throws Count ", Ġ" Complex Method ") ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġdo Invoke ( method : ĠMethod , Ġargs : ĠArray < out ĠAny > ?, Ġmax NumberOf Attempts : ĠInt ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck If Closed () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġremaining Attempts Ġ= Ġmax NumberOf Attempts ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġlast Exception : ĠThrowable ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( remaining Attempts Ġ!= Ġ 0 Ġ&& Ġ! reconnect ing RPC Connection . is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug Ġ{ Ġ" Inv oking ĠRPC Ġ$ method ..." Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġmethod . invoke ( reconnect ing RPC Connection . proxy , Ġ*( args Ġ?: Ġempty Array ())). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug Ġ{ Ġ" RPC Ġ$ method Ġinvoked Ġsuccessfully ." Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠInvocation TargetException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( method . name . equals (" shutdown ", Ġtrue )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Shutdown Ġinvoked , Ġstop Ġreconnect ing .", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect ing RPC Connection . notify Server And Close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( e . target Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠReject ed Command Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Node Ġis Ġbeing Ġshutdown . ĠOperation Ġ${ method . name } Ġrejected . ĠSh ut ting Ġdown ...", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe . target Exception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠConnection FailureException Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Failed Ġto Ġperform Ġoperation Ġ${ method . name }. ĠConnection Ġdropped . ĠRetry ing .... ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect ing RPC Connection . reconnect OnError ( e ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck If Is Start Flow ( method , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRPC Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre throw If Un recover able ( e . target Exception Ġas ĠRPC Exception ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Failed Ġto Ġperform Ġoperation Ġ${ method . name }. ĠRPC Exception . ĠRetry ing .... ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect ing RPC Connection . reconnect OnError ( e ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 1 0 0 0 ) Ġ// ĠTODO Ġ- Ġexplain Ġwhy Ġthis Ġsleep Ġis Ġnecessary ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck If Is Start Flow ( method , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPermission Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRPC Exception (" User Ġdoes Ġnot Ġhave Ġpermission Ġto Ġperform Ġoperation Ġ${ method . name }. ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Failed Ġto Ġperform Ġoperation Ġ${ method . name }. ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe . target Exception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Exception Ġ= Ġe . target Exception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġremaining Attempts -- ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( reconnect ing RPC Connection . is Closed ()) Ġreturn Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMax Rpc Retry Exception ( max NumberOf Attempts , Ġmethod , Ġlast Exception ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġcheck If Closed () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( reconnect ing RPC Connection . is Closed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRPC Exception (" Cannot Ġexecute ĠRPC Ġcommand Ġafter Ġclient Ġhas Ġshut Ġdown .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġre throw If Un recover able ( e : ĠRPC Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( e . cause Ġis ĠNot Serializable Exception ) Ġ{ Ġ// ĠDo Ġnot Ġtry Ġto Ġreconnect Ġwhen Ġwe Ġcan 't Ġserialize ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġinvoke ( proxy : ĠAny , Ġmethod : ĠMethod , Ġargs : ĠArray < out ĠAny >? ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( method . returnType ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠData Feed :: class . java Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠInter cept Ġthe Ġdata Ġfeed Ġmethods Ġand Ġreturn Ġa ĠRe connecting Observable Ġinstance ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinitial Feed : ĠData Feed < Any , ĠAny ?> Ġ= Ġunchecked Cast ( do Invoke ( method , Ġargs , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect ing RPC Connection . grace ful Re connect . max Attempts )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġobservable Ġ= ĠRe connecting Observable ( reconnect ing RPC Connection , Ġinitial Feed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThis Ġhandles Ġreconnect ing Ġand Ġcreates Ġnew Ġfeeds . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunchecked Cast ( this . invoke ( reconnect ing RPC Connection . proxy , Ġmethod , Ġargs )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Feed . copy ( updates Ġ= Ġobservable ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO Ġ- Ġadd Ġhandlers Ġfor ĠObservable Ġreturn Ġtypes . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġdo Invoke ( method , Ġargs , Ġreconnect ing RPC Connection . grace ful Re connect . max Attempts ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġretry Flows Pool . shutdown () ĊĠĠĠĠĠĠĠ Ġreconnect ing RPC Connection . force Close () ĊĠĠĠ Ġ} Ċ } Ċ