package Ġdagger ok Ċ Ċ import Ġorg . springframework . beans . factory . annotation . Value Ċ import Ġorg . springframework . boot . Application Runner Ċ import Ġorg . springframework . boot . Spring Application Ċ import Ġorg . springframework . boot . autoconfigure . SpringBoot Application Ċ import Ġorg . springframework . context . annotation . Bean Ċ import Ġorg . springframework . context . annotation . Configuration Ċ import Ġorg . springframework . web . reactive . function . client . Exchange Filter Functions Ċ import Ġorg . springframework . web . reactive . function . client . Web Client Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono Ċ import Ġjava . time . Duration Ċ import Ġjava . time . LocalDateTime Ċ Ċ @ Configuration Ċ class ĠWeb Client Configuration Ġ{ ĊĊĠ Ġ@ Value ("\ ${ app . user - service . host }") ĊĠ Ġvar Ġservice Host : ĠString ? Ġ= Ġnull ĊĊĠ Ġ@ Value ("\ ${ app . user - service . username }") ĊĠ Ġvar Ġservice Username : ĠString ? Ġ= Ġnull ĊĊĠ Ġ@ Value ("\ ${ app . user - service . password }") ĊĠ Ġvar Ġservice Password : ĠString ? Ġ= Ġnull ĊĊĠ Ġ@ Bean ĊĠ Ġfun Ġuser Web Client () Ġ= ĠWeb Client ĊĠĠĠĠĠ Ġ. create (" http :// $ service Host : 8 0 8 0 ") ĊĠĠĠĠĠ Ġ. mutate () ĊĠĠĠĠĠ Ġ. filter ( Exchange Filter Functions . basic Authentication ( service Username !! , Ġservice Password !! )) ĊĠĠĠĠĠ Ġ. build () Ċ } Ċ Ċ data Ġclass ĠUser ( var Ġid : ĠString ? Ġ= Ġnull , Ġvar Ġname : ĠString ? Ġ= Ġnull ) Ċ data Ġclass ĠUser Event ( var ĠuserId : ĠString ? Ġ= Ġnull , Ġvar Ġat : ĠLocalDateTime ? Ġ= Ġnull ) Ċ Ċ @ SpringBoot Application Ċ class ĠClient Application Ġ{ ĊĊĠ Ġ@ Bean ĊĠ Ġfun Ġrunner ( user Web Client : ĠWeb Client ) Ġ= ĠApplication Runner Ġ{ ĊĊĠĠĠ Ġval Ġdelay Startup Ġ= ĠMono . delay ( Duration . of Seconds ( 5 )) ĊĠĠĠ Ġval Ġevent Stream Ġ= Ġuser Web Client ĊĠĠĠĠĠĠĠ Ġ. get () ĊĠĠĠĠĠĠĠ Ġ. uri ("") ĊĠĠĠĠĠĠĠ Ġ. retrieve () ĊĠĠĠĠĠĠĠ Ġ. body To Flux ( User :: class . java ) ĊĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġuser Web Client ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. get () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. uri ("/ { id }/ events ", Ġit . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. retrieve () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body To Flux ( User Event :: class . java ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ Ġprintln (" f * ck : Ġ${ it . localized Message }") Ġ} ĊĊĠĠĠ ĠFlux . concat ( delay Startup , Ġevent Stream ) ĊĠĠĠĠĠĠĠ Ġ. skip ( 1 ) Ġ// Ġskip Ġdelay Startup Ġsubscription ĊĠĠĠĠĠĠĠ Ġ. subscribe ({ Ġprintln (" subscribe : Ġ$ it \ n ") Ġ}) ĊĠ Ġ} Ċ } Ċ Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠ ĠSpring Application . run ( Client Application :: class . java , Ġ* args ) Ċ } Ċ