/* Ċ Ġ* Ġ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 . sequence Ċ Ċ 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 . exception . Rule Internal Exception Ċ import Ġcom . exact pro . th 2 . check 1 . grpc . Pre Filter Ċ import Ġcom . exact pro . th 2 . check 1 . rule . Abstract Check Task Test Ċ import Ġcom . exact pro . th 2 . check 1 . rule . sequence . Sequence Check Rule Task . Companion . CHECK _ MESSAGES _ TYPE Ċ import Ġcom . exact pro . th 2 . check 1 . rule . sequence . Sequence Check Rule Task . Companion . CHECK _ SEQUENCE _ TYPE Ċ import Ġcom . exact pro . th 2 . check 1 . util . to Simple Filter Ċ import Ġcom . exact pro . th 2 . common . event . Event Utils Ċ import Ġcom . exact pro . th 2 . common . grpc . Direction Ċ import Ġcom . exact pro . th 2 . common . grpc . Event Ċ 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 . Event Status Ċ import Ġcom . exact pro . th 2 . common . grpc . Filter Operation Ċ 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 . grpc . Value Ċ import Ġcom . exact pro . th 2 . common . grpc . Value Filter Ċ import Ġcom . exact pro . th 2 . common . message . message Ċ import Ġcom . exact pro . th 2 . common . message . message Filter Ċ import Ġio . reactivex . Observable Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . assert All Ċ import Ġorg . junit . jupiter . api . assert Throws Ċ import Ġorg . junit . jupiter . params . Parameterized Test Ċ import Ġorg . junit . jupiter . params . provider . Arguments Ċ import Ġorg . junit . jupiter . params . provider . Arguments . arguments Ċ import Ġorg . junit . jupiter . params . provider . Method Source Ċ import Ġorg . junit . jupiter . params . provider . Value Source Ċ import Ġjava . time . Instant Ċ import Ġjava . util . stream . Stream Ċ import Ġkotlin . test . assertEquals Ċ import Ġkotlin . test . assertNotNull Ċ import Ġkotlin . test . assertTrue Ċ Ċ class ĠTest Sequence Check Task Ġ: ĠAbstract Check Task Test () Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġproto Message Filters : ĠList < Root Message Filter > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue Filter . newBuilder (). set Key ( true ). set Simple Filter (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue Filter . newBuilder (). set Simple Filter (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). build (), ĊĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue Filter . newBuilder (). set Key ( true ). set Simple Filter (" 4 3 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue Filter . newBuilder (). set Simple Filter (" BBB "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). build (), ĊĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue Filter . newBuilder (). set Key ( true ). set Simple Filter (" 4 4 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue Filter . newBuilder (). set Simple Filter (" CCC "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). build () ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġval Ġmessages In Correct Order : ĠList < Message > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠ Ġconstruct Message ( 1 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠ Ġconstruct Message ( 2 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 3 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" BBB "). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠ Ġconstruct Message ( 3 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 4 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" CCC "). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġval Ġpre Filter Ġ= ĠPre Filter . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. put Fields (" A ", ĠValue Filter . newBuilder (). set Key ( true ). set Operation ( Filter Operation . NOT _ EMPTY ). build ()) ĊĠĠĠĠĠĠĠ Ġ. build () ĊĊĠĠĠ Ġ@ Parameterized Test ( name Ġ= Ġ" check Order Ġ= Ġ{ 0 }") ĊĠĠĠ Ġ@ Value Source ( b ooleans Ġ= Ġ[ true , Ġfalse ]) ĊĠĠĠ Ġfun Ġ` messages Ġin Ġright Ġorder Ġpasses `( check Order : ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages In Correct Order ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġcheck Order ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġ/* ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Sequence Rule ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpre Filtering ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠVerification Ġx Ġ 3 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Messages ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠVerification Ġx Ġ 3 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Sequence ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 0 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" check Sequence Rule ", Ġget Events ( 0 ). type ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 1 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" pre Filtering ", Ġget Events ( 0 ). type ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 2 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue Ġ( get Events List (). all Ġ{ ĠVER IFICATION _ TYPE Ġ== Ġit . type Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 3 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( CHECK _ MESSAGES _ TYPE , Ġget Events ( 0 ). type ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 4 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue Ġ( get Events List (). all Ġ{ ĠVER IFICATION _ TYPE Ġ== Ġit . type Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( batch Request [ 5 ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġevents Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" check Sequence ", Ġget Events ( 0 ). type ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġpassed : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . SUCCESS Ġ} Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . SUCCESS , Ġevents List ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Parameterized Test ( name Ġ= Ġ" Messages Ġreordered : Ġ{ 0 }") ĊĠĠĠ Ġ@ Method Source (" indexes To Switch ") ĊĠĠĠ Ġfun Ġ` verification Ġwith Ġcheck Ġorder Ġis Ġfailed Ġin Ġcase Ġmessages Ġare Ġreordered `( indexes To Switch : ĠPair < Int , ĠInt >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages Un ordered Ġ= Ġmessages In Correct Order . to Mutable List (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtmp Ġ= Ġget ( indexes To Switch . first ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset ( indexes To Switch . first , Ġget ( indexes To Switch . second )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset ( indexes To Switch . second , Ġtmp ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages Un ordered ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġtrue ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpassed Ver ifications Ġ= Ġver ifications . filter Ġ{ Ġit . status Ġ== ĠEvent Status . SUCCESS Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġmessages Ġare Ġreordered Ġbut Ġevery Ġverification Ġis Ġpresented . ĠSo , Ġevery Ġverification Ġshould Ġbe Ġpassed ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġpassed Ver ifications . size , Ġ" Unexpected ĠSUCCESS Ġver ifications Ġcount : Ġ$ passed Ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġhave Ġmore Ġthan Ġone Ġmessage Ġattached ") Ġ{ Ġpassed Ver ifications . all Ġ{ Ġit . attached Message Ids Count Ġ== Ġ 1 Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠI ds Ġin Ġthe Ġresult Ġof Ġthe Ġrule Ġare Ġin Ġorder Ġby Ġfilters Ġbecause Ġthe Ġrule Ġcreates Ġevents Ġrelated Ġto Ġver ifications / filters Ġin Ġthe Ġsource Ġorder . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( messages In Correct Order . map Ġ{ Ġit . metadata . id Ġ}, Ġpassed Ver ifications . map Ġ{ Ġit . get Attached Message Ids ( 0 ) Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . FAILED , Ġevents List ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Parameterized Test ( name Ġ= Ġ" check Ġorder : Ġ{ 0 }") ĊĠĠĠ Ġ@ Method Source (" check Order To Switch ") ĊĠĠĠ Ġfun Ġ` check Ġsequence Ġshould Ġdrop Ġa Ġmessage Ġfilter Ġafter Ġmatch Ġby Ġkey Ġfields `( check Order : ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages With Key Fields : ĠList < Message > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 1 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 2 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" BBB "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Filter Ġ= ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue Filter . newBuilder (). set Key ( true ). set Simple Filter (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue Filter . newBuilder (). set Simple Filter (" CCC "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). build () ĊĠĠĠĠĠĠĠ Ġval Ġmessage Filters : ĠList < Root Message Filter > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder ( message Filter ). build (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder ( message Filter ). build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages With Key Fields ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġcheck Order , Ġfilters Param Ġ= Ġmessage Filters ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġroot Event Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . parentId Ġ== Ġparent Event ID Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 2 , Ġroot Event . attached Message Ids Count ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L ), Ġroot Event . attached Message Ids List . map Ġ{ Ġit . sequence Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 2 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġsuccess : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . FAILED Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L ), Ġver ifications . flatMap Ġ{ Ġverification Ġ-> Ġverification . attached Message Ids List . map Ġ{ Ġit . sequence Ġ} Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Sequence Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ SEQUENCE _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Ġsequence Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( Event Status . SUCCESS , Ġchecked Sequence . status ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` check Ġsequence Ġof Ġmessages Ġwith Ġthe Ġsame Ġvalue Ġof Ġkey Ġfield `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages With Key Fields : ĠList < Message > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 1 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 2 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 3 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Filter Ġ= ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue Filter . newBuilder (). set Key ( true ). set Simple Filter (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue Filter . newBuilder (). set Simple Filter (" AAA "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). build () ĊĠĠĠĠĠĠĠ Ġval Ġmessage Filters : ĠList < Root Message Filter > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder ( message Filter ). build (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder ( message Filter ). build (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRoot Message Filter . newBuilder ( message Filter ). build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages With Key Fields ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġtrue , Ġfilters Param Ġ= Ġmessage Filters ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġroot Event Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . parentId Ġ== Ġparent Event ID Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġroot Event . attached Message Ids Count ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L , Ġ 3 L ), Ġroot Event . attached Message Ids List . map Ġ{ Ġit . sequence Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġsuccess : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . SUCCESS Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L , Ġ 3 L ), Ġver ifications . flatMap Ġ{ Ġverification Ġ-> Ġverification . attached Message Ids List . map Ġ{ Ġit . sequence Ġ} Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . SUCCESS , Ġevents List ) Ġ// Ġbecause Ġall Ġkey Ġfields Ġare Ġin Ġa Ġcorrect Ġorder ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` check Ġordering Ġis Ġnot Ġfailed Ġin Ġcase Ġkey Ġfields Ġare Ġmatches Ġthe Ġorder Ġbut Ġthe Ġrest Ġare Ġnot `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages With Key Fields : ĠList < Message > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 1 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 2 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 3 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" BBB 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 3 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 4 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" CCC 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages With Key Fields ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġtrue ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġroot Event Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . parentId Ġ== Ġparent Event ID Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġroot Event . attached Message Ids Count ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L , Ġ 3 L ), Ġroot Event . attached Message Ids List . map Ġ{ Ġit . sequence Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġfailed : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . FAILED Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L , Ġ 3 L ), Ġver ifications . flatMap Ġ{ Ġverification Ġ-> Ġverification . attached Message Ids List . map Ġ{ Ġit . sequence Ġ} Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . SUCCESS , Ġevents List ) Ġ// Ġbecause Ġall Ġkey Ġfields Ġare Ġin Ġa Ġcorrect Ġorder ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Parameterized Test ( name Ġ= Ġ" Messages Ġreordered : Ġ{ 0 }") ĊĠĠĠ Ġ@ Method Source (" indexes To Switch ") ĊĠĠĠ Ġfun Ġ` verification Ġwithout Ġcheck Ġorder Ġis Ġnot Ġfailed Ġin Ġcase Ġmessages Ġare Ġreordered `( indexes To Switch : ĠPair < Int , ĠInt >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages Un ordered Ġ= Ġmessages In Correct Order . to Mutable List (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtmp Ġ= Ġget ( indexes To Switch . first ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset ( indexes To Switch . first , Ġget ( indexes To Switch . second )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset ( indexes To Switch . second , Ġtmp ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages Un ordered ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġfalse ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġpassed : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . SUCCESS Ġ} Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . SUCCESS , Ġevents List ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` rules Ġstops Ġwhen Ġall Ġfilters Ġfound Ġmatch Ġby Ġkey Ġfields `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages With Key Fields : ĠList < Message > Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 1 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" AAA 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 2 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) Ġ// Ġgoes Ġto Ġprocessed Ġmessages Ġbut Ġshould Ġnot Ġgo Ġto Ġthe Ġactual Ġcomparison ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" BBB 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 3 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 3 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" BBB 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 4 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 4 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" CCC 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconstruct Message ( 5 , ĠSESSION _ ALIAS , ĠMESSAGE _ TYPE ) Ġ// Ġshould Ġnot Ġbe Ġprocessed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put All Fields ( map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" A " Ġto ĠValue . newBuilder (). set Simple Value (" 4 2 "). build (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" B " Ġto ĠValue . newBuilder (). set Simple Value (" DDD 1 "). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= ĠObservable . from Iterable ( messages With Key Fields ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Stream : ĠObservable < Stream Container > Ġ= ĠObservable . just ( Stream Container ( Session Key ( SESSION _ ALIAS , ĠDirection . FIRST ), Ġ 1 0 , Ġmessages )) ĊĠĠĠĠĠĠĠ Ġval Ġparent Event ID Ġ= ĠEvent ID . newBuilder (). setId ( Event Utils . generate UUID ()). build () ĊĊĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID , Ġmessage Stream , Ġfalse ). begin () ĊĊĠĠĠĠĠĠĠ Ġval Ġbatch Request Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġval Ġevents List : ĠList < Event > Ġ= Ġbatch Request . flatMap ( Event Batch :: get Events List ) ĊĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġroot Event Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . parentId Ġ== Ġparent Event ID Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 4 , Ġroot Event . attached Message Ids Count ) Ġ// Ġ 3 Ġmatch Ġkey Ġ+ Ġ 1 Ġthat Ġdoesn 't Ġmatch Ġbut Ġis Ġbetween Ġothers ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 2 L , Ġ 3 L , Ġ 4 L ), Ġroot Event . attached Message Ids List . map Ġ{ Ġit . sequence Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchecked Messages Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== ĠCHECK _ MESSAGES _ TYPE Ġ}, Ġ" Cannot Ġfind Ġcheck Messages Ġevent ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġver ifications Ġ= Ġevents List . filter Ġ{ Ġit . parentId Ġ== Ġchecked Messages . id Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġver ifications . size , Ġ" Unexpected Ġver ifications Ġcount : Ġ$ ver ifications ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue (" Some Ġver ifications Ġare Ġnot Ġfailed : Ġ$ ver ifications ") Ġ{ Ġver ifications . all Ġ{ Ġit . status Ġ== ĠEvent Status . FAILED Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 1 L , Ġ 3 L , Ġ 4 L ), Ġver ifications . flatMap Ġ{ Ġverification Ġ-> Ġverification . attached Message Ids List . map Ġ{ Ġit . sequence Ġ} Ġ}) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Check Sequence Status ( Event Status . SUCCESS , Ġevents List ) Ġ// Ġbecause Ġthe Ġactual Ġcomparisons Ġcount Ġequals Ġto Ġexpected ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` failed Ġrule Ġcreation Ġdue Ġto Ġinvalid Ġregex Ġoperation Ġin Ġthe Ġmessage Ġfilter `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstreams Ġ= Ġcreate Streams ( SESSION _ ALIAS , ĠDirection . FIRST , ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage ( MESSAGE _ TYPE , ĠDirection . FIRST , ĠSESSION _ ALIAS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. merge Metadata ( Message Metadata . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put Properties (" key Prop ", Ġ" 4 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put Properties (" not Key Prop ", Ġ" 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ)) ĊĊĠĠĠĠĠĠĠ Ġval Ġevent ID Ġ= ĠEvent ID . newBuilder (). setId (" root "). build () ĊĠĠĠĠĠĠĠ Ġval Ġfilter Ġ= ĠRoot Message Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set MessageType ( MESSAGE _ TYPE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Metadata Filter ( Metadata Filter . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. put Property Filters (" key Prop ", Ġ" 4 2 ". to Simple Filter ( Filter Operation . EQUAL ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage Filter ( message Filter (). put Fields (" key Prop ", ĠValue Filter . newBuilder (). set Operation ( Filter Operation . LIKE ). set Simple Filter (". [" ). build ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĊĠĠĠĠĠĠĠ ĠassertThrows < Rule Internal Exception > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Check Rule Task ( parent Event ID Ġ= Ġevent ID , Ġmessage Stream Ġ= Ġstreams , Ġfilters Param Ġ= ĠlistOf ( filter ), Ġcheck Order Ġ= Ġfalse ). begin () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġevent B atches Ġ= Ġawait Event Batch Request ( 1 0 0 0 L , Ġ 2 ) ĊĠĠĠĠĠĠĠ Ġval Ġevent List Ġ= Ġevent B atches . flatMap ( Event Batch :: get Events List ) ĊĠĠĠĠĠĠĠ Ġassert All ({ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 3 , Ġevent List . size ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġevent List . filter Ġ{ Ġit . type Ġ== Ġ" internal Error " Ġ}. size ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġassert Check Sequence Status ( expected Status : ĠEvent Status , Ġevents List : ĠList < Event >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcheck Sequence Event Ġ= ĠassertNotNull ( events List . find Ġ{ Ġit . type Ġ== Ġ" check Sequence " Ġ}, Ġ" Cannot Ġfind Ġcheck Sequence Ġevent ") ĊĠĠĠĠĠĠĠ ĠassertEquals ( expected Status , Ġcheck Sequence Event . status ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsequence Check Rule Task ( ĊĠĠĠĠĠĠĠ Ġparent Event ID : ĠEvent ID , ĊĠĠĠĠĠĠĠ Ġmessage Stream : ĠObservable < Stream Container >, ĊĠĠĠĠĠĠĠ Ġcheck Order : ĠBoolean , ĊĠĠĠĠĠĠĠ Ġpre Filter Param : ĠPre Filter Ġ= Ġpre Filter , ĊĠĠĠĠĠĠĠ Ġfilters Param : ĠList < Root Message Filter > Ġ= Ġproto Message Filters , ĊĠĠĠĠĠĠĠ Ġmax Event Batch Content Size : ĠInt Ġ= Ġ 1 0 2 4 Ġ* Ġ 1 0 2 4 ĊĠĠĠ Ġ): ĠSequence Check Rule Task Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠSequence Check Rule Task ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdescription Ġ= Ġ" Test ", ĊĠĠĠĠĠĠĠĠĠĠĠ ĠstartTime Ġ= ĠInstant . now (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsession Key Ġ= ĠSession Key ( SESSION _ ALIAS , ĠDirection . FIRST ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtimeout Ġ= Ġ 5 0 0 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmax Event Batch Content Size Ġ= Ġmax Event Batch Content Size , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproto Pre Filter Ġ= Ġpre Filter Param , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproto Message Filters Ġ= Ġfilters Param , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Order Ġ= Ġcheck Order , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġparent Event ID Ġ= Ġparent Event ID , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Stream Ġ= Ġmessage Stream , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Batch Router Ġ= Ġclient Stub ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġindexes To Switch (): ĠStream < Arguments > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠStream . of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarguments ( 0 Ġto Ġ 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarguments ( 0 Ġto Ġ 2 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarguments ( 1 Ġto Ġ 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġcheck Order To Switch (): ĠStream < Arguments > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠStream . of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarguments ( false ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarguments ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ