package Ġde . br ing me ister . spring . aws . k inesis Ċ Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . junit . Class Rule Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . runner . RunWith Ċ import Ġorg . springframework . beans . factory . annotation . Autowired Ċ import Ġorg . springframework . boot . test . context . Test Configuration Ċ import Ġorg . springframework . context . annotation . Bean Ċ import Ġorg . springframework . context . annotation . Import Ċ import Ġorg . springframework . test . context . junit 4 . Spring Runner Ċ import Ġjava . time . Instant . now Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ @ RunWith ( Spring Runner :: class ) Ċ @ Container Test Ċ @ Import ( Kotlin Listener Test . Config :: class ) Ċ class ĠKotlin Listener Test Ġ{ ĊĊĠĠĠ Ġ@ Autowired ĊĠĠĠ Ġlateinit Ġvar Ġoutbound : ĠAws K inesis Outbound Gateway ĊĊĠĠĠ Ġ@ Test Configuration ĊĠĠĠ Ġclass ĠConfig Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ@ Bean ĊĠĠĠĠĠĠĠ Ġfun Ġevent Handler () Ġ= Ġobject Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ K inesis Listener ( STREAM ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfun Ġhandle ( data : ĠString , Ġmetadata : ĠString ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( data ). isEqualTo ( EXPECTED _ DATA ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( metadata ). isEqualTo ( EXPECTED _ METADATA ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Records Counter . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ K inesis Listener ( stream Ġ= ĠB ATCH _ STREAM , Ġdata Class Ġ= ĠData :: class , Ġmeta Class Ġ= ĠMetadata :: class ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfun Ġhandle Batch ( events : ĠMap < Data , ĠMetadata >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( events ). has Size ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( events . entries . first (). key . data ). isEqualTo ( EXPECTED _ DATA ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( events . entries . first (). value . type ). isEqualTo ( EXPECTED _ METADATA ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Records Counter . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Class Rule ĊĠĠĠĠĠĠĠ Ġ@ Jvm Field ĊĠĠĠĠĠĠĠ Ġval Ġk inesis Ġ= ĠCont ainers . k inesis () ĊĊĠĠĠĠĠĠĠ Ġ@ Class Rule ĊĠĠĠĠĠĠĠ Ġ@ Jvm Field ĊĠĠĠĠĠĠĠ Ġval Ġd ynamodb Ġ= ĠCont ainers . d ynamo Db () ĊĊĠĠĠĠĠĠĠ Ġconst Ġval ĠST REAM Ġ= Ġ" foo - event - stream " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠB ATCH _ STREAM Ġ= Ġ" foo - event - stream - batch " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠEXPECT ED _ DATA Ġ= Ġ" my - data " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠEXPECT ED _ METADATA Ġ= Ġ" my - metadata " ĊĊĠĠĠĠĠĠĠ Ġlateinit Ġvar Ġexpected Records Counter : ĠCount DownLatch ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġsend Ġand Ġreceive Ġevents `() Ġ{ ĊĠĠĠĠĠĠĠ Ġexpected Records Counter Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠĠĠ Ġoutbound . send ( STREAM , ĠRecord ( EXPECTED _ DATA , ĠEXPECT ED _ METADATA )) ĊĊĠĠĠĠĠĠĠ Ġval Ġrecords Processed Ġ= ĠwaitFor Records ToBe Processed () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( records Processed ). isTrue () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġsend Ġand Ġreceive Ġevents Ġin Ġa Ġbatch `() Ġ{ ĊĠĠĠĠĠĠĠ Ġexpected Records Counter Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠĠĠ Ġoutbound . send ( BATCH _ STREAM , ĠRecord ( Data ( EXPECTED _ DATA , Ġnow (). to Epoch Milli ()), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMetadata ( EXPECTED _ METADATA , Ġnow (). to Epoch Milli ()))) ĊĊĠĠĠĠĠĠĠ Ġval Ġrecords Processed Ġ= ĠwaitFor Records ToBe Processed () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( records Processed ). isTrue () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠwaitFor Records ToBe Processed () Ġ= Ġexpected Records Counter . await ( 1 , ĠTimeUnit . MIN UTES ) Ċ } Ċ Ċ data Ġclass ĠData ( val Ġdata : ĠString , Ġval Ġdate : ĠLong ) Ċ data Ġclass ĠMetadata ( val Ġtype : ĠString , Ġval Ġdate : ĠLong ) ĊĊ