package Ġio . datal br y . connector . sdk . consumer . generic Ċ Ċ import Ġcom . fasterxml . jackson . module . kotlin . jackson ObjectMapper Ċ import Ġio . datal br y . prec ise . api . schema . document . Document Ċ import Ġio . datal br y . connector . api . C rawl Processor Ċ import Ġio . datal br y . connector . api . Document Edge Ċ import Ġio . datal br y . connector . api . Document Node Ċ import Ġio . datal br y . connector . api . Node Ċ import Ġio . datal br y . connector . api . document . Document C rawl Processor Ċ import Ġio . datal br y . connector . api . document . Document Processor Ċ import Ġorg . slf 4 j . LoggerFactory Ċ Ċ /** Ċ Ġ* ĠThe Ġ[ Generic C rawl Processor ] Ġis Ġthe Ġmain Ġentry Ġpoint Ġfor Ġthe Ġhigh Ġlevel Ġlayer Ġof Ġthe ĠConnector ĠSDK . Ċ Ġ* ĠIt 's Ġbeing Ġused Ġfor Ġhandling Ġgeneric Ġ[ C rawl Processor ] Ġwithout Ġtheir Ġtype Ġinformation . Ċ Ġ* Ċ Ġ* ĠIntern ally Ġusing Ġreflection Ġon Ġeach Ġof Ġthose Ġ[ C rawl Processor ] s . ĠAll Ġ[ C rawl Processor ] Ġare Ġbeing Ġwrapped Ġin Ġa Ċ Ġ* Ġ[ Reflection Processor ] Ġto Ġprovide Ġthe Ġtype Ġinformation Ġat Ġruntime . Ċ Ġ* Ċ Ġ* Ġ Ġ- Ġ[ Reflection Processor ] Ġare Ġbeing Ġimplemented Ġby Ġ[ Reflection Processor Impl ] Ġwhich Ġis Ġusing Ġquiet Ċ Ġ* ĠĠĠ Ġextens ively Ġreflection . ĠAs Ġusually Ġreflection Ġis Ġslow Ġat Ġruntime , Ġso Ġmost Ġreflection Ġis Ġdone Ġat Ċ Ġ* ĠĠĠ Ġconstruction Ġtime Ġand Ġonly Ġthe Ġ" invoke " Ġcall Ġof Ġthe Ġ[ Processor . process ] Ġmethod Ġis Ġbeing Ġdone Ġat Ċ Ġ* ĠĠĠ Ġactual ĠRuntime Ġusing Ġreflection . Ċ Ġ* Ċ Ġ* Ġ@ param Ġcrawl Processors Ġto Ġwrap Ġwith Ġreflection Ġbased Ġcalls Ċ Ġ* Ċ Ġ* Ġ@ author Ġt imo Ġgr uen Ġ- Ġ 2 0 2 0 - 1 1 - 1 0 Ċ Ġ*/ Ċ class ĠGeneric C rawl Processor ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġcrawl Processors : ĠCollection < Document Processor < *, Ġ* >> Ċ ) ĊĠĠĠ Ġ: ĠDocument C rawl Processor Ċ { ĊĠĠĠ Ġprivate Ġval Ġinner Ġ: ĠCollection < Reflection Processor > Ġ= Ġcrawl Processors . map Ġ{ ĠReflection Processor Impl ( it ) Ġ} ĊĠĠĠ Ġprivate Ġval Ġmapper : ĠCollection < Item Mapper < Any >> Ġ= Ġinner . map Ġ{ ĠGeneric Item Mapper ( it . pro duces (). kotlin ) Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġj ack son Ġ= Ġj ack son ObjectMapper () ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġinner ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. groupBy Ġ{ Ġit . consume s () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ(_, Ġprocessors ) Ġ-> Ġprocessors . size Ġ> Ġ 1 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Multiple Ġprocessors Ġfor Ġ[ ${ it . key . simple Name }] Ġfound . Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" This Ġis Ġsupported , Ġbut Ġmight Ġresult Ġinto Ġless Ġstability ," Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Ġas Ġhuge Ġchunks Ġof Ġthe Ġsource Ġsystem Ġare Ġtouched Ġat Ġonce " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġprocess ( value : ĠDocument Edge ): ĠNode < Document Edge , ĠDocument > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpair Ġ= Ġinner ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġvalue . headers [ TYPE _ KEY ] Ġ== Ġit . consume s (). name Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġto Obj ( value . payload , Ġit . consume s ()) Ġto Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġ( obj , Ġprocessor ) Ġ-> Ġprocessor . process ( obj ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit Ġto Ġmapper . first { Ġm Ġ-> Ġm . supports ( it )} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . second . get Documents ( it . first ) Ġto Ġit . second . get Edges ( it . first ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. fold ( empty Set < Document >() Ġto Ġempty Set < Document Edge >()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġacc , Ġi Ġ-> Ġacc . first . union ( i . first ) Ġto Ġacc . second . union ( i . second ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn ĠDocument Node ( value . uuid , Ġpair . first , Ġpair . second ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġto Obj ( payload : ĠMap < String , ĠString >, Ġtype : ĠClass < *> ): ĠAny Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġj ack son . convert Value ( payload , Ġj ack son . type Factory . construct Type ( type )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Generic C rawl Processor :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġconst Ġval ĠTYPE _ KEY Ġ= Ġ" type " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠUUID _ KEY Ġ= Ġ" uuid " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠPRO DU CER _ KEY Ġ= Ġ" producer " ĊĠĠĠ Ġ} Ċ } Ċ