/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 - 2 0 2 1 ĠExact pro Ġ( Exact pro ĠSystems ĠLimited ) Ċ Ġ* Ċ Ġ* Ġ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 . exact pro . th 2 . act . ssh . messages Ċ Ċ import Ġcom . exact pro . th 2 . act . ssh . cfg . Execution Ċ import Ġcom . exact pro . th 2 . act . ssh . cfg . Publication Configuration Ċ import Ġcom . exact pro . th 2 . common . grpc . Direction Ċ import Ġcom . exact pro . th 2 . common . grpc . Message ID Ċ import Ġcom . exact pro . th 2 . common . grpc . Raw Message Ċ import Ġcom . exact pro . th 2 . common . grpc . Raw Message Batch Ċ import Ġcom . exact pro . th 2 . common . message . to Timestamp Ċ import Ġcom . exact pro . th 2 . common . schema . message . Message Router Ċ import Ġcom . exact pro . th 2 . common . schema . message . Queue Attribute Ċ import Ġcom . google . protobuf . ByteString Ċ import Ġmu . Kotlin Logging Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġjavax . annotation . concurrent . Thread Safe Ċ Ċ @ Thread Safe Ċ class ĠMessage Publisher ( ĊĠĠĠ Ġprivate Ġval Ġrouter : ĠMessage Router < Raw Message Batch >, ĊĠĠĠ Ġprivate Ġval Ġdefault Configuration : ĠPublic ation Configuration , Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġclass ĠSession Info Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġsequence : ĠLong Ġ= ĠInstant . now (). run Ġ{ Ġepoch Second Ġ* ĠN AN OS _ IN _ SECONDS Ġ+ Ġnano Ġ} ĊĠĠĠĠĠĠĠ Ġfun Ġget And Increment (): ĠLong Ġ= Ġsequence ++ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġsequences By Alias : ĠMutable Map < String , ĠSession Info > Ġ= ĠConcurrent HashMap () ĊĊĠĠĠ Ġfun Ġpublish ( ĊĠĠĠĠĠĠĠ Ġoutput : ĠString , ĊĠĠĠĠĠĠĠ Ġexecution : ĠExecution , ĊĠĠĠĠĠĠĠ Ġparameters : ĠMap < String , ĠString >, ĊĠĠĠĠĠĠĠ Ġalias : ĠString ĊĠĠĠ Ġ): ĠMessage ID ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcfg Ġ= Ġexecution . message Publication Ġ?: Ġdefault Configuration ĊĠĠĠĠĠĠĠ Ġif Ġ(! cfg . enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info Ġ{ Ġ" Skip Ġoutput Ġpublication Ġfor Ġcommand Ġ${ execution . alias }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġinfo Ġ= Ġget Info For Alias ( alias ) ĊĠĠĠĠĠĠĠ Ġreturn Ġrun Catch ing Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuilder Ġ= ĠRaw Message . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Body ( ByteString . copyFrom ( output , ĠChar sets . UTF _ 8 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( info ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġbuilder . fill Metadata ( info , Ġexecution , Ġparameters , Ġalias ). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouter . send All ( Raw Message Batch . newBuilder (). add Messages ( message ). build (), ĠQueue Attribute . FIRST . value ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage . metadata . id ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. on Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info Ġ{ Ġ" Output Ġfor Ġcommand Ġ${ execution . alias } Ġwith Ġparameters Ġ$ parameters Ġwas Ġpublished Ġunder Ġsession Ġalias Ġ$ alias " Ġ} ĊĠĠĠĠĠĠĠ Ġ}. on Failure Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error ( it ) Ġ{ Ġ" Cannot Ġpublish Ġoutput Ġfor Ġcommand Ġ${ execution . alias } Ġwith Ġparameters Ġ$ parameters " Ġ} ĊĠĠĠĠĠĠĠ Ġ}. get OrNull () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠRaw Message . Builder . fill Metadata ( ĊĠĠĠĠĠĠĠ Ġinfo : ĠSession Info , ĊĠĠĠĠĠĠĠ Ġexecution : ĠExecution , ĊĠĠĠĠĠĠĠ Ġparameters : ĠMap < String , ĠString >, ĊĠĠĠĠĠĠĠ Ġalias : ĠString , ĊĠĠĠ Ġ) Ġ= Ġapply Ġ{ ĊĠĠĠĠĠĠĠ Ġmetadata Builder . apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġid Builder . apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdirection Ġ= ĠDirection . FIRST ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Ġ= Ġinfo . get And Increment () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection Id Builder . session Alias Ġ= Ġalias ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtimestamp Ġ= ĠInstant . now (). to Timestamp () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput Properties ( EXEC UTION _ ALIAS _ PARAMETER , Ġexecution . alias ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput All Properties ( parameters ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Info For Alias ( alias : ĠString ): ĠSession Info Ġ= Ġsequences By Alias . compute If Absent ( alias ) Ġ{ ĠSession Info () Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠEXEC UTION _ ALIAS _ PARAMETER Ġ= Ġ" act . ssh . execution - alias " ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠN AN OS _ IN _ SECONDS : ĠLong Ġ= ĠTimeUnit . SECONDS . to Nan os ( 1 ) ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠLOGGER Ġ= ĠKotlin Logging . logger Ġ{ Ġ} ĊĠĠĠ Ġ} Ċ }