package Ġmob i . water dog . event bus Ċ Ċ import Ġcom . zaxxer . hikari . HikariConfig Ċ import Ġcom . zaxxer . hikari . Hikari DataSource Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġmob i . water dog . event bus . model . Event Input Ċ import Ġmob i . water dog . event bus . sql . Database Connection Ċ import Ġmob i . water dog . event bus . sql . Event Table Ċ import Ġmob i . water dog . event bus . sql . Local Event Store Sql Ċ import Ġorg . am sh ove . k lu ent .` should Ġbe Ġtrue ` Ċ import Ġorg . am sh ove . k lu ent .` should Ġequal ` Ċ import Ġorg . jetbrains . ex posed . sql . Schema Utils Ċ import Ġorg . junit . jupiter . api . After All Ċ import Ġorg . junit . jupiter . api . After Each Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . Test Instance Ċ import Ġorg . k oin . test . K oin Test Ċ Ċ @ Test Instance ( Test Instance . Lifecycle . PER _ CLASS ) Ċ class ĠEvent Bus Test Ġ: ĠK oin Test Ġ{ ĊĊĠĠĠ Ġ@ After All ĊĠĠĠ Ġfun Ġafter All () Ġ{ ĊĠĠĠĠĠĠĠ Ġebp . shutdown () ĊĠĠĠĠĠĠĠ Ġdb c . query Ġ{ ĠSchema Utils . drop ( Event Table ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġtarget Topic 1 Ġ= Ġ" My Topic 1 " ĊĠĠĠ Ġprivate Ġval Ġtarget Topic 2 Ġ= Ġ" My Topic 2 " ĊĠĠĠ Ġprivate Ġval Ġebp Ġ= ĠEvent Bus Provider ( Event Backend . Local ) ĊĠĠĠ Ġprivate Ġval ĠdataSource Ġ= ĠH ikari DataSource ( HikariConfig (). apply Ġ{ ĊĠĠĠĠĠĠĠ Ġdriver ClassName Ġ= Ġ" org . h 2 . Driver " ĊĠĠĠĠĠĠĠ Ġjdbc Url Ġ= Ġ" jdbc : h 2 : mem : test " ĊĠĠĠĠĠĠĠ Ġmaximum PoolSize Ġ= Ġ 5 ĊĠĠĠĠĠĠĠ Ġis Auto Commit Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġtransaction Isolation Ġ= Ġ" TRANSACTION _ REPEAT ABLE _ READ " ĊĠĠĠĠĠĠĠ Ġvalidate () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġprivate Ġval Ġdb c Ġ= ĠDatabase Connection ( dataSource ) ĊĠĠĠ Ġprivate Ġval Ġlocal Event Store Ġby Ġlazy Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdb c . query Ġ{ ĠSchema Utils . create ( Event Table ) Ġ} ĊĠĠĠĠĠĠĠ ĠLocal Event Store Sql ( dbc ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġval Ġevent Producer : ĠEvent Producer Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġebp . setup Producer ( local Event Store ) ĊĠĠĠĠĠĠĠ Ġebp . get Producer () ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġval Ġevent Consumer : ĠEvent Consumer Ġby Ġlazy Ġ{ Ġebp . get Consumer () Ġ} ĊĊĠĠĠ Ġ@ After Each ĊĠĠĠ Ġfun Ġcleanup () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠlistOf ( target Topic 1 , Ġtarget Topic 2 ). forEach Ġ{ Ġtopic Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Consumer . list Sub scribers ( topic ). forEach Ġ{ Ġconsumer Id Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Consumer . unsubscribe ( topic , Ġconsumer Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Consumer . list Sub scribers ( topic ). size Ġ` should Ġequal ` Ġ 0 ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` The Ġproducer Ġis Ġable Ġto Ġsend Ġevents Ġto Ġa Ġtopic `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġjson Msg Ġ= Ġ"{\" foo \":\" bar \" }" ĊĊĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġ 1 0 ĊĠĠĠĠĠĠĠ Ġevent Consumer . stream ( target Topic 1 ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString ( it . payload ) Ġ` should Ġequal ` Ġjson Msg ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrepeat ( n ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Ġ= ĠEvent Input ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtarget Topic 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Sample ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" application / json ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjson Msg . toByteArray () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Producer . send ( event ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` The Ġproducer Ġis Ġable Ġto Ġsend Ġand Ġwait Ġfor Ġack Ġevents Ġto Ġa Ġtopic `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġjson Msg Ġ= Ġ"{\" foo \":\" bar \" }" ĊĠĠĠĠĠĠĠ Ġvar Ġcount Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġ 1 0 ĊĠĠĠĠĠĠĠ Ġevent Consumer . stream ( target Topic 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . msg Type Ġ== Ġ" Sample " Ġ&& Ġit . mime Type Ġ== Ġ" application / json " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠString ( it . payload ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit Ġ` should Ġequal ` Ġjson Msg ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcount Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġevent Consumer . list Sub scribers ( target Topic 2 ). size Ġ` should Ġequal ` Ġ 1 ĊĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( n ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Ġ= ĠEvent Input ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtarget Topic 2 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Sample ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" application / json ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjson Msg . toByteArray () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġevent Producer . send ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult .` should Ġbe Ġtrue `() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcount Ġ` should Ġequal ` Ġn ĊĠĠĠ Ġ} Ċ }