package Ġstreams Ċ Ċ import Ġorg . apache . commons . lang 3 . StringUtils Ċ import Ġorg . neo 4 j . logging . Log Ċ import Ġstreams . config . Streams Config Ċ import Ġstreams . events . EntityType Ċ import Ġstreams . events . Rel Key Strategy ĊĊ Ċ private Ġinline Ġfun Ġ< re ified ĠT > Ġfilter Map ( config : ĠMap < String , ĠString >, Ġrouting Prefix : ĠString , Ġdb Name : ĠString Ġ= Ġ"", Ġrouting Suffix : ĠString ? Ġ= Ġnull , Ġlog : ĠLog ? Ġ= Ġnull ): ĠList < T > Ġ{ ĊĠĠĠ Ġval Ġentity Type Ġ= Ġwhen Ġ( T :: class ) Ġ{ ĊĠĠĠĠĠĠĠ ĠNode Routing Configuration :: class Ġ-> ĠEntityType . node ĊĠĠĠĠĠĠĠ ĠRelationship Routing Configuration :: class Ġ-> ĠEntityType . relationship ĊĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalArgumentException (" The Ġclass Ġmust Ġbe Ġan Ġinstance Ġof ĠRouting Configuration ") ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn Ġconfig ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Keys Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstart With Prefix And Not End With Suffix Ġ= Ġit . startsWith ( routing Prefix ) Ġ&& Ġrouting Suffix ?. let Ġ{ Ġsuffix Ġ-> Ġ! it . endsWith ( suffix ) Ġ} Ġ?: Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . contains ( Streams Routing Configuration Constants . FROM )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtopic Db Name Ġ= Ġit . replace ( routing Prefix , ĠStringUtils . EMPTY ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. split ( Streams Routing Configuration Constants . FROM )[ 1 ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart With Prefix And Not End With Suffix Ġ&& Ġtopic Db Name Ġ== Ġdb Name Ġ// Ġfor Ġ` from .< db >` Ġwe Ġcompare Ġthe Ġrouting Ġprefix Ġand Ġthe Ġdb Ġname ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġfor Ġthe Ġdefault Ġdb Ġwe Ġonly Ġfilter Ġby Ġrouting Prefix ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdb Name Ġ== Ġ"" Ġ&& Ġstart With Prefix And Not End With Suffix ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġprefix And Topic Ġ= Ġit . key . split ( Streams Routing Configuration Constants . FROM )[ 0 ] ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Strategy Ġ= Ġrouting Suffix ?. let Ġ{ Ġsuffix Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprint (" suffix Ġ- Ġ$ suffix ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . entries . first OrNull { Ġit . key . startsWith ( prefix And Topic ) Ġ&& Ġit . key . endsWith ( suffix ) Ġ} ?. value ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ?: ĠRel Key Strategy . DEFAULT . toString (). toLowerCase () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRouting Configuration Factory ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. get Routing Configuration ( prefix And Topic . replace ( routing Prefix , ĠStringUtils . EMPTY ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . value , Ġentity Type , Ġkey Strategy , Ġlog ) Ġas ĠList < T > ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ } Ċ Ċ private Ġobject ĠStreams Routing Configuration Constants Ġ{ ĊĠĠĠ Ġconst Ġval ĠNODE _ ROUT ING _ KEY _ PREFIX : ĠString Ġ= Ġ" streams . source . topic . nodes ." ĊĠĠĠ Ġconst Ġval ĠREL _ ROUT ING _ KEY _ PREFIX : ĠString Ġ= Ġ" streams . source . topic . relationships ." ĊĠĠĠ Ġconst Ġval ĠS CHE MA _ POL LING _ INTERVAL Ġ= Ġ" streams . source . schema . polling . interval " ĊĠĠĠ Ġconst Ġval ĠFROM Ġ= Ġ". from ." ĊĠĠĠ Ġconst Ġval ĠKEY _ STR ATE GY _ SUFFIX Ġ= Ġ". key _ strategy " Ċ } Ċ Ċ data Ġclass ĠStreams Event Router Configuration ( val Ġenabled : ĠBoolean Ġ= ĠStreams Config . SOURCE _ ENABLED _ VALUE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġprocedures Enabled : ĠBoolean Ġ= ĠStreams Config . PROC EDURE S _ ENABLED _ VALUE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnode Routing : ĠList < Node Routing Configuration > Ġ= ĠlistOf ( Node Routing Configuration ()), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrel Routing : ĠList < Relationship Routing Configuration > Ġ= ĠlistOf ( Relationship Routing Configuration ()), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġschema Polling Interval : ĠLong Ġ= Ġ 3 0 0 0 0 0 ) Ġ{ ĊĊĠĠĠ Ġfun Ġall Topics (): ĠList < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnode Topics Ġ= Ġnode Routing . map Ġ{ Ġit . topic Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġrel Topics Ġ= Ġrel Routing . map Ġ{ Ġit . topic Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġnode Topics Ġ+ Ġrel Topics ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĊĠĠĠĠĠĠĠ Ġfun Ġfrom ( streams Config : ĠMap < String , ĠString >, Ġdb Name : ĠString , Ġis Default Db : ĠBoolean , Ġlog : ĠLog ? Ġ= Ġnull ): ĠStreams Event Router Configuration Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġnode Routing Ġ= Ġfilter Map < Node Routing Configuration >( config Ġ= Ġstreams Config , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Prefix Ġ= ĠStreams Routing Configuration Constants . NODE _ ROUT ING _ KEY _ PREFIX , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdb Name Ġ= Ġdb Name ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġrel Routing Ġ= Ġfilter Map < Relationship Routing Configuration >( config Ġ= Ġstreams Config , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Prefix Ġ= ĠStreams Routing Configuration Constants . REL _ ROUT ING _ KEY _ PREFIX , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdb Name Ġ= Ġdb Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Suffix Ġ= ĠStreams Routing Configuration Constants . KEY _ STR ATE GY _ SUFFIX , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Ġ= Ġlog ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Default Db ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnode Routing Ġ+= Ġfilter Map ( config Ġ= Ġstreams Config , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Prefix Ġ= ĠStreams Routing Configuration Constants . NODE _ ROUT ING _ KEY _ PREFIX ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrel Routing Ġ+= Ġfilter Map ( config Ġ= Ġstreams Config , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Prefix Ġ= ĠStreams Routing Configuration Constants . REL _ ROUT ING _ KEY _ PREFIX , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Suffix Ġ= ĠStreams Routing Configuration Constants . KEY _ STR ATE GY _ SUFFIX , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Ġ= Ġlog ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdefault Ġ= ĠStreams Event Router Configuration () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġdefault . copy ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġenabled Ġ= ĠStreams Config . is Source Enabled ( streams Config , Ġdb Name ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocedures Enabled Ġ= ĠStreams Config . has Pro cedures Enabled ( streams Config , Ġdb Name ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnode Routing Ġ= Ġif Ġ( node Routing . isEmpty ()) ĠlistOf ( Node Routing Configuration ( topic Ġ= Ġdb Name )) Ġelse Ġnode Routing , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrel Routing Ġ= Ġif Ġ( rel Routing . isEmpty ()) ĠlistOf ( Relationship Routing Configuration ( topic Ġ= Ġdb Name )) Ġelse Ġrel Routing , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschema Polling Interval Ġ= Ġstreams Config . get OrDefault ( Streams Routing Configuration Constants . SCHEMA _ POL LING _ INTERVAL , Ġdefault . schema Polling Interval ). toString (). to Long () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} Ċ } Ċ