/******************************************************************************* Ċ Ġ* ĠCopyright Ġ 2 0 2 1 - 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 . lw dat apro vider Ċ Ċ import Ġcom . exact pro . c rad le . messages . Stored Message Filter Ċ import Ġcom . exact pro . c rad le . messages . Stored MessageId Ċ import Ġcom . exact pro . c rad le . test events . B atched Stored Test Event Metadata Ċ import Ġcom . exact pro . c rad le . test events . Stored Test EventId Ċ import Ġcom . exact pro . c rad le . test events . Stored Test Event Metadata Ċ import Ġcom . exact pro . th 2 . common . grpc . Connection ID Ċ import Ġcom . exact pro . th 2 . common . grpc . Message ID Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġio . prometheus . client . Gauge Ċ import Ġio . prometheus . client . Histogram Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . channels .* Ċ import Ġmu . Kotlin Logging Ċ import Ġjava . io . IOException Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . Executors Ċ import Ġkotlin . coroutines . coroutine Context Ċ import Ġkotlin . system . measure Time Millis Ċ Ċ private Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{ Ġ} Ċ Ċ suspend Ġfun ĠObjectMapper . asString Suspend ( data : ĠAny ?): ĠString Ġ{ ĊĠĠĠ Ġval Ġmapper Ġ= Ġthis ĊĊĠĠĠ Ġreturn Ġwith Context ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠ Ġmapper . writeValue AsString ( data ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun ĠStored Message Filter . convert ToString (): ĠString Ġ{ ĊĠĠĠ Ġval Ġfilter Ġ= Ġthis ĊĊĠĠĠ Ġreturn Ġ"( limit =${ filter . limit } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" direction =${ filter . direction ?. value } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" timestamp From =${ filter . timestamp From ?. value } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" timestamp To =${ filter . timestamp To ?. value } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" stream =${ filter . stream Name ?. value } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" index Value =${ filter . index ?. value } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" index Operation =${ filter . index ?. operation ?. name }" Ċ } Ċ Ċ suspend Ġfun Ġ< T > Ġlog Time ( methodName : ĠString , Ġlambda : Ġsuspend Ġ() Ġ-> ĠT ): ĠT ? Ġ{ ĊĠĠĠ Ġreturn Ġwith Context ( coroutine Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġresult : ĠT ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠ Ġmeasure Time Millis Ġ{ Ġresult Ġ= Ġlambda . invoke () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. also Ġ{ Ġlogger . debug Ġ{ Ġ" c rad le : Ġ$ methodName Ġtook Ġ${ it } ms " Ġ} Ġ} ĊĊĠĠĠĠĠĠĠ Ġresult ĊĠĠĠ Ġ} Ċ } Ċ Ċ data Ġclass ĠMetrics ( ĊĠĠĠ Ġprivate Ġval Ġhistogram Time : ĠHistogram , ĊĠĠĠ Ġprivate Ġval Ġgauge : ĠG auge Ċ ) Ġ{ ĊĊĠĠĠ Ġconstructor ( variable Name : ĠString , Ġdescription Name : ĠString ) Ġ: Ġthis ( ĊĠĠĠĠĠĠĠ Ġhistogram Time Ġ= ĠHistogram . build ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ variable Name }_ hist _ time ", Ġ" Time Ġof Ġ$ description Name " ĊĠĠĠĠĠĠĠ Ġ). buckets (. 0 0 5 , Ġ. 0 1 , Ġ. 0 2 5 , Ġ. 0 5 , Ġ. 0 7 5 , Ġ. 1 , Ġ. 2 5 , Ġ. 5 , Ġ. 7 5 , Ġ 1 . 0 , Ġ 2 . 5 , Ġ 5 . 0 , Ġ 7 . 5 , Ġ 1 0 . 0 , Ġ 2 5 . 0 , Ġ 5 0 . 0 , Ġ 7 5 . 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. register (), ĊĠĠĠĠĠĠĠ Ġgauge Ġ= ĠG auge . build ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ variable Name }_ gauge ", Ġ" Quantity Ġof Ġ$ description Name Ġusing ĠG auge " ĊĠĠĠĠĠĠĠ Ġ). register () ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġfun Ġstart Ob serve (): ĠHistogram . Timer Ġ{ ĊĠĠĠĠĠĠĠ Ġgauge . inc () ĊĠĠĠĠĠĠĠ Ġreturn Ġhistogram Time . start Timer () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstop Ob serve ( timer : ĠHistogram . Timer ) Ġ{ ĊĠĠĠĠĠĠĠ Ġgauge . dec () ĊĠĠĠĠĠĠĠ Ġtimer . observe Duration () ĊĠĠĠ Ġ} Ċ } Ċ Ċ suspend Ġfun Ġ< T > Ġlog Metrics ( metrics : ĠMetrics , Ġlambda : Ġsuspend Ġ() Ġ-> ĠT ): ĠT ? Ġ{ ĊĠĠĠ Ġreturn Ġwith Context ( coroutine Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtimer Ġ= Ġmetrics . start Ob serve () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlambda . invoke () ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmetrics . stop Ob serve ( timer ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġval Ġwriter Dispatcher Ġ= ĠExec utors . new Single Thread Executor (). as Coroutine Dispatcher () ĊĊ Ċ fun Ġmin Instant ( first : ĠInstant , Ġsecond : ĠInstant ): ĠInstant Ġ{ ĊĠĠĠ Ġreturn Ġif Ġ( first . is Before ( second )) Ġ{ ĊĠĠĠĠĠĠĠ Ġfirst ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġsecond ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun Ġmax Instant ( first : ĠInstant , Ġsecond : ĠInstant ): ĠInstant Ġ{ ĊĠĠĠ Ġreturn Ġif Ġ( first . is After ( second )) Ġ{ ĊĠĠĠĠĠĠĠ Ġfirst ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġsecond ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ fun ĠInstant . is Before OrEqual ( other : ĠInstant ): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġthis . is Before ( other ) Ġ|| Ġthis Ġ== Ġother Ċ } Ċ Ċ fun ĠInstant . is After OrEqual ( other : ĠInstant ): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġthis . is After ( other ) Ġ|| Ġthis Ġ== Ġother Ċ } ĊĊ Ċ suspend Ġfun Ġ< E > ĠReceive Channel < E >. receive Available (): ĠList < E > Ġ{ ĊĠĠĠ Ġval Ġall Messages Ġ= Ġmutable ListOf < E >() ĊĠĠĠ Ġall Messages . add ( receive ()) ĊĠĠĠ Ġvar Ġnext Ġ= Ġpoll () ĊĠĠĠ Ġwhile Ġ( next Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġall Messages . add ( next ) ĊĠĠĠĠĠĠĠ Ġnext Ġ= Ġpoll () ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn Ġall Messages Ċ } ĊĊ Ċ fun ĠStored Test Event Metadata . try To Get Test Events ( parent EventId : ĠStored Test EventId ? Ġ= Ġnull ): ĠCollection < B atched Stored Test Event Metadata >? Ġ{ ĊĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . batch Metadata ?. test Events ?. let Ġ{ Ġevents Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( parent EventId Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents . filter Ġ{ Ġit . parentId Ġ== Ġparent EventId Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . error ( e ) Ġ{ Ġ} ĊĠĠĠĠĠĠĠ Ġnull ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ fun ĠStored MessageId . convertTo Proto (): ĠMessage ID Ġ{ ĊĠĠĠ Ġreturn ĠMessage ID . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. set Sequence ( index ) ĊĠĠĠĠĠĠĠ Ġ. set Direction ( c rad le Direction To Grpc ( direction )) ĊĠĠĠĠĠĠĠ Ġ. setConnection Id ( Connection ID . newBuilder (). set Session Alias ( stream Name )) ĊĠĠĠĠĠĠĠ Ġ. build () Ċ } Ċ