/* Ċ Ġ* Ġ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 . check 1 . rule Ċ Ċ import Ġcom . exact pro . sf . common . messages . I Message Ċ import Ġcom . exact pro . sf . comparison . Comparator Settings Ċ import Ġcom . exact pro . sf . comparison . Comparison Result Ċ import Ġcom . exact pro . sf . comparison . Message Comparator Ċ import Ġcom . exact pro . sf . scri ptr un ner . StatusType Ċ import Ġcom . exact pro . th 2 . check 1 . Abstract Session Observer Ċ import Ġcom . exact pro . th 2 . check 1 . Session Key Ċ import Ġcom . exact pro . th 2 . check 1 . Stream Container Ċ import Ġcom . exact pro . th 2 . check 1 . event . bean . builder . Verification Builder Ċ import Ġcom . exact pro . th 2 . check 1 . exception . Rule Internal Exception Ċ import Ġcom . exact pro . th 2 . check 1 . metrics . Rule Metric Ċ import Ġcom . exact pro . th 2 . check 1 . util . Verification Util Ċ import Ġcom . exact pro . th 2 . common . event . Event Ċ import Ġcom . exact pro . th 2 . common . event . Event . Status . FAILED Ċ import Ġcom . exact pro . th 2 . common . event . Event . Status . PAS SED Ċ import Ġcom . exact pro . th 2 . common . event . Event Utils Ċ import Ġcom . exact pro . th 2 . common . grpc . Checkpoint Ċ import Ġcom . exact pro . th 2 . common . grpc . Event Batch Ċ import Ġcom . exact pro . th 2 . common . grpc . Event ID Ċ import Ġcom . exact pro . th 2 . common . grpc . Message Ċ import Ġcom . exact pro . th 2 . common . grpc . Message Filter Ċ import Ġcom . exact pro . th 2 . common . grpc . Message Metadata Ċ import Ġcom . exact pro . th 2 . common . grpc . Metadata Filter Ċ import Ġcom . exact pro . th 2 . common . grpc . Root Message Filter Ċ import Ġcom . exact pro . th 2 . common . message . toJson Ċ import Ġcom . exact pro . th 2 . common . message . to Readable Body Collection Ċ import Ġcom . exact pro . th 2 . common . schema . message . Message Router Ċ import Ġcom . exact pro . th 2 . s ail fish . utils . Proto To I Message Converter Ċ import Ġcom . google . protobuf . Text Format . short Debug String Ċ import Ġio . reactivex . Observable Ċ import Ġio . reactivex . Single Ċ import Ġio . reactivex . dis pos ables . Disposable Ċ import Ġio . reactivex . schedul ers . Schedulers Ċ import Ġio . reactivex . subjects . Single Subject Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . ExecutorService Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . Fork Join Pool Ċ import Ġjava . util . concurrent . TimeUnit . MILLISECONDS Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ /** Ċ Ġ* ĠImplements Ġcommon Ġlogic Ġfor Ġcheck Ġtask . Ċ Ġ* Ġ@ param Ġmax Event Batch Content Size Ġmax Ġsize Ġin Ġbytes Ġof Ġsummary Ġevents Ġcontent Ġin Ġa Ġbatch Ċ Ġ* Ġ** Class Ġin Ġnot Ġthread - safe ** Ċ Ġ*/ Ċ abstract Ġclass ĠAbstract Check Task ( ĊĠĠĠ Ġval Ġdescription : ĠString ?, ĊĠĠĠ Ġprivate Ġval Ġtimeout : ĠLong , ĊĠĠĠ Ġprivate Ġval Ġmax Event Batch Content Size : ĠInt , ĊĠĠĠ Ġsubmit Time : ĠInstant , ĊĠĠĠ Ġprotected Ġval Ġsession Key : ĠSession Key , ĊĠĠĠ Ġprivate Ġval Ġparent Event ID : ĠEvent ID , ĊĠĠĠ Ġprivate Ġval Ġmessage Stream : ĠObservable < Stream Container >, ĊĠĠĠ Ġprivate Ġval Ġevent Batch Router : ĠMessage Router < Event Batch > Ċ ) Ġ: ĠAbstract Session Observer < Message Container >() Ġ{ ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire ( max Event Batch Content Size Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"' max Event Batch Content Size ' Ġshould Ġbe Ġgreater Ġthan Ġzero , Ġactual : Ġ$ max Event Batch Content Size " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġrequire ( timeout Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"' timeout ' Ġshould Ġbe Ġset Ġor Ġbe Ġgreater Ġthan Ġzero , Ġactual : Ġ$ timeout " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġvar Ġhandled Message Counter : ĠLong Ġ= Ġ 0 ĊĊĠĠĠ Ġprotected Ġval Ġconverter Ġ= ĠProto To I Message Converter ( Verification Util . FACTORY _ PROXY , Ġnull , Ġnull ) ĊĠĠĠ Ġprotected Ġval Ġroot Event : ĠEvent Ġ= ĠEvent . from ( submit Time ). description ( description ) ĊĊĠĠĠ Ġprivate Ġval Ġsequence Subject Ġ= ĠSingle Subject . create < Legacy >() ĊĠĠĠ Ġprivate Ġval Ġhas Next Task Ġ= ĠAtomic Boolean ( false ) ĊĠĠĠ Ġprivate Ġval Ġtask State Ġ= ĠAtomic Reference ( State . CREATED ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠUsed Ġfor Ġobserve Ġmessages Ġin Ġone Ġthread . ĊĠĠĠĠ Ġ* ĠIt Ġprovides Ġthe Ġfeature Ġto Ġwrite Ġno Ġthread - safe Ġcode Ġin Ġchildren Ġclasses . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠExecutor Ġis Ġshared Ġbetween Ġconnected Ġtasks . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġlateinit Ġvar ĠexecutorService : ĠExecutor Service ĊĠĠĠ Ġprivate Ġvar Ġ_ endTime : ĠInstant ? Ġ= Ġnull ĊĠĠĠ Ġval ĠendTime : ĠInstant ? ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġ_ endTime ĊĊĠĠĠ Ġprotected Ġenum Ġclass ĠState Ġ{ ĊĠĠĠĠĠĠĠ ĠCRE ATED , ĊĠĠĠĠĠĠĠ ĠBEGIN , ĊĠĠĠĠĠĠĠ ĠTIME OUT , ĊĠĠĠĠĠĠĠ ĠCOMP LETED , ĊĠĠĠĠĠĠĠ ĠP UB L ISHED ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġend Future : ĠDis posable ĊĊĠĠĠ Ġprivate Ġvar Ġlast Sequence Ġ= ĠDEFAULT _ SEQUENCE ĊĊĠĠĠ Ġoverride Ġfun ĠonStart () Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . onStart () ĊĊĠĠĠĠĠĠĠ Ġ// Init Ġor Ġre - init Ġvariable Ġin ĠTASK _ SCHEDUL ER Ġthread ĊĠĠĠĠĠĠĠ Ġhandled Message Counter Ġ= Ġ 0 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . onError ( e ) ĊĊĠĠĠĠĠĠĠ Ġroot Event . status ( FAILED ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( Event Utils . create Message Bean ( e . message )) ĊĠĠĠĠĠĠĠ Ġend (" Error Ġ${ e . message } Ġreceived Ġin Ġmessage Ġstream ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠShutdown Ġthe Ġexecutor Ġthat Ġis Ġused Ġto Ġperform Ġthis Ġtask Ġin Ġcase Ġit Ġdoesn 't Ġhave Ġa Ġnext Ġtask ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġtask Ġdoesn 't Ġhave Ġa Ġnext Ġtask , Ġotherwise Ġit Ġwill Ġreturn Ġfalse ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġtry Shutdown Executor (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( hasNext Task . get ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . warn (" Cannot Ġshutdown Ġexecutor Ġfor Ġtask Ġ'$ description ' Ġthat Ġhas Ġa Ġconnected Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠexecutorService . shutdown () ĊĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠRegisters Ġa Ġtask Ġas Ġthe Ġnext Ġtask Ġin Ġthe Ġcontinuous Ġverification Ġchain . ĠIts Ġ[ begin ] Ġmethod Ġwill Ġbe Ġcalled ĊĠĠĠĠ Ġ* Ġwhen Ġthe Ġcurrent Ġtask Ġcompletes Ġits Ġcheck Ġor Ġif Ġthe Ġtimeout Ġis Ġover Ġfor Ġit . ĊĠĠĠĠ Ġ* ĠThe Ġscheduler Ġfor Ġthe Ġcurrent Ġtask Ġwill Ġbe Ġpassed Ġto Ġthe Ġnext Ġtask . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġmethod Ġshould Ġbe Ġcalled Ġonly Ġonce Ġotherwise Ġit Ġthrows ĠIllegalStateException . ĊĠĠĠĠ Ġ* Ġ@ throws ĠIllegalStateException Ġwhen Ġmethod Ġis Ġcalled Ġmore Ġthan Ġonce . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsubscribe Next Task ( check Task : ĠAbstract Check Task ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( hasNext Task . compare AndSet ( false , Ġtrue )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Subject . subscribe Ġ{ Ġlegacy Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexecutor Ġ= Ġif Ġ( legacy . executor Service . is Shutdown ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . warn (" Executor Ġhas Ġbeen Ġshutdown Ġbefore Ġnext Ġtask Ġhas Ġbeen Ġsubscribed . ĠCreate Ġa Ġnew Ġone ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate ExecutorService () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlegacy . executor Service ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Task . begin ( legacy . last Sequence , Ġexecutor ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Task Ġ{} Ġ({ }) Ġsubscribed Ġto Ġtask Ġ{} Ġ({ })", Ġcheck Task . description , Ġcheck Task . hashCode (), Ġdescription , ĠhashCode ()) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Subscription Ġto Ġlast Ġsequence Ġfor Ġtask Ġ$ description Ġ(${ hashCode () }) Ġis Ġalready Ġexecuted , Ġsubscriber Ġ${ check Task . description } Ġ(${ check Task . hashCode () }) ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠOb serve Ġa Ġmessage Ġsequence Ġfrom Ġthe Ġcheckpoint . ĊĠĠĠĠ Ġ* ĠTask Ġsubscribe Ġto Ġmessages Ġstream Ġwith Ġits Ġsequence Ġafter Ġcall . ĊĠĠĠĠ Ġ* ĠThis Ġmethod Ġshould Ġbe Ġcalled Ġonly Ġonce Ġotherwise Ġit Ġthrows ĠIllegalStateException . ĊĠĠĠĠ Ġ* Ġ@ param Ġcheckpoint Ġmessage Ġsequence Ġfrom Ġprevious Ġtask . ĊĠĠĠĠ Ġ* Ġ@ throws ĠIllegalStateException Ġwhen Ġmethod Ġis Ġcalled Ġmore Ġthan Ġonce . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġbegin ( checkpoint : ĠCheck point ? Ġ= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġbegin ( checkpoint ?. get Sequence ( session Key ) Ġ?: ĠDEFAULT _ SEQUENCE ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠIt Ġis Ġcalled Ġwhen Ġthe Ġtimeout Ġis Ġover Ġand Ġthe Ġtask Ġis Ġnot Ġcomplete Ġyet ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprotected Ġopen Ġfun Ġon Timeout () Ġ{} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠMark s Ġthe Ġtask Ġas Ġsuccessfully Ġcompleted . ĠIf Ġthe Ġtask Ġtimeout Ġhad Ġbeen Ġexited Ġand Ġthen Ġthe Ġtask Ġwas Ġmarked Ġas Ġsuccessfully Ġcompleted ĊĠĠĠĠ Ġ* Ġthe Ġtask Ġwill Ġbe Ġconsidered Ġas Ġsuccessfully Ġcompleted Ġbecause Ġit Ġhad Ġactually Ġfound Ġthat Ġit Ġshould ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprotected Ġfun Ġcheck Complete () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠLOGGER . info (" Check Ġcompleted Ġfor Ġsession Ġalias Ġ'{}' Ġwith Ġsequence Ġ'{} '", Ġsession Key , Ġlast Sequence ) ĊĠĠĠĠĠĠĠ Ġval Ġprev Value Ġ= Ġtask State . get AndSet ( State . COMPLETED ) ĊĠĠĠĠĠĠĠ Ġdispose () ĊĠĠĠĠĠĠĠ Ġend Future . dispose () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( prev Value Ġ== ĠState . TIMEOUT ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Task Ġ'{}' Ġfor Ġsession Ġalias Ġ'{}' Ġis Ġcompleted Ġright Ġafter Ġtimeout Ġexited . ĠConsider Ġit Ġas Ġcompleted ", Ġdescription , Ġsession Key ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Task Ġ'{}' Ġfor Ġsession Ġalias Ġ'{}' Ġis Ġcompleted Ġnormally ", Ġdescription , Ġsession Key ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠProvides Ġa Ġfeature Ġto Ġdefine Ġcustom Ġfilter Ġfor Ġobserve . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprotected Ġopen Ġfun ĠObservable < Message Container >. task Pipeline () Ġ: ĠObservable < Message Container > Ġ= Ġthis ĊĊĠĠĠ Ġprotected Ġabstract Ġfun Ġname (): ĠString ĊĠĠĠ Ġprotected Ġabstract Ġfun Ġtype (): ĠString ĊĠĠĠ Ġprotected Ġabstract Ġfun Ġsetup ( root Event : ĠEvent ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠOb serve Ġa Ġmessage Ġsequence Ġfrom Ġthe Ġprevious Ġtask . ĊĠĠĠĠ Ġ* ĠTask Ġsubscribe Ġto Ġmessages Ġstream Ġwith Ġsequence Ġafter Ġcall . ĊĠĠĠĠ Ġ* ĠThis Ġmethod Ġshould Ġbe Ġcalled Ġonly Ġonce Ġotherwise Ġit Ġthrows ĠIllegalStateException . ĊĠĠĠĠ Ġ* Ġ@ param Ġsequence Ġmessage Ġsequence Ġfrom Ġthe Ġprevious Ġtask . ĊĠĠĠĠ Ġ* Ġ@ param ĠexecutorService Ġexecutor Ġto Ġschedule Ġpipeline Ġexecution . ĊĠĠĠĠ Ġ* Ġ@ throws ĠIllegalStateException Ġwhen Ġmethod Ġis Ġcalled Ġmore Ġthan Ġonce . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġbegin ( sequence : ĠLong Ġ= ĠDEFAULT _ SEQUENCE , ĠexecutorService : ĠExecutor Service Ġ= Ġcreate ExecutorService ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġconfigure Root Event () ĊĠĠĠĠĠĠĠ Ġif Ġ(! task State . compare AndSet ( State . CREATED , ĠState . BEGIN )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Task Ġ$ description Ġalready Ġhas Ġbeen Ġstarted ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠLOGGER . info (" Check Ġbegin Ġfor Ġsession Ġalias Ġ'{}' Ġwith Ġsequence Ġ'{}' Ġtimeout Ġ'{} '", Ġsession Key , Ġsequence , Ġtimeout ) ĊĠĠĠĠĠĠĠ ĠRule Metric . increment Active Rule ( type ()) ĊĠĠĠĠĠĠĠ Ġthis . last Sequence Ġ= Ġsequence ĊĠĠĠĠĠĠĠ Ġthis . executor Service Ġ= ĠexecutorService ĊĠĠĠĠĠĠĠ Ġval Ġscheduler Ġ= ĠS chedulers . from ( executor Service ) ĊĊĠĠĠĠĠĠĠ Ġend Future Ġ= ĠSingle . timer ( timeout , ĠMIL LI SECONDS , ĠS chedulers . computation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġ_ Ġ-> Ġend (" Timeout Ġis Ġexited ") Ġ} ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Stream . observe On ( scheduler ) Ġ// ĠDefined Ġscheduler Ġto Ġexecution Ġin Ġone Ġthread Ġto Ġavoid Ġrace - condition . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do Finally ( this :: task Finished ) Ġ// Ġwill Ġbe Ġexecuted Ġif Ġthe Ġsource Ġis Ġcomplete Ġor Ġan Ġerror Ġreceived Ġor Ġthe Ġtimeout Ġis Ġexited . ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAll Ġsources Ġabove Ġwill Ġbe Ġdisposed Ġon Ġthis Ġscheduler . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThis Ġmethod Ġshould Ġbe Ġcalled Ġas Ġcloser Ġas Ġpossible ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġto Ġthe Ġactual Ġdispose Ġthat Ġyou Ġwant Ġto Ġexecute Ġon Ġthis Ġscheduler ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbecause Ġother Ġoperations Ġare Ġexecuted Ġon Ġthe Ġsame Ġsingle - thread Ġscheduler . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġwe Ġmove Ġ[ Observable # unsubscribe On ] Ġafter Ġthem , Ġthey Ġwon 't Ġbe Ġdisposed Ġuntil Ġthe Ġscheduler Ġis Ġfree . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIn Ġthe Ġworst - case Ġscenario , Ġit Ġmight Ġnever Ġhappen . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. unsubscribe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. continue Ob serve ( session Key , Ġsequence ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandled Message Counter ++ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( it . metadata . id ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġroot Event . message ID ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map To Message Container () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. task Pipeline () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( this ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error (" An Ġinternal Ġerror Ġoccurred Ġwhile Ġexecuting Ġrule ", Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġroot Event . add Sub Event With Same Period () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" An Ġerror Ġoccurred Ġwhile Ġexecuting Ġrule ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" internal Error ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( FAILED ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exception ( exception , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtask Finished () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRule Internal Exception (" An Ġinternal Ġerror Ġoccurred Ġwhile Ġexecuting Ġrule ", Ġexception ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġtask Finished () Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcurrent State Ġ= Ġtask State . get () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Fin ishes Ġtask Ġ'$ description ' Ġin Ġstate Ġ$ currentState ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( currentState Ġ== ĠState . TIMEOUT ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall On Timeout Callback () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpublish Event () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Task Ġ'$ description ' Ġhas Ġbeen Ġfinished ") ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġ" Cannot Ġfinish Ġtask Ġ'$ description '" ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error ( message , Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Batch Router . send ( Event Batch . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setParent EventId ( parent Event ID ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. add Events ( Event . start () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" Check Ġrule Ġ$ description Ġproblem ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" Exception ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( FAILED ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( Event Utils . create Message Bean ( message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( Event Utils . create Message Bean ( ex . message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Proto ( parent Event ID )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build ()) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRule Metric . decrement Active Rule ( type ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Subject . on Success ( Legacy ( executor Service , Ġlast Sequence )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcall On Timeout Callback () Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Timeout () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error (" Cannot Ġexecute Ġ' on Timeout ' Ġmethod ", Ġex ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate ExecutorService (): ĠExecutor Service Ġ= ĠExec utors . new Single Thread Executor () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDis poses Ġthe Ġtask Ġwhen Ġthe Ġtimeout Ġis Ġover Ġor Ġthe Ġmessage Ġstream Ġis Ġcompleted Ġnormally Ġor Ġwith Ġan Ġexception . ĊĠĠĠĠ Ġ* ĠTask Ġunsubscribe Ġfrom Ġthe Ġmessage Ġstream . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġreason Ġthe Ġcause Ġwhy Ġa Ġtask Ġmust Ġbe Ġstopped ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġend ( reason : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( task State . compare AndSet ( State . BEGIN , ĠState . TIMEOUT )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Stop Ġtask Ġfor Ġsession Ġalias Ġ'{}' Ġwith Ġsequence Ġ'{}' Ġbecause : Ġ{}", Ġsession Key , Ġlast Sequence , Ġreason ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdispose () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġend Future . dispose () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Task Ġfor Ġsession Ġalias Ġ'{}' Ġis Ġalready Ġcompleted . ĠIgnore Ġ' end ' Ġmethod Ġcall Ġwith Ġreason : Ġ{}", Ġsession Key , Ġreason ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . onComplete () ĊĠĠĠĠĠĠĠ Ġend (" Message Ġstream Ġis Ġcompleted ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠPrepare Ġthe Ġroot Ġevent Ġor Ġchildren Ġevents Ġfor Ġpublication . ĊĠĠĠĠ Ġ* ĠThis Ġmethod Ġis Ġinvoked Ġin Ġ[ State . PUBL ISHED ] Ġstate . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprotected Ġopen Ġfun Ġcomplete Event ( c anceled : ĠBoolean ) Ġ{} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠPub lishes Ġthe Ġevent Ġto Ġ[ event Batch Router ]. ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġpublish Event () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprev State Ġ= Ġtask State . get AndSet ( State . PUBL ISHED ) ĊĠĠĠĠĠĠĠ Ġif Ġ( prev State Ġ!= ĠState . PUBL ISHED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Event Or Report Error ( prev State ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ_ endTime Ġ= ĠInstant . now () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbatches Ġ= Ġroot Event . dis per se To B atches ( max Event Batch Content Size , Ġparent Event ID ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠRESP ONSE _ EXECUT OR . execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbatches . forEach Ġ{ Ġbatch Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Sending Ġevent Ġbatch Ġparent Ġid Ġ'{} '", Ġparent Event ID . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Batch Router . send ( batch ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( LOGGER . isDebugEnabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Sent Ġevent Ġbatch Ġ'{} '", Ġshort Debug String ( batch )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error (" Can Ġnot Ġsend Ġevent Ġbatch Ġ'{} '", Ġshort Debug String ( batch ), Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Event Ġtree Ġid Ġ'{}' Ġparent Ġid Ġ'{}' Ġis Ġalready Ġpublished ", Ġroot Event . id , Ġparent Event ID ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcomplete Event Or Report Error ( prev State : ĠState ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Event ( prev State Ġ== ĠState . TIMEOUT ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error (" Result Ġevent Ġcannot Ġbe Ġcompleted ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġroot Event . add Sub Event With Same Period () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" Check Ġresult Ġevent Ġcannot Ġbuild Ġcompletely ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" event Not Complete ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( Event Utils . create Message Bean (" An Ġunexpected Ġexception Ġhas Ġbeen Ġthrown Ġduring Ġresult Ġcheck Ġbuild ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( Event Utils . create Message Bean ( e . message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( FAILED ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġconfigure Root Event () Ġ{ ĊĠĠĠĠĠĠĠ Ġroot Event . name ( name ()). type ( type ()) ĊĠĠĠĠĠĠĠ Ġsetup ( root Event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun Ġmatch Filter ( ĊĠĠĠĠĠĠĠ Ġmessage Container : ĠMessage Container , ĊĠĠĠĠĠĠĠ Ġmessage Filter : ĠS ail fish Filter , ĊĠĠĠĠĠĠĠ Ġmetadata Filter : ĠS ail fish Filter ?, ĊĠĠĠĠĠĠĠ Ġmatch Names : ĠBoolean Ġ= Ġtrue , ĊĠĠĠĠĠĠĠ Ġsignificant : ĠBoolean Ġ= Ġtrue ĊĠĠĠ Ġ): ĠAg gregated Filter Result Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Comparison Result : ĠComparison Result ? Ġ= Ġmetadata Filter ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Comparator . compare ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Container . metadata Message , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . message , Ġit . comparator Settings , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmatch Names ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ?. also Ġ{ Ġcomparison Result Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Metadata Ġcomparison Ġresult \ n Ġ{}", Ġcomparison Result ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( metadata Filter Ġ!= Ġnull Ġ&& Ġmetadata Comparison Result Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( LOGGER . isDebugEnabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Metadata Ġfor Ġmessage Ġ{} Ġdoes Ġnot Ġmatch Ġthe Ġfilter Ġby Ġkey Ġfields . ĠSkip Ġmessage Ġchecking ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Container . proto Message . metadata . id . toJson ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠAg gregated Filter Result . EMPTY ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcomparison Result : ĠComparison Result ? Ġ= Ġmessage Filter . let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Comparator . compare ( message Container . s ail fish Message , Ġit . message , Ġit . comparator Settings , Ġmatch Names ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( comparison Result Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Comparison Ġresult Ġfor Ġthe Ġmessage Ġ'{}' Ġwith Ġthe Ġmessage Ġ` {} ` Ġdoes Ġnot Ġmatch Ġthe Ġfilter Ġby Ġkey Ġfields Ġor Ġmessage Ġtype ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Container . s ail fish Message . name , Ġmessage Filter . message . name ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Compare Ġmessage Ġ'{}' Ġresult \ n {}", Ġmessage Container . s ail fish Message . name , Ġcomparison Result ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( comparison Result Ġ!= Ġnull Ġ|| Ġmetadata Comparison Result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sign ificant ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Sequence Ġ= Ġmessage Container . proto Message . metadata . id . sequence ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAg gregated Filter Result ( comparison Result , Ġmetadata Comparison Result ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAg gregated Filter Result . EMPTY ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠDEFAULT _ SEQUENCE Ġ= ĠLong . MIN _ VALUE ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠRESP ONSE _ EXECUT OR Ġ= ĠFork Join Pool . common Pool () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠRoot Message Filter . metadata Filter OrNull (): ĠMetadata Filter ? Ġ= ĊĠĠĠĠĠĠĠ Ġif Ġ( has Metadata Filter ()) Ġmetadata Filter Ġelse Ġnull ĊĊĠĠĠ Ġprotected Ġfun ĠRoot Message Filter . to Compare Settings (): ĠComparator Settings Ġ= ĊĠĠĠĠĠĠĠ ĠComparator Settings (). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . meta Container Ġ= ĠVerification Util . to Meta Container ( this . message Filter , Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . ignored Fields Ġ= Ġthis . comparison Settings . ignore Fields List . toSet () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this . comparison Settings . check Repe ating Group Order ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . is Check Groups Order Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . is Keep Result Group Order Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠMessage Filter . to Compare Settings (): ĠComparator Settings Ġ= ĊĠĠĠĠĠĠĠ ĠComparator Settings (). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . meta Container Ġ= ĠVerification Util . to Meta Container ( this , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠMetadata Filter . to Comparison Settings (): ĠComparator Settings Ġ= ĊĠĠĠĠĠĠĠ ĠComparator Settings (). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . meta Container Ġ= ĠVerification Util . to Meta Container ( this ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠEvent . append Event With Verification ( proto Message : ĠMessage , Ġproto Filter : ĠRoot Message Filter , Ġcomparison Result : ĠComparison Result ): ĠEvent Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġverification Component Ġ= ĠVerification Builder () ĊĠĠĠĠĠĠĠ Ġcomparison Result . results . forEach Ġ{ Ġ( key : ĠString ?, Ġvalue : ĠComparison Result ?) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġverification Component . verification ( key , Ġvalue , Ġproto Filter . message Filter , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwith ( proto Message . metadata ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname (" Verification Ġ'${ message Type }' Ġmessage ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" Verification ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( if Ġ( comparison Result . getStatus Type () Ġ== ĠStatus Type . FAILED ) ĠFAILED Ġelse ĠPAS SED ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. message ID ( id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( verification Component . build ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( proto Filter . has Description ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdescription ( proto Filter . description . value ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠEvent . append Event With Verification ( metadata : ĠMessage Metadata , Ġmetadata Filter : ĠMetadata Filter , Ġcomparison Result : ĠComparison Result ): ĠEvent Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġverification Component Ġ= ĠVerification Builder () ĊĠĠĠĠĠĠĠ Ġcomparison Result . results . forEach Ġ{ Ġ( key : ĠString ?, Ġvalue : ĠComparison Result ?) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġverification Component . verification ( key , Ġvalue , Ġmetadata Filter ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwith ( metadata ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname (" Verification Ġ'${ message Type }' Ġmetadata ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" Verification ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( if Ġ( comparison Result . getStatus Type () Ġ== ĠStatus Type . FAILED ) ĠFAILED Ġelse ĠPAS SED ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. message ID ( id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( verification Component . build ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠEvent . append Event With Ver ifications And Filters ( proto Message Filters : ĠCollection < Root Message Filter >, Ġcomparison Containers : ĠCollection < Comparison Container >): ĠEvent Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ( message Filter Ġin Ġproto Message Filters ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcomparison Container Ġ= Ġcomparison Containers . first OrNull Ġ{ Ġit . proto Filter Ġ=== Ġmessage Filter Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomparison Container ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġappend Events With Verification ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ?: Ġappend Events With Filter ( message Filter ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠEvent . append Events With Filter ( root Message Filter : ĠRoot Message Filter ): ĠEvent Ġ= Ġthis . apply Ġ{ ĊĠĠĠĠĠĠĠ Ġadd Sub Event With Same Period () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" Message Ġfilter ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. type (" Filter ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. status ( FAILED ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. body Data ( root Message Filter . to Readable Body Collection ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠEvent . append Events With Verification ( comparison Container : ĠComparison Container ): ĠEvent Ġ= Ġthis . apply Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġproto Filter Ġ= Ġcomparison Container . proto Filter ĊĠĠĠĠĠĠĠ Ġadd Sub Event With Same Period () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. append Event With Verification ( comparison Container . proto Actual , Ġproto Filter , Ġcomparison Container . result . message Result !! ) ĊĠĠĠĠĠĠĠ Ġif Ġ( proto Filter . has Metadata Filter ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġadd Sub Event With Same Period () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. append Event With Verification ( comparison Container . proto Actual . metadata , Ġproto Filter . metadata Filter , Ġcomparison Container . result . metadata Result !! ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun ĠProto To I Message Converter . from Proto Pre Filter ( proto Pre Message Filter : ĠRoot Message Filter , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Name : ĠString Ġ= Ġproto Pre Message Filter . message Type ): ĠI Message Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġfrom Proto Filter ( proto Pre Message Filter . message Filter , Ġmessage Name ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠObservable < Message >. map To Message Container (): ĠObservable < Message Container > Ġ= ĊĠĠĠĠĠĠĠ Ġmap Ġ{ Ġmessage Ġ-> ĠMessage Container ( message , Ġconverter . from Proto Message ( message , Ġfalse )) Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFilters Ġincoming Ġ{@ link ĠStream Container } Ġvia Ġsession Ġalias Ġand Ġthen ĊĠĠĠĠ Ġ* Ġfilters Ġthe Ġmessage Ġwhich Ġits Ġsequence Ġis Ġgreater Ġthan Ġpassed ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun ĠObservable < Stream Container >. continue Ob serve ( session Key : ĠSession Key , Ġsequence : ĠLong ): ĠObservable < Message > Ġ= ĊĠĠĠĠĠĠĠ Ġfilter Ġ{ Ġstream Container Ġ-> Ġstream Container . session Key Ġ== Ġsession Key Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap ( Stream Container :: buffered Messages ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġmessage Ġ-> Ġmessage . metadata . id . sequence Ġ> Ġsequence Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠCheck point . get Sequence ( session Key : ĠSession Key ): ĠLong Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsequence Ġ= Ġsession Alias To Direction Checkpoint Map [ session Key . session Alias ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ? . direction To Sequence Map ?. get ( session Key . direction . number ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ( sequence Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( LOGGER . is Warn Enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . warn (" Checkpoint Ġ'{}' Ġdoesn 't Ġcontain Ġsequence Ġfor Ġsession Ġ'{} '", Ġshort Debug String ( this ), Ġsession Key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . info (" Use Ġsequence Ġ'{}' Ġfrom Ġcheckpoint Ġfor Ġsession Ġ'{} '", Ġsequence , Ġsession Key ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġsequence Ġ?: ĠDEFAULT _ SEQUENCE ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġdata Ġclass ĠLegacy ( val ĠexecutorService : ĠExecutor Service , Ġval Ġlast Sequence : ĠLong ) Ċ } Ċ