/* Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 . ĠJames ĠAdam Ġand Ġthe ĠOpen ĠData ĠManagement ĠPlatform Ġcontributors . Ċ Ġ* Ċ Ġ* Ġ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 . opend mp . processor . run Ċ Ċ import Ġcom . fasterxml . jackson . module . kotlin . jackson ObjectMapper Ċ import Ġio . opend mp . common . exception . NotImplemented Exception Ċ import Ġio . opend mp . common . exception . Processor Definition Exception Ċ import Ġio . opend mp . common . exception . Run Plan Logic Exception Ċ import Ġio . opend mp . common . exception . Unsupported Processor Type Exception Ċ import Ġio . opend mp . common . model . Processor Run Model Ċ import Ġio . opend mp . common . model . Processor Type Ċ import Ġio . opend mp . common . model . Source Model Ċ import Ġio . opend mp . common . model . Source Type Ċ import Ġio . opend mp . common . model . properties . Script Language Ċ import Ġio . opend mp . common . util . Message Util Ċ import Ġio . opend mp . processor . config . Spring Context Ċ import Ġio . opend mp . processor . domain . Run Plan Ċ import Ġio . opend mp . processor . run . processors .* Ċ import Ġorg . apache . camel . Exchange Ċ import Ġorg . apache . camel . Logging Level Ċ import Ġorg . apache . camel . builder . Default ErrorHandler Builder Ċ import Ġorg . apache . camel . builder . Route Builder Ċ import Ġorg . apache . camel . component . aws . s 3 . S 3 Constants Ċ import Ġorg . apache . camel . http . base . Http Operation FailedException Ċ import Ġorg . apache . camel . model .* Ċ import Ġorg . apache . camel . spi . Ide mpot ent Repository Ċ import Ġorg . apache . http . client . utils . URL Encoded Utils Ċ import Ġorg . apache . http . message . Basic NameValuePair Ċ import Ġjava . util .* Ċ Ċ /** Ċ Ġ* ĠRun Plan Route Builder Ġ- Ġtakes Ġa ĠRun ĠPlan Ġand Ġbuilds Ġcamel Ġroutes Ġto Ġset Ġup Ġa Ġdata Ġpipeline Ċ Ġ* ĠThis Ġis Ġkind Ġof Ġthe Ġheart Ġof Ġthe Ġwhole Ġsystem Ġright Ġhere Ċ Ġ*/ Ċ class ĠRun Plan Route Builder ( private Ġval Ġrun Plan : ĠRun Plan , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġnum Retries : ĠInt Ġ= Ġ 2 ): ĠRoute Builder () Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġrepo Ġ= ĠSpring Context . context . getBean (" ide mpot ent Repo ", ĠIde mpot ent Repository :: class . java ) ĊĊĠĠĠ Ġprivate Ġfun Ġgenerate Ingest Endpoint ( source : ĠSource Model ) Ġ: ĠProcessor Definition < *> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen ( source . source Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSource Type . ING EST _ FILE _ DROP Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom (" file ://${ source . source Location !! }? read Lock = changed & bridge ErrorHandler = true ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. log (" Ingest ing Ġfrom Ġfile Ġ\ ${ header . Camel FileName }") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSource Type . ING EST _ FTP Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom (" ftp ://${ source . source Location !! }? binary = true & bridge ErrorHandler = true ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. log (" Ingest ing Ġfrom Ġftp Ġhost Ġ\ ${ header . Camel File Host } Ġfile : Ġ\ ${ header . Camel FileName }") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSource Type . ING EST _ S 3 Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbucket Ġ= Ġsource . additional Properties ?. get (" bucket ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠProcessor Definition Exception (" Bucket Ġmust Ġbe Ġspecified Ġfor ĠS 3 Ġingest ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Prefix Ġ= Ġsource . source Location ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠProcessor Definition Exception (" No ĠS 3 Ġkey Ġprefix Ġprovided !") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom (" aws - s 3 :// $ bucket ? prefix =$ key Prefix & bridge ErrorHandler = true & delete After Read = false ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader ( S 3 Constants . CONTENT _ TYPE , Ġconstant (" application / octet - stream ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader ( S 3 Constants . BUCKET _ NAME , Ġconstant ( bucket )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. ide mpot ent Consumer ( header ( S 3 Constants . KEY ), repo ). completion E ager ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. log (" Ingest ing Ġfrom ĠS 3 : Ġ\ ${ header . Camel AW SS 3 Key }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠNotImplementedException (" Source Type Ġ${ source . source Type } Ġnot Ġsupported ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġconfigure () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Set Ġup Ġerror Ġhandling ĊĠĠĠĠĠĠĠ ĠerrorHandler ( dead Letter Channel (" direct :${ run Plan . id }- dead ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. retry At tempted LogLevel ( Logging Level . WARN ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. allow Re delivery While Stopping ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. maximum Re deliver ies ( num Retries ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. re delivery Delay ( 1 0 0 0 L )) ĊĊĠĠĠĠĠĠĠ Ġfrom (" direct :${ run Plan . id }- dead ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader (" run Plan ", Ġconstant ( run Plan . id )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. bean ( Failure Handler (), Ġ" process Failure ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to (" log : io . opend mp . processor . run ? level = ERROR ") ĊĊĠĠĠĠĠĠĠ Ġ// Build Ġthe Ġroutes ĊĠĠĠĠĠĠĠ Ġrun Plan . starting Processors . forEach Ġ{ Ġsp id Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsp Ġ= Ġrun Plan . processors [ sp id ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġerror (" Starting Ġprocessor Ġmissing Ġfrom Ġprocessor Ġmap ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstart Route ( sp ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget ErrorHandler ForType ( ptype : ĠProcessor Type ) Ġ: ĠDefault ErrorHandler Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdead Letter Uri Ġ= Ġ" direct :${ run Plan . id }- dead " ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen ( ptype ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . COLLECT Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdead Letter Channel ( dead Letter Uri ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. maximum Re deliver ies ( 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. allow Re delivery While Stopping ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. re delivery Delay ( 1 5 0 0 L ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdead Letter Channel ( dead Letter Uri ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. maximum Re deliver ies ( 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġstart Route Ġ- Ġstarts Ġa Ġchain Ġof Ġprocessors ĊĠĠĠĠ Ġ* ĠStarting Ġprocessors Ġare Ġnot Ġdependent Ġon Ġother Ġprocessors Ġand Ġcan Ġstart Ġrunning Ġright Ġaway ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġstart Route ( sp : ĠProcessor Run Model ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠHere Ġwe Ġassume Ġ( and Ġwill Ġenforce ) Ġthat Ġa Ġstarting Ġprocessor Ġhas Ġonly Ġone Ġinput ĊĠĠĠĠĠĠĠ Ġ// ĠCamel Ġdoesn 't Ġreally Ġsupport Ġit Ġand ĠI Ġcan 't Ġimagine Ġa Ġuse Ġcase Ġfor Ġit . ĊĊĠĠĠĠĠĠĠ Ġval Ġsource Ġ= Ġsp . inputs . first () ĊĠĠĠĠĠĠĠ Ġval Ġsource Ep Ġ= Ġwhen ( sp . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . ING EST Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgenerate Ingest Endpoint ( source ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠUnsupported Processor Type Exception (" The Ġprocessor Ġtype Ġ${ sp . type } Ġis Ġnot Ġsupported Ġas Ġa Ġstarting Ġprocessor ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġdeps : ĠList < Processor Run Model ?> Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Plan . processor Dependency Map [ sp . id ]?. map Ġ{ Ġrun Plan . processors [ it ] Ġ} Ġ?: ĠlistOf () ĊĠĠĠĠĠĠĠ Ġval Ġroute Id Ġ= Ġ"${ run Plan . id }-${ sp . id }" ĊĠĠĠĠĠĠĠ Ġsource Ep ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader (" processor ", Ġconstant ( sp . id )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. convert Body To ( ByteArray :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSet Ġa Ġroute Id Ġto Ġmake Ġfinding Ġthis Ġroute Ġin Ġthe ĠCamel ĠContext Ġeasier Ġlater ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. route Id ( route Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. startup Order ( Utils . getNext Startup Order ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. process ( Data Wrapper ( sp )) ĊĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeps . size Ġ== Ġ 1 Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdest Ġ= Ġ" direct :${ run Plan . id }-${ deps . first () !!. id }" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource Ep . to ( dest ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeps . size Ġ> Ġ 1 Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġwe Ġhave Ġmore Ġthan Ġ 1 Ġprocessor Ġexpecting Ġoutput Ġfrom Ġthis Ġprocessor , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdo Ġa Ġmulticast ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdest Ġ= Ġdeps . map Ġ{ Ġd Ġ-> Ġ" sed a :${ run Plan . id }-${ d !!. id } "} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource Ep ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. mult icast (). on Prepare ( Multi Prepare Processor ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. parallel Processing () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to (* dest . to Typed Array ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRun Plan Logic Exception (" Starting Ġprocessor Ġhas Ġno Ġoutputs ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠContinue Ġbuilding Ġwith Ġthe Ġdependencies ĊĠĠĠĠĠĠĠ Ġdeps . forEach Ġ{ Ġthis . continue Route ( it !! , Ġdeps . size Ġ> Ġ 1 ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġcontinue Route Ġ- Ġfor Ġthe Ġprocessors Ġfrom Ġthe Ġmiddle Ġof Ġa Ġchain Ġto Ġthe Ġend ĊĠĠĠĠ Ġ* Ġ@ param Ġcur Proc ĠThe Ġprocessor Ġto Ġexecute ĊĠĠĠĠ Ġ* Ġ@ param Ġmulti Ġ- ĠIf Ġthe Ġsource Ġis Ġfrom Ġa Ġmulticast Ġor Ġsingle Ġ( direct ) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġcontinue Route ( cur Proc : ĠProcessor Run Model , Ġmulti : ĠBoolean Ġ= Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsource Ep Ġ= Ġif ( multi ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" sed a :${ run Plan . id }-${ cur Proc . id }" ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" direct :${ run Plan . id }-${ cur Proc . id }" ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġdeps : ĠList < Processor Run Model ?> Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Plan . processor Dependency Map [ cur Proc . id ]?. map Ġ{ Ġrun Plan . processors [ it ] Ġ} Ġ?: ĠlistOf () ĊĠĠĠĠĠĠĠ Ġval Ġproc Ġ= Ġwhen ( cur Proc . type ){ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . SCRIPT Ġ-> ĠScript Processor ( cur Proc ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . EXTERNAL Ġ-> ĠExternal Processor ( cur Proc ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . COLLECT Ġ-> ĠCollect Processor ( cur Proc ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProcessor Type . PLUGIN Ġ-> ĠPlugin Processor ( cur Proc ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠUnsupported Processor Type Exception (" The Ġprocessor Ġtype Ġ${ cur Proc . type } Ġis Ġnot Ġsupported ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġroute Id Ġ= Ġ"${ run Plan . id }-${ cur Proc . id }" ĊĊĠĠĠĠĠĠĠ Ġval Ġcont Route Ġ= Ġfrom ( source Ep ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. route Id ( route Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. error Handler ( get ErrorHandler ForType ( cur Proc . type )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. startup Order ( Utils . getNext Startup Order ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader (" processor ", Ġconstant ( cur Proc . id )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠMake Ġa Ġservice Ġcall , Ġif Ġapplicable ĊĠĠĠĠĠĠĠ Ġservice Call ( cont Route , Ġcur Proc ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠExecute Ġthis Ġprocessor ĊĠĠĠĠĠĠĠ Ġcont Route . process ( proc ). id ( cur Proc . id ) ĊĊĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeps . size Ġ== Ġ 1 Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcont Route ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to (" direct :${ run Plan . id }-${ deps . first () !!. id }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeps . size Ġ> Ġ 1 Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcont Route . mult icast (). on Prepare ( Multi Prepare Processor ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. parallel Processing () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to (*( deps . map { Ġd Ġ-> Ġ" sed a :${ run Plan . id }-${ d !!. id } "} . to Typed Array ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcont Route . process ( Completion Processor ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. id ("${ run Plan . id }-${ cur Proc . id }- complete "). end () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġdeps . forEach Ġ{ Ġcontinue Route ( it !! , Ġdeps . size Ġ> Ġ 1 ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget QueryParams ( proc : ĠProcessor Run Model ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprop Str Ġ= ĠMessage Util . mapper . writeValue AsString ( proc . properties ) ĊĠĠĠĠĠĠĠ Ġval Ġb 6 4 Props Ġ= ĠBase 6 4 . getUrl Encoder (). encode ToString ( prop Str . toByteArray ()) ĊĠĠĠĠĠĠĠ Ġreturn Ġ" properties =$ b 6 4 Props " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠMake Ġa Ġservice Ġcall ! ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġservice Call ( route : ĠRoute Definition , Ġproc : ĠProcessor Run Model ) Ġ Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprops Ġ= Ġproc . properties !! ĊĠĠĠĠĠĠĠ Ġ// Find Ġa Ġservice Ġto Ġcall ĊĠĠĠĠĠĠĠ Ġval Ġservice Ġ= Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Do Ġwe Ġhave Ġa Ġservice Ġin Ġthe Ġheader Ġ( plugins )? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprops . containsKey (" serviceName ") Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprops [" serviceName "]. toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Python Ġscript Ġprocessor ? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproc . type Ġ== ĠProcessor Type . SCRIPT Ġ&& ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠScript Language . valueOf ( props [" language "]. toString ()) Ġ== ĠScript Language . PYTHON Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" python - script - processor " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproc . type Ġ== ĠProcessor Type . SCRIPT Ġ&& ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠScript Language . valueOf ( props [" language "]. toString ()) Ġ== ĠScript Language . CLO J URE Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" clojure - script - processor " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġreturn Ġ// Don 't Ġservice Ġcall ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġroute ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. log (" M aking Ġcall Ġto Ġ$ service ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. circuit Breaker (). inherit ErrorHandler ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. service Call () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader ( Exchange . HTTP _ METHOD , Ġconstant (" POST ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. service Call Configuration (" basic Service Call ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. name ( service ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. uri (" http :// $ service / process ? throw Exception On Failure = false & ${ get QueryParams ( proc )} ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. end () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. end Circuit Breaker () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. log (" completed Ġcall Ġto Ġ$ service ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. remove Headers (" Camel Service Call * ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. process ( Service Call Response Processor ()) ĊĠĠĠ Ġ} Ċ Ċ }