package Ġcom . github . j pth i ery . heim d all . infra Ċ Ċ /* ĊĠĠĠ ĠCopyright Ġ 2 0 1 9 ĠJean - Pas cal ĠTh i ery ĊĊĠĠ Ġ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 . Ċ */ Ċ Ċ import Ġcom . github . j pth i ery . heim d all . domain .* Ċ import Ġredis . clients . j edis . J edis Ċ import Ġredis . clients . j edis . J edis Pool Ċ import Ġjava . util .* Ċ Ċ class ĠRedis Event Store ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġpool : ĠJ edis Pool , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġevent Store Key For ger : ĠRedis Event Store Key For ger , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġmap pers : ĠMap < Class < out ĠEvent >, ĠRedis Event Store Mapper < out ĠEvent >> Ċ ) Ġ: ĠEvent Store Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġ< C Ġ: ĠCommand , ĠS Ġ: ĠState , ĠE Ġ: ĠEvent > Ġget Event For Aggregate ( aggregate : ĠAggregate < C , ĠS , ĠE >, Ġid : ĠStream Id ): ĠList < E > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġget Mapper ( aggregate . get EventType ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġmapper Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġevent Store Key For ger . forge Key ( aggregate , Ġid ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry In Transaction Ġ{ Ġjed is Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( j edis . exists ( key )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġredis Entries Ġ= Ġjed is . l range ( key , Ġ 0 , Ġjed is . llen ( key )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġredis Entries ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġevent Str Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmapper . deserialize ( event Str ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġaggregate . get EventType (). isAssignableFrom ( it :: class . java ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toList () Ġas ĠList < E > ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġempty List () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse ( emptyList ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġ< I Ġ: ĠStream Id , ĠC Ġ: ĠCommand , ĠS Ġ: ĠState , ĠE Ġ: ĠEvent > Ġappend Events ( aggregate : ĠAggregate < C , ĠS , ĠE >, Ġid : ĠI , Ġevents : ĠList < E >, ĠinitialState Version : ĠInt ): ĠApp ended Event Result Ġ{ ĊĊĠĠĠĠĠĠĠ Ġreturn Ġget Mapper ( aggregate . get EventType ()). map Ġ{ Ġmapper Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġevent Store Key For ger . forge Key ( aggregate , Ġid ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry In Transaction Ġ{ Ġjed is Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġredis Entries Ġ= Ġevents ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmapper . serialize ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjed is . lp ush ( key , Ġ* redis Entries . to Typed Array ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevents Stored Ġ= Ġget Event For Aggregate ( aggregate , Ġid ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSuccessful App ended Event Result ( aggregate . replay ( events Stored ). version ) Ġas ĠApp ended Event Result ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. orElse ( Failed App ended Event Result (" Unable Ġto Ġfind Ġand Ġevent Ġmapper Ġfor Ġaggregate Ġ${ aggregate :: class . java . simple Name } ")) ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdata Ġclass ĠRedis Configuration ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhost : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġport : ĠInt ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġfun Ġ< T > Ġtry In Transaction ( action : Ġ( J edis ) Ġ-> ĠT ): ĠT Ġ= Ġpool . resource . use Ġ{ Ġjed is Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġtransaction Ġ= Ġjed is . multi () ĊĠĠĠĠĠĠĠ Ġval Ġres Ġ= Ġaction ( j edis ) ĊĠĠĠĠĠĠĠ Ġtransaction . exec () ĊĠĠĠĠĠĠĠ Ġres ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġprivate Ġfun Ġget Mapper ( eventType : ĠClass < out ĠEvent >): ĠOptional < Redis Event Store Mapper < out ĠEvent >> Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( map pers . containsKey ( eventType )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠOptional . of Nullable ( map pers [ eventType ]) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠOptional . empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinterface ĠRedis Event Store Key For ger Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġfor ge Key ( aggregate : ĠAggregate < *, Ġ*, Ġ* >, Ġid : ĠStream Id ): ĠString ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinterface ĠRedis Event Store Mapper < E Ġ: ĠEvent > Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġserialize ( event : ĠEvent ): ĠString ĊĠĠĠĠĠĠĠ Ġfun Ġdeserialize ( data : ĠString ): ĠE ĊĠĠĠ Ġ} Ċ }