/* Ċ Ġ* Ġ============ LICENSE _ START ================================================ ======= Ċ Ġ* Ġd cae gen 2 - collect ors - ves hv Ċ Ġ* Ġ================================================================= =============== Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 8 - 2 0 2 1 ĠN OK IA Ċ Ġ* Ġ================================================================= =============== Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ* Ġ============ LICENSE _ END ================================================ ========= Ċ Ġ*/ Ċ package Ġorg . on ap . dca e . collect ors . ves hv . impl Ċ Ċ import Ġarrow . core . None Ċ 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 . n ha ar man . mock it ok otlin 2 . wh enever Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . jetbrains . spe k . api . Spe k Ċ import Ġorg . jetbrains . spe k . api . dsl . describe Ċ import Ġorg . jetbrains . spe k . api . dsl . given Ċ import Ġorg . jetbrains . spe k . api . dsl . it Ċ import Ġorg . jetbrains . spe k . api . dsl . on Ċ import Ġorg . on ap . dca e . collect ors . ves hv . config . api . model . Route Ċ import Ġorg . on ap . dca e . collect ors . ves hv . config . api . model . Routing Ċ import Ġorg . on ap . dca e . collect ors . ves hv . boundary . Sink Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . Rout ed Message Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event Domain Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event St nd Defined Namespace Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . logging . ClientContext Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Message Ċ import Ġorg . on ap . dca e . collect ors . ves hv . model . Successfully Cons umed Message Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . common Header Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . empty Wire Protocol Frame Ċ import Ġorg . on ap . dca e gen 2 . services . sdk . model . streams . dma ap . Kafka Sink Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . test . Step Verifier ĊĊ Ċ /** Ċ Ġ* Ġ@ author ĠP io tr ĠJ as zc zy k Ġ< pio tr . jas zc zy k @ nok ia . com > Ċ Ġ* Ġ@ since ĠMay Ġ 2 0 1 8 Ċ Ġ*/ Ċ object ĠRouter Test Ġ: ĠSpe k ({ ĊĊĠĠĠ Ġdescribe (" Router ") Ġ{ ĊĊĠĠĠĠĠĠĠ Ġwhenever ( perf 3 g pp Sink Mock . topic Name ()). thenReturn ( perf 3 g pp Topic ) ĊĠĠĠĠĠĠĠ Ġwhenever ( ves 3 g pp Heartbeat Sink Mock . topic Name ()). thenReturn ( ves 3 g pp Heartbeat Topic ) ĊĠĠĠĠĠĠĠ Ġwhenever ( syslog Sink Mock . topic Name ()). thenReturn ( syslog Topic ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Sink Map Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( perf 3 g pp Topic , Ġlazy Of ( message Sink Mock )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( ves 3 g pp Heartbeat Topic , Ġlazy Of ( message Sink Mock )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( syslog Topic , Ġlazy Of ( message Sink Mock )) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġgiven (" sample Ġrouting Ġspecification ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcut Ġ= Ġrouter ( default Routing , Ġmessage Sink Map ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġon (" message Ġwith Ġexisting Ġroute Ġ( rtp m )") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhenever ( message Sink Mock . send ( ro uted Perf 3 G pp Message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( Flux . just ( successfully Cons umed Perf 3 g pp Message )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit (" should Ġbe Ġproperly Ġro uted ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġcut . route ( perf 3 g pp Message ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( result ). isNotNull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStep Verifier . create ( result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. expect Next ( successfully Cons umed Perf 3 g pp Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. verify Complete () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( perf 3 g pp Sink Mock ). topic Name () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( message Sink Mock ). send ( ro uted Perf 3 G pp Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġon (" message Ġwith Ġexisting Ġroute Ġ( syslog )") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhenever ( message Sink Mock . send ( ro uted Sys log Message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( Flux . just ( successfully Cons umed Sys log Message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġcut . route ( syslog Message ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit (" should Ġbe Ġproperly Ġro uted ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStep Verifier . create ( result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. expect Next ( successfully Cons umed Sys log Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. verify Complete () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( syslog Sink Mock ). topic Name () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( message Sink Mock ). send ( ro uted Sys log Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġon (" message Ġwith Ġexisting Ġst nd Defined Ġroute Ġ( ves 3 g pp Heartbeat )") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhenever ( message Sink Mock . send ( ro uted V es 3 g pp Heartbeat Message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( Flux . just ( successfully Cons umed V es 3 g pp Heartbeat Message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġcut . route ( ves 3 g pp Heartbeat Message ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit (" should Ġbe Ġproperly Ġro uted ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStep Verifier . create ( result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. expect Next ( successfully Cons umed V es 3 g pp Heartbeat Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. verify Complete () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( ves 3 g pp Heartbeat Sink Mock ). topic Name () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġverify ( message Sink Mock ). send ( ro uted V es 3 g pp Heartbeat Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġon (" message Ġwith Ġunknown Ġroute ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= ĠV es Message ( common Header ( V es Event Domain . HE ART BE AT ), Ġempty Wire Protocol Frame ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġcut . route ( message ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit (" should Ġnot Ġhave Ġroute Ġavailable ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStep Verifier . create ( result ). verify Complete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }) Ċ Ċ private Ġfun Ġrouter ( routing : ĠRouting , Ġkafka Publisher Map : ĠMap < String , ĠLazy < Sink >>) Ġ= ĊĠĠĠĠĠĠĠ ĠRouter ( routing , Ġkafka Publisher Map , ĠClient Context (), Ġmock ()) Ċ Ċ private Ġconst Ġval Ġperf 3 g pp Topic Ġ= Ġ" PERF _ PERF " Ċ private Ġval Ġperf 3 g pp Sink Mock Ġ= Ġmock < Kafka Sink >() Ċ private Ġval Ġdefault 3 g pp Route Ġ= ĠRoute ( V es Event Domain . PERF 3 G PP . domain Name , Ġperf 3 g pp Sink Mock ) Ċ Ċ private Ġconst Ġval Ġv es 3 g pp Heartbeat Topic Ġ= Ġ" SEC _ 3 G PP _ HE ART BE AT _ OUTPUT " Ċ private Ġval Ġv es 3 g pp Heartbeat Sink Mock Ġ= Ġmock < Kafka Sink >() Ċ private Ġval Ġdefault V es 3 g pp Heartbeat Route Ġ= ĊĠĠĠĠĠĠĠ ĠRoute ( V es Event St nd Defined Namespace . V ES _ 3 G PP _ HE ART BE AT . st nd Defined Namespace , Ġv es 3 g pp Heartbeat Sink Mock ) Ċ Ċ private Ġconst Ġval Ġsyslog Topic Ġ= Ġ" SYS _ LOG " Ċ private Ġval Ġsyslog Sink Mock Ġ= Ġmock < Kafka Sink >() Ċ private Ġval Ġdefault Sys log Route Ġ= ĠRoute ( V es Event Domain . SYS LOG . domain Name , Ġsyslog Sink Mock ) Ċ Ċ private Ġval Ġdefault Routing Ġ= ĠlistOf ( default 3 g pp Route , Ġdefault V es 3 g pp Heartbeat Route , Ġdefault Sys log Route ) ĊĊ Ċ private Ġval Ġmessage Sink Mock Ġ= Ġmock < Sink >() Ċ private Ġval Ġdefault _ partition Ġ= ĠNone Ċ Ċ private Ġval Ġperf 3 g pp Message Ġ= ĠV es Message ( common Header ( V es Event Domain . PERF 3 G PP ), Ġempty Wire Protocol Frame ()) Ċ private Ġval Ġro uted Perf 3 G pp Message Ġ= ĠRout ed Message ( perf 3 g pp Message , Ġperf 3 g pp Topic , Ġdefault _ partition ) Ċ private Ġval Ġsuccessfully Cons umed Perf 3 g pp Message Ġ= ĠSuccessfully Cons umed Message ( ro uted Perf 3 G pp Message ) Ċ Ċ private Ġval Ġv es 3 g pp Heartbeat Message Ġ= ĠV es Message ( common Header ( domain Ġ= ĠV es Event Domain . ST ND _ DEFINED , ĊĠĠĠĠĠĠĠ Ġst nd Defined Namespace Ġ= ĠV es Event St nd Defined Namespace . V ES _ 3 G PP _ HE ART BE AT ), Ġempty Wire Protocol Frame ()) Ċ private Ġval Ġro uted V es 3 g pp Heartbeat Message Ġ= ĠRout ed Message ( ves 3 g pp Heartbeat Message , Ġv es 3 g pp Heartbeat Topic , Ġdefault _ partition ) Ċ private Ġval Ġsuccessfully Cons umed V es 3 g pp Heartbeat Message Ġ= ĠSuccessfully Cons umed Message ( ro uted V es 3 g pp Heartbeat Message ) Ċ Ċ private Ġval Ġsyslog Message Ġ= ĠV es Message ( common Header ( V es Event Domain . SYS LOG ), Ġempty Wire Protocol Frame ()) Ċ private Ġval Ġro uted Sys log Message Ġ= ĠRout ed Message ( syslog Message , Ġsyslog Topic , Ġdefault _ partition ) Ċ private Ġval Ġsuccessfully Cons umed Sys log Message Ġ= ĠSuccessfully Cons umed Message ( ro uted Sys log Message )