/** Ċ Ġ* ĠCopyright Ġ 2 0 2 0 ĠEP AM ĠSystems Ċ Ġ* Ċ Ġ* Ġ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 Ġcom . ep am . drill . core . ws Ċ Ċ import Ġcom . ben ash er 4 4 . uuid .* Ċ import Ġcom . ep am . drill .* Ċ import Ġcom . ep am . drill . common .* Ċ import Ġcom . ep am . drill . logger .* Ċ import Ġcom . ep am . drill . transport .* Ċ import Ġcom . ep am . drill . transport . common . ws .* Ċ import Ġcom . ep am . drill . ws .* Ċ import Ġkotlinx . atomic fu .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . serialization .* Ċ import Ġkotlinx . serialization . protobuf .* Ċ import Ġkotlin . coroutines .* Ċ import Ġkotlin . time .* Ċ Ċ @ Shared Immutable Ċ private Ġval Ġws Logger Ġ= ĠLogging . logger (" Dr ill Websocket ") Ċ Ċ private Ġval Ġdispatcher Ġ= Ġnew Single Thread Context (" sender Ġcoroutine ") ĊĊ Ċ class ĠW s Socket Ġ: ĠCor outine Scope Ġ{ ĊĊĠĠĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context Ġ= Ġdispatcher ĊĊĠĠĠ Ġprivate Ġval ĠerrorMessage Ġ= Ġatomic ("") ĊĠĠĠ Ġprivate Ġval Ġis InstanceId Generated Ġ= Ġatomic ( false ) ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġtopic Register () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġconnect ( admin Url : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġurl Ġ= ĠURL ("$ admin Url / agent / attach ") ĊĠĠĠĠĠĠĠ Ġif Ġ( agent Config . instanceId . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis InstanceId Generated . update Ġ{ Ġtrue Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . debug Ġ{ Ġ" InstanceId Ġwill Ġbe Ġgenerated Ġon Ġeach ĠWS Ġconnection " Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġcheck And Generate InstanceId () ĊĠĠĠĠĠĠĠ Ġheaders Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmutable Map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAgent Config Param Ġto ĠProto Buf . encode To HexString ( Agent Config . serializer (), Ġagent Config ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠNeed Sync Param Ġto Ġp storage . none (). toString (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttp Headers . Content Encoding Ġto Ġ" def late " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġws Logger . info Ġ{ Ġ" connecting Ġwith Ġinstance Id Ġ'${ agent Config . instanceId }' ..." Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġws Client Ġ= ĠWS Client Factory . create Client ( url ) ĊĠĠĠĠĠĠĠ Ġws . value Ġ= Ġws Client ĊĠĠĠĠĠĠĠ Ġws Client . on Open Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . info Ġ{ Ġ" Agent Ġconnected Ġwith Ġinstance Id Ġ'${ agent Config . instanceId } '" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠerrorMessage . update Ġ{ Ġ"" Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġws Client . on Binary Message Ġ{ Ġraw Message Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġraw Message . to Ws Message () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdestination Ġ= Ġmessage . destination ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= ĠW s Router [ destination ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( topic Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( topic ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPlugin Topic Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġplugin Metadata Ġ= ĠProto Buf . decode From ByteArray ( Plugin Binary . serializer (), Ġmessage . data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġduration Ġ= Ġmeasure Time Ġ{ Ġtopic . block ( plugin Metadata . meta , Ġplugin Metadata . data ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . debug Ġ{ Ġ"' $ destination ' Ġtook Ġ$ duration " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSender . send ( Message ( MessageType . MESSAGE _ DE LI VER ED , Ġ"/ agent / load ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInfo Topic Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġduration Ġ= Ġmeasure Time Ġ{ Ġtopic . run ( message . data ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . debug Ġ{ Ġ"' $ destination ' Ġtook Ġ$ duration " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSender . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . MESSAGE _ DE LI VER ED , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdestination ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠGeneric Topic < *> Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġduration Ġ= Ġmeasure Time Ġ{ Ġtopic . deserialize And Run ( message . data ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . debug Ġ{ Ġ"' $ destination ' Ġtook Ġ$ duration " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSender . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . MESSAGE _ DE LI VER ED , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdestination ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . warn Ġ{ Ġ" topic Ġwith Ġname Ġ'$ destination ' Ġdidn 't Ġregister " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġws Client . onError Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( errorMessage . value Ġ!= Ġmessage ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . error Ġ{ Ġ"[ Duplicates Ġin Ġdebug ] ĠWS Ġerror : Ġ$ message " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠerrorMessage . update Ġ{ Ġmessage Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . debug Ġ{ Ġ" WS Ġerror : Ġ$ message " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġws Client . onClose Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck And Generate InstanceId () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġws Logger . info Ġ{ Ġ" Websocket Ġclosed . ĠOn Ġnext Ġconnection Ġinstance Id Ġwill Ġbe Ġ'${ agent Config . instanceId } '" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠerrorMessage . update Ġ{ Ġ"" Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcheck And Generate InstanceId () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( is InstanceId Generated . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġagent Config Ġ= Ġagent Config . copy ( instanceId Ġ= Ġuuid 4 (). toString ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ private Ġfun ĠByteArray . to Ws Message () Ġ= ĠProto Buf . decode From ByteArray ( Message . serializer (), Ġthis ) Ċ