package Ġcom . ts bone v . cq rs . core Ċ Ċ import Ġcom . ts bone v . cq rs . core . event store . Create Snapshot Request Ċ import Ġcom . ts bone v . cq rs . core . event store . Event Payload Ċ import Ġcom . ts bone v . cq rs . core . event store . Event Publisher Ċ import Ġcom . ts bone v . cq rs . core . event store . Event Store Ċ import Ġcom . ts bone v . cq rs . core . event store . Get Events From Streams Request Ċ import Ġcom . ts bone v . cq rs . core . event store . Get Events Response Ċ import Ġcom . ts bone v . cq rs . core . event store . Save Events Request Ċ import Ġcom . ts bone v . cq rs . core . event store . Save Events Response Ċ import Ġcom . ts bone v . cq rs . core . event store . Save Options Ċ import Ġcom . ts bone v . cq rs . core . snapshot . Message Format Ċ import Ġcom . ts bone v . cq rs . core . snapshot . Snapshot Ċ import Ġjava . io . ByteArray InputStream Ċ import Ġjava . util . UUID ĊĊ Ċ class ĠSimple Identity Aggregate Repository ( Ċ ĉ private Ġval Ġevent Store : ĠEvent Store , Ċ ĉ private Ġval Ġmessage Format : ĠMessage Format , Ċ ĉ private Ġval Ġevent Publisher : ĠEvent Publisher , Ċ ĉ private Ġval Ġconfiguration : ĠAggregate Configuration Ċ ) Ġ: ĠIdentity Aggregate Repository Ġ{ ĊĊ ĉ override Ġfun Ġ< T Ġ: ĠAggregate Root > Ġsave ( stream : ĠString , Ġaggregate : ĠT , Ġidentity : ĠIdentity ) Ġ{ Ċĉ ĉ var Ġaggregate Id Ġ= Ġaggregate . getId () Ċĉ ĉ if Ġ( aggregate Id Ġ== Ġ"") Ġ{ Ċĉĉ ĉ aggregate Id Ġ= ĠUUID . randomUUID (). toString () Ċĉ ĉ } ĊĊĉ ĉ val Ġun committed Events Ġ= Ġaggregate . get Events () ĊĊĉ ĉ val Ġevents With Payload Ġ= Ġun committed Events . map Ġ{ Ċĉĉ ĉ Event With Binary Payload ( it , ĠBinary Payload ( message Format . format ToBytes ( it ))) Ċĉ ĉ } ĊĊĉ ĉ val Ġevents Ġ= Ġevents With Payload . map Ġ{ Ċĉĉ ĉ Event Payload ( Ċĉĉĉ ĉ aggregate Id , Ċĉĉĉ ĉ it . event :: class . java . simple Name , Ċĉĉĉ ĉ identity . time . to Epoch Milli (), Ċĉĉĉ ĉ identity . id , Ċĉĉĉ ĉ it . payload Ċĉĉ ĉ ) Ċĉ ĉ } ĊĊĉ ĉ /** Ċĉĉ Ġ* ĠIf Ġthere Ġare Ġno Ġnews Ġchanges , Ġthe Ġaggregate Ġdoesn 't Ġneed Ġupdating . Ċĉĉ Ġ*/ Ċĉ ĉ if Ġ( events . isEmpty ()) Ġreturn ĊĊĉ ĉ val Ġaggregate Class Ġ= Ġaggregate :: class . java Ċĉ ĉ val Ġaggregate Type Ġ= Ġaggregate Class . simple Name ĊĊĉ ĉ val Ġtopic Name Ġ= Ġconfiguration . topic Name ( aggregate ) Ċĉ ĉ val Ġsave Events Request Ġ= ĠSave Events Request ( identity . tenant , Ġstream , Ġaggregate Type , Ġevents ) Ċĉ ĉ val Ġresponse Ġ= Ġevent Store . save Events ( Ċĉĉ ĉ save Events Request , Ċĉĉ ĉ Save Options ( version Ġ= Ġaggregate . get Expected Version (), Ġtopic Name Ġ= Ġtopic Name ) Ċĉ ĉ ) ĊĊĉ ĉ when Ġ( response ) Ġ{ Ċĉĉ ĉ is ĠSave Events Response . Success Ġ-> Ġ{ Ċĉĉĉ ĉ try Ġ{ Ċĉĉĉĉ ĉ event Publisher . publish ( events With Payload ) Ċĉĉĉĉ ĉ aggregate . commit Events () Ċĉĉĉ ĉ } Ġcatch Ġ( ex : ĠPublish ErrorException ) Ġ{ Ċĉĉĉĉ ĉ event Store . revert Events ( identity . tenant , Ġstream , Ġevents . size ) Ċĉĉĉĉ ĉ throw Ġex Ċĉĉĉ ĉ } Ċĉĉ ĉ } ĊĊĉĉ ĉ is ĠSave Events Response . Event Collision Ġ-> Ġ{ Ċĉĉĉ ĉ throw ĠEvent Collision Exception ( aggregate . getId (), Ġresponse . expected Version ) Ċĉĉ ĉ } ĊĊĉĉ ĉ is ĠSave Events Response . Snapshot Required Ġ-> Ġ{ Ċĉĉĉ ĉ val Ġcurrent Aggregate Ġ= Ġbuild Aggregate From History ( Ċĉĉĉĉ ĉ aggregate Class , Ċĉĉĉĉ ĉ response . current Events , Ċĉĉĉĉ ĉ response . version , Ċĉĉĉĉ ĉ aggregate . getId (), Ċĉĉĉĉ ĉ response . current Snapshot Ċĉĉĉ ĉ ) ĊĊĉĉĉ ĉ val Ġnew Snapshot Ġ= Ġcurrent Aggregate . get Snapshot Mapper (). to Snapshot ( current Aggregate , Ġmessage Format ) Ċĉĉĉ ĉ val Ġcreate Snapshot Response Ġ= Ġevent Store . save Events ( Ċĉĉĉĉ ĉ save Events Request , Ċĉĉĉĉ ĉ Save Options ( Ċĉĉĉĉĉ ĉ version Ġ= Ġnew Snapshot . version , Ċĉĉĉĉĉ ĉ create Snapshot Request Ġ= ĠCreate Snapshot Request ( true , Ġnew Snapshot ) Ċĉĉĉĉ ĉ ) Ċĉĉĉ ĉ ) ĊĊĉĉĉ ĉ when Ġ( create Snapshot Response ) Ġ{ Ċĉĉĉĉ ĉ is ĠSave Events Response . Success Ġ-> Ġ{ Ċĉĉĉĉĉ ĉ try Ġ{ Ċĉĉĉĉĉĉ ĉ event Publisher . publish ( events With Payload ) Ċĉĉĉĉĉĉ ĉ aggregate . commit Events () Ċĉĉĉĉĉ ĉ } Ġcatch Ġ( ex : ĠPublish ErrorException ) Ġ{ Ċĉĉĉĉĉĉ ĉ event Store . revert Events ( aggregate Type , Ġaggregate . getId (), Ġevents . size ) Ċĉĉĉĉĉĉ ĉ throw Ġex Ċĉĉĉĉĉ ĉ } Ċĉĉĉĉ ĉ } Ċĉĉĉĉ ĉ is ĠSave Events Response . Event Collision Ġ-> Ġ{ Ċĉĉĉĉĉ ĉ throw ĠEvent Collision Exception ( aggregate . getId (), Ġcreate Snapshot Response . expected Version ) Ċĉĉĉĉ ĉ } ĊĊĉĉĉĉ ĉ else Ġ-> Ġthrow ĠIllegalStateException (" Unable Ġto Ġsave Ġevents .") Ċĉĉĉ ĉ } Ċĉĉ ĉ } ĊĊĉĉ ĉ else Ġ-> Ġthrow ĠIllegalStateException (" Unable Ġto Ġsave Ġevents .") Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ /** Ċĉ Ġ* ĠCreates Ġa Ġnew Ġor Ġupdates Ġan Ġexisting Ġaggregate Ġin Ġthe Ġrepository . Ċĉ Ġ* Ċĉ Ġ* ĠThis Ġmethod Ġensures Ġa Ġone Ġto Ġone Ġrelationship Ġof Ġstream Ġand Ġaggregate Ġand Ġwill Ġuse Ċĉ Ġ* ĠAggregate Type _ Aggregate Id Ġpattern Ġfor Ġthe Ġname Ġof Ġthe Ġstream Ġwhere Ġdata Ġwill Ġbe Ġwritten . Ċĉ Ġ* Ċĉ Ġ* Ġ@ param Ġaggregate Ġthe Ġaggregate Ġto Ġbe Ġregistered Ċĉ Ġ* Ġ@ throws ĠEvent Collision Exception Ġis Ġthrown Ġin Ġcase Ġof Ċĉ Ġ*/ Ċ ĉ @ Throws ( Event Collision Exception :: class , ĠPublish ErrorException :: class ) Ċ ĉ override Ġfun Ġ< T Ġ: ĠAggregate Root > Ġsave ( aggregate : ĠT , Ġidentity : ĠIdentity ) Ġ{ Ċĉ ĉ val Ġaggregate Class Ġ= Ġaggregate :: class . java Ċĉ ĉ val Ġaggregate Type Ġ= Ġaggregate Class . simple Name Ċĉ ĉ var Ġaggregate Id Ġ= Ġaggregate . getId () ĊĊĉ ĉ if Ġ( aggregate Id Ġ== Ġ"") Ġ{ Ċĉĉ ĉ aggregate Id Ġ= ĠUUID . randomUUID (). toString () Ċĉ ĉ } ĊĊĉ ĉ return Ġthis . save ( Stream Key . of ( aggregate Type , Ġaggregate Id ), Ġaggregate , Ġidentity ) Ċ ĉ } ĊĊ ĉ override Ġfun Ġ< T Ġ: ĠAggregate Root > Ġget ById ( stream : ĠString , Ġaggregate Id : ĠString , Ġtype : ĠClass < T >, Ġidentity : ĠIdentity ): ĠT Ġ{ Ċĉ ĉ when Ġ( val Ġresponse Ġ= Ġevent Store . get Events From Streams ( Ċĉĉ ĉ Get Events From Streams Request ( identity . tenant , ĠlistOf ( stream ))) Ċĉĉ ĉ ) Ġ{ Ċĉĉ ĉ is ĠGet Events Response . Success Ġ-> Ġ{ Ċĉĉĉ ĉ if Ġ( response . aggreg ates . isEmpty ()) Ġ{ Ċĉĉĉĉ ĉ throw ĠAggregate NotFoundException ( aggregate Id ) Ċĉĉĉ ĉ } ĊĊĉĉĉ ĉ val Ġaggregate Ġ= Ġresponse . aggreg ates . find Ġ{ Ġit . events [ 0 ]. aggregate Id Ġ== Ġaggregate Id Ġ} Ċĉĉĉĉ ĉ ?: Ġthrow ĠAggregate NotFoundException ( aggregate Id ) ĊĊĉĉĉ ĉ // we Ġare Ġsure Ġthat Ġonly Ġone Ġaggregate Ġwill Ġbe Ġreturned Ċĉĉĉ ĉ return Ġbuild Aggregate From History ( Ċĉĉĉĉ ĉ type , Ċĉĉĉĉ ĉ aggregate . events , Ċĉĉĉĉ ĉ aggregate . version , Ċĉĉĉĉ ĉ aggregate Id , Ċĉĉĉĉ ĉ response . aggreg ates . first (). snapshot Ċĉĉĉ ĉ ) Ċĉĉ ĉ } Ċĉĉ ĉ else Ġ-> Ġthrow ĠIllegalStateException (" unknown Ġstate ") Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ override Ġfun Ġ< T Ġ: ĠAggregate Root > Ġget ById ( id : ĠString , Ġtype : ĠClass < T >, Ġidentity : ĠIdentity ): ĠT Ġ{ Ċĉ ĉ val Ġstream Ġ= ĠStream Key . of ( type . simple Name , Ġid ) ĊĊĉ ĉ when Ġ( val Ġresponse Ġ= Ġevent Store . get Events From Streams ( Ċĉĉ ĉ Get Events From Streams Request ( identity . tenant , ĠlistOf ( stream ))) Ċĉĉ ĉ ) Ġ{ Ċĉĉ ĉ is ĠGet Events Response . Success Ġ-> Ġ{ Ċĉĉĉ ĉ if Ġ( response . aggreg ates . isEmpty ()) Ġ{ Ċĉĉĉĉ ĉ throw ĠAggregate NotFoundException ( id ) Ċĉĉĉ ĉ } Ċĉĉĉ ĉ val Ġaggregate Ġ= Ġresponse . aggreg ates . first () ĊĊĉĉĉ ĉ return Ġbuild Aggregate From History ( Ċĉĉĉĉ ĉ type , Ċĉĉĉĉ ĉ aggregate . events , Ċĉĉĉĉ ĉ aggregate . version , Ċĉĉĉĉ ĉ id , Ċĉĉĉĉ ĉ response . aggreg ates . first (). snapshot Ċĉĉĉ ĉ ) Ċĉĉ ĉ } Ċĉĉ ĉ else Ġ-> Ġthrow ĠIllegalStateException (" Unknown Ġstate ") Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ override Ġfun Ġ< T Ġ: ĠAggregate Root > Ġget ById s ( ids : ĠList < String >, Ġtype : ĠClass < T >, Ġidentity : ĠIdentity ): ĠMap < String , ĠT > Ġ{ Ċĉ ĉ val Ġaggregate Type Ġ= Ġtype . simple Name Ċĉ ĉ if Ġ( ids . isEmpty ()) Ġ{ Ċĉĉ ĉ return Ġmap Of () Ċĉ ĉ } ĊĊĉ ĉ val Ġstream Ids Ġ= Ġids . map Ġ{ ĠStream Key . of ( aggregate Type , Ġit ) Ġ} ĊĊĉ ĉ when Ġ( val Ġresponse Ġ= Ġevent Store . get Events From Streams ( Ċĉĉ ĉ Get Events From Streams Request ( identity . tenant , Ġstream Ids )) Ċĉĉ ĉ ) Ġ{ Ċĉĉ ĉ is ĠGet Events Response . Success Ġ-> Ġ{ Ċĉĉĉ ĉ val Ġresult Ġ= Ġmutable Map Of < String , ĠT >() Ċĉĉĉ ĉ response . aggreg ates . forEach Ġ{ Ċĉĉĉĉ ĉ val Ġaggregate Id Ġ= Ġit . events [ 0 ]. aggregate Id Ċĉĉĉĉ ĉ result [ aggregate Id ] Ġ= Ġbuild Aggregate From History ( Ċĉĉĉĉĉ ĉ type , Ċĉĉĉĉĉ ĉ it . events , Ċĉĉĉĉĉ ĉ it . version , Ċĉĉĉĉĉ ĉ aggregate Id , Ċĉĉĉĉĉ ĉ it . snapshot Ċĉĉĉĉ ĉ ) Ċĉĉĉ ĉ } Ċĉĉĉ ĉ return Ġresult Ċĉĉ ĉ } Ċĉĉ ĉ else Ġ-> Ġthrow ĠIllegalStateException (" Unknown Ġstate ") Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ private Ġfun Ġ< T Ġ: ĠAggregate Root > Ġbuild Aggregate From History ( type : ĠClass < T >, Ġevents : ĠList < Event Payload >, Ġversion : ĠLong , Ġid : ĠString , Ġsnapshot : ĠSnapshot ? Ġ= Ġnull ): ĠT Ġ{ Ċĉ ĉ val Ġadapter Ġ= ĠAggregate Adapter < T >(" apply ") Ċĉ ĉ adapter . fetch Metadata ( type ) Ċĉ ĉ val Ġhistory Ġ= Ġmutable ListOf < Any >() Ċĉ ĉ events . forEach Ġ{ Ċĉĉ ĉ if Ġ( message Format . supports Kind ( it . kind )) Ġ{ Ċĉĉĉ ĉ val Ġevent Ġ= Ġmessage Format . parse < Any >( ByteArray InputStream ( it . data . payload ), Ġit . kind ) Ċĉĉĉ ĉ history . add ( event ) Ċĉĉ ĉ } Ċĉ ĉ } ĊĊĉ ĉ /* Ċĉĉ Ġ* ĠCreate Ġa Ġnew Ġinstance Ġof Ġthe Ġaggregate Ċĉĉ Ġ*/ Ċĉ ĉ var Ġaggregate : ĠT Ċĉ ĉ try Ġ{ Ċĉĉ ĉ aggregate Ġ= Ġtype . newInstance () Ċĉĉ ĉ if Ġ( snapshot Ġ!= Ġnull ) Ġ{ Ċĉĉĉ ĉ aggregate Ġ= Ġaggregate . from Snapshot ( snapshot . data . payload , Ġsnapshot . version , Ġmessage Format ) Ġas ĠT Ċĉĉ ĉ } Ċĉ ĉ } Ġcatch Ġ( e : ĠInstanti ation Exception ) Ġ{ Ċĉĉ ĉ throw ĠH yd ration Exception ( id , Ġ" target Ġtype : Ġ'${ type . name }' Ġcannot Ġbe Ġinstantiated ") Ċĉ ĉ } Ġcatch Ġ( e : ĠIllegal AccessException ) Ġ{ Ċĉĉ ĉ throw ĠH yd ration Exception ( id , Ġ" target Ġtype : Ġ'${ type . name }' Ġhas Ġno Ġdefault Ġconstructor ") Ċĉ ĉ } Ċĉ ĉ aggregate . build From History ( history , Ġversion ) Ċĉ ĉ return Ġaggregate Ċ ĉ } Ċ }