package Ġcom . rich ode mus . autoplay list . event Ċ Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . do Return Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . mock Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . verify Ċ import Ġcom . rich ode mus . autoplay list . dto . RefreshToken Ċ import Ġcom . rich ode mus . autoplay list . dto . Spot ify UserId Ċ import Ġcom . rich ode mus . autoplay list . dto . UserId Ċ import Ġcom . rich ode mus . autoplay list . event . g cs . Google Cloud Storage Ċ import Ġcom . rich ode mus . autoplay list . event . g cs . Google Cloud Storage Adapter Ċ import Ġcom . rich ode mus . autoplay list . dto . events . Event Ċ import Ġcom . rich ode mus . autoplay list . dto . events . EventId Ċ import Ġcom . rich ode mus . autoplay list . dto . events . EventType Ċ import Ġcom . rich ode mus . autoplay list . dto . events . User Created Ċ import Ġio . micro meter . core . instrument . Counter Ċ import Ġio . micro meter . core . instrument . Meter Registry Ċ import Ġkotlinx . coroutines . Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Deferred Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . as Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . async Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . junit . Before Ċ import Ġorg . junit . Test Ċ import Ġjava . util . UUID Ċ import Ġjava . util . concurrent . Executors Ċ Ċ class ĠRe playing Event Store Test ĊĠĠĠ Ġ: ĠCor outine Scope Ġ{ ĊĠĠĠ Ġoverride Ġval Ġcoroutine Context Ġ= ĠDispatch ers . Default ĊĠĠĠ Ġ// language = JSON ĊĠĠĠ Ġprivate Ġval Ġevent Json Ġ= Ġ"{\" id \":\" 2 b 0 4 1 3 7 8 - 6 6 af - 4 cb 0 - a 2 b 0 - 1 8 6 fc 0 bdc 1 3 9 \",\" type \":\" USER _ CREATED \",\" timestamp \":\" 2 0 1 8 - 0 8 - 2 9 T 1 1 : 2 7 : 1 4 . 6 2 3 0 1 3 Z \",\" userId \":\" 2 b 0 4 1 3 7 8 - 6 6 af - 4 cb 0 - a 2 b 0 - 1 8 6 fc 0 bdc 1 3 9 \" }". trim Indent (). toByteArray () ĊĊĠĠĠ Ġprivate Ġval Ġevent Ġ= ĠUser Created ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠEvent Id ( UUID . fromString (" 2 b 0 4 1 3 7 8 - 6 6 af - 4 cb 0 - a 2 b 0 - 1 8 6 fc 0 bdc 1 3 9 ")), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠEventType . USER _ CREATED , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" 2 0 1 8 - 0 8 - 2 9 T 1 1 : 2 7 : 1 4 . 6 2 3 0 1 3 Z ", ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUserId ( UUID . fromString (" 2 b 0 4 1 3 7 8 - 6 6 af - 4 cb 0 - a 2 b 0 - 1 8 6 fc 0 bdc 1 3 9 ")) ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġregistry Mock : ĠMeter Registry ĊĊĠĠĠ Ġ@ Before ĊĠĠĠ Ġfun ĠsetUp () Ġ{ ĊĠĠĠĠĠĠĠ Ġregistry Mock Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġcounter (" events ") Ġ} Ġdo Return Ġmock < Counter > Ġ{} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Add ĠConsumer Ġafter Ġevent Ġis Ġproduced `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Mock Ġ= Ġmock < Google Cloud Storage > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġread () Ġ} Ġdo Return Ġempty List < Pair < Long , ĠDeferred < ByteArray >> >() ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= ĠRe playing Event Store ( Google Cloud Storage Adapter ( Event Ser de (), Ġclient Mock ), Ġregistry Mock ) ĊĊĠĠĠĠĠĠĠ Ġtarget . produce ( event ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġrun Blocking Ġ{ Ġtarget . read Events ( 1 ). await () Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat ( result ). contains Only ( event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Add ĠConsumer Ġbefore Ġevent Ġis Ġproduced `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Mock Ġ= Ġmock < Google Cloud Storage > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġread () Ġ} Ġdo Return Ġempty List < Pair < Long , ĠDeferred < ByteArray >> >() ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= ĠRe playing Event Store ( Google Cloud Storage Adapter ( Event Ser de (), Ġclient Mock ), Ġregistry Mock ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Async Ġ= Ġtarget . read Events ( 1 ) ĊĊĠĠĠĠĠĠĠ Ġtarget . produce ( event ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġrun Blocking Ġ{ Ġresult Async . await () Ġ} ĊĠĠĠĠĠĠĠ ĠassertThat ( result ). contains Only ( event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Create Ġstore Ġwith Ġpersisted Ġevents `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Mock Ġ= Ġmock < Google Cloud Storage > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġread () Ġ} Ġdo Return ĠlistOf ( 0 L Ġto Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Json ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= ĠRe playing Event Store ( Google Cloud Storage Adapter ( Event Ser de (), Ġclient Mock ), Ġregistry Mock ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġrun Blocking Ġ{ Ġtarget . read Events ( 1 ). await () Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat ( result ). contains Only ( event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Produce Ġevent `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Mock Ġ= Ġmock < Google Cloud Storage > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġread () Ġ} Ġdo Return Ġempty List < Pair < Long , ĠDeferred < ByteArray >> >() ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= ĠRe playing Event Store ( Google Cloud Storage Adapter ( Event Ser de (), Ġclient Mock ), Ġregistry Mock ) ĊĊĠĠĠĠĠĠĠ Ġtarget . produce ( event ) ĊĊĠĠĠĠĠĠĠ Ġverify ( client Mock ). write (" 0 ", Ġevent Json ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Period ically Ġadd Ġmultiple Ġconsumers Ġwhile Ġproducing `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexecutor Ġ= ĠExec utors . new Cached ThreadPool () ĊĠĠĠĠĠĠĠ Ġval Ġdispatcher Ġ= Ġexecutor . as Coroutine Dispatcher () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclient Mock Ġ= Ġmock < Google Cloud Storage > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġread () Ġ} Ġdo Return Ġempty List < Pair < Long , ĠDeferred < ByteArray >> >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevents Ġ= Ġ( 1 .. 5 _ 0 0 0 ). map Ġ{ Ġevent . copy ( timestamp Ġ= Ġit . toString ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtarget Ġ= ĠRe playing Event Store ( Google Cloud Storage Adapter ( Event Ser de (), Ġclient Mock ), Ġregistry Mock ) ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents . forEach Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtarget . produce ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresults Ġ= Ġ( 1 .. 1 0 0 ). map Ġ{ Ġindex Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( index . to Long ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtarget . read Events ( events . size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. map Ġ{ Ġit . await () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresults . forEach Ġ{ ĠassertThat ( it ). isEqualTo ( events ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor . shutdown () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠEvent Store . read Events ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcount : ĠInt , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdispatcher : ĠCor outine Dispatcher Ġ= ĠDispatch ers . Default ĊĠĠĠ Ġ): ĠDeferred < List < Event >> Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġevents Ġ= ĠlistOf < Event >() ĊĠĠĠĠĠĠĠ Ġthis . consume Ġ{ Ġevents Ġ= Ġevents Ġ+ Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġasync ( dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( events . size Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ async Ġevents ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ