/** Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 ĠE mer ald Pay , ĠInc Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 1 9 ĠE TC DEV ĠGmbH Ċ Ġ* Ċ Ġ* Ġ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 . Ċ Ġ*/ Ċ package Ġio . em eral dp ay . d sh ack le . upstream . ethereum Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġio . em eral dp ay . d sh ack le . Defaults Ċ import Ġio . em eral dp ay . d sh ack le . Global Ċ import Ġio . em eral dp ay . d sh ack le . config . Up streams Config Ċ import Ġio . em eral dp ay . d sh ack le . upstream . Upstream Availability Ċ import Ġio . em eral dp ay . d sh ack le . upstream . rp cc lient . Json Rpc Request Ċ import Ġio . em eral dp ay . d sh ack le . upstream . rp cc lient . Json Rpc Response Ċ import Ġio . inf init ape . ether jar . rpc . json . Sync ing Json Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . scheduling . concurrent . Custom izable Thread Factory Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono Ċ import Ġreactor . core . scheduler . Schedulers Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . Executors Ċ Ċ class ĠEthereum Upstream Validator ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġupstream : ĠEthereum Upstream , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġoptions : ĠUp streams Config . Options Ċ ) Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( E thereum Upstream Validator :: class . java ) ĊĠĠĠĠĠĠĠ Ġval Ġscheduler Ġ= ĠS chedulers . from Executor ( Executors . new Cached ThreadPool ( Custom izable Thread Factory (" ethereum - validator "))) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval ĠobjectMapper : ĠObjectMapper Ġ= ĠGlobal . object Mapper ĊĊĠĠĠ Ġfun Ġvalidate (): ĠMono < Upstream Availability > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġupstream ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. get Api () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. read ( Json Rpc Request (" eth _ sync ing ", ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap ( Json Rpc Response :: require Result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠobjectMapper . readValue ( it , ĠSync ing Json :: class . java ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. timeout ( Defaults . timeout Internal , ĠMono . error ( Exception (" Validation Ġtimeout Ġfor ĠSync ing "))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( value . is Sync ing ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . just ( Upstream Availability . SYNC ING ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupstream ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. get Api () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. read ( Json Rpc Request (" net _ peer Count ", ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap ( Json Rpc Response :: require String Result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map ( Integer :: decode ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. timeout ( Defaults . timeout Internal , ĠMono . error ( Exception (" Validation Ġtimeout Ġfor ĠPe ers "))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġcount Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmin Peers Ġ= Ġoptions . min Peers Ġ?: Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( count Ġ< Ġmin Peers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠUp stream Availability . IM M ATURE ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠUp stream Availability . OK ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Return ( Upstream Availability . UN AVAILABLE ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstart (): ĠFlux < Upstream Availability > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFlux . interval ( Duration . of Seconds ( 1 5 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalidate () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }