package Ġpl . ge ostream ing . r store . vertx . websocket Ċ Ċ import Ġio . vertx . core . Vert x Ċ import Ġio . vertx . core . buffer . Buffer Ċ import Ġio . vertx . core . http . HttpClient Options Ċ import Ġkotlinx . coroutines . experimental . async Ċ import Ġkotlinx . coroutines . experimental . channels . Channel Ċ import Ġkotlinx . coroutines . experimental . channels . consume Each Ċ import Ġkotlinx . coroutines . experimental . launch Ċ import Ġkotlinx . coroutines . experimental . new Single Thread Context Ċ import Ġpl . ge ostream ing . r store . core . channels . Chan ne led Remote Ċ import Ġpl . ge ostream ing . r store . core . msg . Rs Op Req Ċ import Ġpl . ge ostream ing . r store . core . msg . Rs Op Resp Ċ import Ġkotlin . coroutines . experimental . Coroutine Context Ċ Ċ /** Ċ Ġ* ĠCreated Ġby Ġl k ole k Ġon Ġ 0 1 . 0 7 . 2 0 1 7 . Ċ Ġ*/ Ċ class ĠRemote Vertex Websocket Replica ( ĊĠĠĠĠĠĠĠ Ġval Ġvert x : ĠVert x , ĊĠĠĠĠĠĠĠ Ġval Ġhost : String , ĊĠĠĠĠĠĠĠ Ġval Ġport : Int , ĊĊĠĠĠĠĠĠĠ Ġrepl Id : Int , ĊĠĠĠĠĠĠĠ Ġcontext : ĠCor outine Context Ġ= Ġnew Single Thread Context (" remote ĠCh ann Ġr ${ repl Id }") Ċ ): ĠCh an ne led Remote ( ĊĠĠĠĠĠĠĠ Ġrepl Id , ĊĠĠĠĠĠĠĠ ĠChannel < Rs Op Req >( 1 0 ), ĊĠĠĠĠĠĠĠ ĠChannel < Rs Op Resp >( 1 0 ), ĊĠĠĠĠĠĠĠ Ġcontext Ċ ) Ċ { ĊĠĠĠ Ġval Ġin b Ġ= Ġin box Remote Ġas ĠChannel < Rs Op Req >; ĊĠĠĠ Ġval Ġout b Ġ= Ġout box Remote Ġas ĠChannel < Rs Op Resp >; ĊĠĠĠ Ġprivate Ġval Ġbase Uri Ġ= Ġ"/ api / r store "; ĊĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfst Cfg Ġ= Ġfst Binary . fst Cfg ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġhttp Cl Ġ= Ġ Ġvert x . create HttpClient ( HttpClient Options (). apply Ġ{ ĊĠĠĠĠĠĠĠ Ġdefault Host Ġ= Ġhost ĊĠĠĠĠĠĠĠ Ġdefault Port Ġ= Ġport ĊĠĠĠĠĠĠĠ Ġ// ĠĠĠĠĠĠĠĠĠĠĠ Ġlog Activity = true ĊĠĠĠĠĠĠĠ Ġmax PoolSize Ġ= Ġ 5 ĊĠĠĠĠĠĠĠ Ġmax Websocket Message Size Ġ= Ġmax Websocket Frame Size Ġ* Ġ 1 6 * 1 0 0 ĊĠĠĠ Ġ}) ĊĊĠĠĠ Ġinit { ĊĠĠĠĠĠĠĠ Ġhttp Cl . websocket ( base Uri Ġ+ Ġ"/ channel ") Ġ{ Ġsocket Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("/ channel Ġconnected "); ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . binary MessageHandler Ġ{ Ġx Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" binary Ġmsg Ġsize =${ x . length () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġo pr Ġ= Ġfst Cfg . as Object ( x . bytes ) Ġas ? ĠRs Op Resp ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( op r Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġasync ( context ) Ġ{ Ġout b . send ( op r ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" !!! Ġcant Ġdeserialize !") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} catch Ġ( ex : Exception ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġex . printStackTrace () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( context ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġin b . consume Each Ġ{ Ġop Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġb Ġ= Ġfst Cfg . as ByteArray ( op ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . write Binary Message ( Buffer . buffer ( b )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} Ċ Ċ }