/* Ċ Ġ* Ġ============ LICENSE _ START ================================================ ======= Ċ Ġ* Ġd cae gen 2 - collect ors - ves hv Ċ Ġ* Ġ================================================================= =============== Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 8 - 2 0 1 9 Ġ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 . tests . component Ċ Ċ import Ġarrow . core . None Ċ 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 . it Ċ import Ġorg . on ap . dca e . collect ors . ves hv . config . api . model . Collector Configuration Ċ import Ġorg . on ap . dca e . collect ors . ves hv . config . api . model . Routing Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event Domain . HE ART BE AT Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event Domain . MEAS UREMENT Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event Domain . OTHER Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Event Domain . PERF 3 G PP Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . ALTER N ATE _ PERF 3 G PP _ TOPIC Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . MAX _ PAYLOAD _ SIZE _ BYTES Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . PERF 3 G PP _ TOPIC Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . St oring Sink Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . basic Routing Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . f akes . two Domains ToOne Topic Routing Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . garbage Frame Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . message With Invalid Wire Frame Header Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . message With Payload Of Size Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . ves Wire Frame Message Ċ import Ġorg . on ap . dca e . collect ors . ves hv . tests . utils . wire Frame Message With Invalid Payload Ċ Ċ /** Ċ Ġ* Ġ@ 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 ĠV es Hv Specification Ġ: ĠSpe k ({ ĊĠĠĠ Ġdebug Rx ( false ) ĊĊĠĠĠ Ġdescribe (" V ES ĠHigh ĠVolume ĠCollector ") Ġ{ ĊĠĠĠĠĠĠĠ Ġit (" should Ġhandle Ġmultiple ĠH V ĠR AN Ġevents ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġsut . handle Connection ( sink , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" should Ġsend Ġall Ġevents ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. has Size ( 2 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġcreate Ġsink Ġl az ily ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġjust Ġconnecting Ġshould Ġnot Ġcreate Ġsink ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsut . handle Connection () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsut . close (). block () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( sink . closed ). isFalse () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġclose Ġsink Ġwhen Ġclosing Ġcollector Ġprovider ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġgiven ĠS ink Ġinitialized ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNote : Ġas ĠSt oring Sink Ġis Ġ( ho pe fully ) Ġcreated Ġl az ily , Ġ" valid " Ġv es Ġmessage Ġneeds Ġto Ġbe Ġsent ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsut . handle Connection ( ves Wire Frame Message ( PERF 3 G PP )) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsut . close (). block () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( sink . closed ). isTrue () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdescribe (" Memory Ġmanagement ") Ġ{ ĊĠĠĠĠĠĠĠ Ġit (" should Ġrelease Ġmemory Ġfor Ġeach Ġhandled Ġand Ġdropped Ġmessage ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġvalid Message Ġ= Ġv es Wire Frame Message ( PERF 3 G PP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg With Invalid Frame Ġ= Ġmessage With Invalid Wire Frame Header () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg With Too Big Payload Ġ= Ġmessage With Payload Of Size ( MAX _ PAYLOAD _ SIZE _ BYTES Ġ+ Ġ 1 , ĠPER F 3 G PP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexpected Ref Cnt Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhandled Events Ġ= Ġsut . handle Connection ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsink , Ġvalid Message , Ġmsg With Invalid Frame , Ġmsg With Too Big Payload ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( handled Events ). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( valid Message . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" handled Ġmessage Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg With Invalid Frame . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" message Ġwith Ġinvalid Ġframe Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg With Too Big Payload . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" message Ġwith Ġpayload Ġexceed ing Ġ 1 M iB Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġrelease Ġmemory Ġfor Ġeach Ġmessage Ġwith Ġinvalid Ġpayload ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġvalid Message Ġ= Ġv es Wire Frame Message ( PERF 3 G PP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg With Invalid Payload Ġ= Ġwire Frame Message With Invalid Payload () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexpected Ref Cnt Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhandled Events Ġ= Ġsut . handle Connection ( sink , Ġvalid Message , Ġmsg With Invalid Payload ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( handled Events ). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( valid Message . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" handled Ġmessage Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg With Invalid Payload . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" message Ġwith Ġinvalid Ġpayload Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġrelease Ġmemory Ġfor Ġeach Ġmessage Ġwith Ġgarbage Ġframe ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġvalid Message Ġ= Ġv es Wire Frame Message ( PERF 3 G PP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg With G arbage Frame Ġ= Ġgarbage Frame () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexpected Ref Cnt Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhandled Events Ġ= Ġsut . handle Connection ( sink , Ġvalid Message , Ġmsg With G arbage Frame ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( handled Events ). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( valid Message . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" handled Ġmessage Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg With G arbage Frame . ref Cnt ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. described As (" message Ġwith Ġgarbage Ġframe Ġshould Ġbe Ġreleased ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( expected Ref Cnt ) ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdescribe (" message Ġrouting ") Ġ{ ĊĠĠĠĠĠĠĠ Ġit (" should Ġdirect Ġmessage Ġto Ġa Ġtopic Ġby Ġmeans Ġof Ġrouting Ġconfiguration ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġsut . handle Connection ( sink , Ġv es Wire Frame Message ( PERF 3 G PP )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages ). described As (" number Ġof Ġro uted Ġmessages "). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġmessages [ 0 ] ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg . target Topic ). described As (" ro uted Ġmessage Ġtopic "). isEqualTo ( PERF 3 G PP _ TOPIC ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg . partition ). described As (" ro uted Ġmessage Ġpartition "). isEqualTo ( None ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġbe Ġable Ġto Ġdirect Ġ 2 Ġmessages Ġfrom Ġdifferent Ġdomains Ġto Ġone Ġtopic ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink ( two Domains ToOne Topic Routing ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġsut . handle Connection ( sink , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( HE ART BE AT ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( MEAS UREMENT )) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages ). described As (" number Ġof Ġro uted Ġmessages "). has Size ( 3 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages [ 0 ]. target Topic ). described As (" first Ġmessage Ġtopic ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( PERF 3 G PP _ TOPIC ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages [ 1 ]. target Topic ). described As (" second Ġmessage Ġtopic ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( PERF 3 G PP _ TOPIC ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages [ 2 ]. target Topic ). described As (" last Ġmessage Ġtopic ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( ALTER N ATE _ PERF 3 G PP _ TOPIC ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġit (" should Ġdrop Ġmessage Ġif Ġroute Ġwas Ġnot Ġfound ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġsut . handle Connection ( sink , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( OTHER , Ġ" first "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP , Ġ" second "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( HE ART BE AT , Ġ" third ")) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( messages ). described As (" number Ġof Ġro uted Ġmessages "). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġmessages [ 0 ] ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg . target Topic ). described As (" ro uted Ġmessage Ġtopic "). isEqualTo ( PERF 3 G PP _ TOPIC ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( msg . message . header . eventId ). described As (" ro uted Ġmessage Ġevent Id "). isEqualTo (" second ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdescribe (" request Ġvalidation ") Ġ{ ĊĠĠĠĠĠĠĠ Ġit (" should Ġreject Ġmessage Ġwith Ġpayload Ġgreater Ġthan Ġ 1 ĠM iB Ġand Ġall Ġsubsequent Ġmessages ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( sut , Ġsink ) Ġ= Ġv es Hv With St oring Sink () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhandled Messages Ġ= Ġsut . handle Connection ( sink , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP , Ġ" first "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage With Payload Of Size ( MAX _ PAYLOAD _ SIZE _ BYTES Ġ+ Ġ 1 , ĠPER F 3 G PP ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġv es Wire Frame Message ( PERF 3 G PP )) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( handled Messages ). has Size ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( handled Messages . first (). message . header . eventId ). isEqualTo (" first ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }) Ċ Ċ private Ġfun Ġv es Hv With St oring Sink ( routing : ĠRouting Ġ= Ġbasic Routing ): ĠPair < S ut , ĠSt oring Sink > Ġ{ ĊĠĠĠ Ġval Ġsink Ġ= ĠSt oring Sink () ĊĠĠĠ Ġval Ġsut Ġ= ĠS ut ( Collector Configuration ( routing , ĠMAX _ PAYLOAD _ SIZE _ BYTES ), Ġsink ) ĊĠĠĠ Ġreturn ĠPair ( sut , Ġsink ) Ċ } Ċ