package Ġio . app write . services Ċ Ċ import Ġio . app write . Client Ċ import Ġio . app write . exceptions . App write Exception Ċ import Ġio . app write . extensions . forEach Async Ċ import Ġio . app write . extensions . fromJson Ċ import Ġio . app write . extensions . json Cast Ċ import Ġio . app write . models .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . Dispatch ers . IO Ċ import Ġokhttp 3 . Request Ċ import Ġokhttp 3 . Response Ċ import Ġokhttp 3 . WebSocket Ċ import Ġokhttp 3 . WebSocket Listener Ċ import Ġokhttp 3 . internal . concurrent . Task Runner Ċ import Ġokhttp 3 . internal . ws . Real WebSocket Ċ import Ġjava . util .* Ċ import Ġandroid . util . Log Ċ import Ġkotlin . coroutines . Coroutine Context Ċ Ċ class ĠReal time ( client : ĠClient ) Ġ: ĠService ( client ), ĠCor outine Scope Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġjob Ġ= ĠJob () ĊĊĠĠĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context ĊĠĠĠĠĠĠĠ Ġget () Ġ= ĠDispatch ers . Main Ġ+ Ġjob ĊĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠTYPE _ ERROR Ġ= Ġ" error " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠTYPE _ EVENT Ġ= Ġ" event " ĊĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠDE BOUN CE _ M ILL IS Ġ= Ġ 1 L ĊĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġsocket : ĠReal WebSocket ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġactive Channels Ġ= Ġmutable Set Of < String >() ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġactive Subscriptions Ġ= Ġmutable Map Of < Int , ĠReal time Callback >() ĊĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġsub Call Depth Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġreconnect Attempts Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġsubscriptions Counter Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġreconnect Ġ= Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Socket () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( active Channels . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġquery Param Builder Ġ= ĠStringBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. append (" project =${ client . config [" project "] }") ĊĊĠĠĠĠĠĠĠ Ġactive Channels . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġquery Param Builder ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. append ("& channels [] =$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġrequest Ġ= ĠRequest . Builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. url ("${ client . end Point Realtime }/ realtime ?$ query Param Builder ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( socket Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose Socket () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsocket Ġ= ĠReal WebSocket ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtask Runner Ġ= ĠTask Runner . INSTANCE , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoriginal Request Ġ= Ġrequest , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlistener Ġ= ĠApp write WebSocket Listener (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrandom Ġ= ĠRandom (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġping Interval Millis Ġ= Ġclient . http . ping Interval Millis . to Long (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġextensions Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġminimum Def late Size Ġ= Ġclient . http . min WebSocket Message To Compress ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġsocket !!. connect ( client . http ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġclose Socket () Ġ{ ĊĠĠĠĠĠĠĠ Ġsocket ?. close ( Realtime Code . POLICY _ VI OL ATION . value , Ġnull ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Timeout () Ġ= Ġwhen Ġ{ ĊĠĠĠĠĠĠĠ Ġreconnect Attempts Ġ< Ġ 5 Ġ-> Ġ 1 0 0 0 L ĊĠĠĠĠĠĠĠ Ġreconnect Attempts Ġ< Ġ 1 5 Ġ-> Ġ 5 0 0 0 L ĊĠĠĠĠĠĠĠ Ġreconnect Attempts Ġ< Ġ 1 0 0 Ġ-> Ġ 1 0 0 0 0 L ĊĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ 6 0 0 0 0 L ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsubscribe ( ĊĠĠĠĠĠĠĠ Ġvar arg Ġchannels : ĠString , ĊĠĠĠĠĠĠĠ Ġcallback : Ġ( Realtime Response Event < Any >) Ġ-> ĠUnit , ĊĠĠĠ Ġ) Ġ= Ġsubscribe ( ĊĠĠĠĠĠĠĠ Ġchannels Ġ= Ġchannels , ĊĠĠĠĠĠĠĠ ĠAny :: class . java , ĊĠĠĠĠĠĠĠ Ġcallback ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġfun Ġ< T > Ġsubscribe ( ĊĠĠĠĠĠĠĠ Ġvar arg Ġchannels : ĠString , ĊĠĠĠĠĠĠĠ Ġpayload Type : ĠClass < T >, ĊĠĠĠĠĠĠĠ Ġcallback : Ġ( Realtime Response Event < T >) Ġ-> ĠUnit , ĊĠĠĠ Ġ): ĠReal time Subscription Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcounter Ġ= Ġsubscriptions Counter ++ ĊĊĠĠĠĠĠĠĠ Ġactive Channels . addAll ( channels ) ĊĠĠĠĠĠĠĠ Ġactive Subscriptions [ counter ] Ġ= ĠReal time Callback ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannels . toList (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpayload Type , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcallback Ġas Ġ( Realtime Response Event <* >) Ġ-> ĠUnit ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsub Call Depth ++ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( DE BOUN CE _ M ILL IS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sub Call Depth Ġ== Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Socket () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsub Call Depth -- ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn ĠReal time Subscription Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactive Subscriptions . remove ( counter ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclean Up (* channels ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Socket () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġclean Up ( var arg Ġchannels : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġactive Channels . removeAll Ġ{ Ġchannel Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! channels . contains ( channel )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ removeAll Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactive Subscriptions . values . none Ġ{ Ġcallback Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallback . channels . contains ( channel ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinner Ġclass ĠApp write WebSocket Listener Ġ: ĠWebSocket Listener () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Open ( web Socket : ĠWebSocket , Ġresponse : ĠResponse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . on Open ( web Socket , Ġresponse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect Attempts Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Message ( web Socket : ĠWebSocket , Ġtext : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . on Message ( web Socket , Ġtext ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġtext . fromJson < Realtime Response >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( message . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTYPE _ ERROR Ġ-> Ġhandle ResponseError ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTYPE _ EVENT Ġ-> Ġhandle Response Event ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġhandle ResponseError ( message : ĠReal time Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġmessage . data . json Cast < App write Exception >() ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġsuspend Ġfun Ġhandle Response Event ( message : ĠReal time Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Ġ= Ġmessage . data . json Cast < Realtime Response Event < Any >>() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event . channels . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! event . channels . any Ġ{ Ġactive Channels . contains ( it ) Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactive Subscriptions . values . forEach Async Ġ{ Ġsubscription Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event . channels . any Ġ{ Ġsubscription . channels . contains ( it ) Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . payload Ġ= Ġevent . payload . json Cast ( subscription . payload Class ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription . callback ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Closing ( web Socket : ĠWebSocket , Ġcode : ĠInt , Ġreason : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . on Closing ( web Socket , Ġcode , Ġreason ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! reconnect Ġ|| Ġcode Ġ== ĠReal time Code . POLICY _ VI OL ATION . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtimeout Ġ= Ġget Timeout () ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog . e ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis @ Realtime :: class . java . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Realtime Ġdisconnected . ĠRe - connecting Ġin Ġ${ timeout Ġ/ Ġ 1 0 0 0 } Ġseconds .", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠApp write Exception ( reason , Ġcode ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreconnect Attempts ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Socket () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( web Socket : ĠWebSocket , Ġt : ĠThrowable , Ġresponse : ĠResponse ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . on Failure ( web Socket , Ġt , Ġresponse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġt . printStackTrace () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }