/* Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 . Ġ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 . messaging Ċ Ċ import Ġcom . fasterxml . jackson . core . Json Processing Exception Ċ import Ġcom . fasterxml . jackson . module . kotlin . jackson ObjectMapper Ċ import Ġio . opend mp . common . message . Collection Complete Message Ċ import Ġio . opend mp . common . message . Run Plan Failure Message Ċ import Ġio . opend mp . common . message . Run Plan Start Failure Message Ċ import Ġorg . apache . camel . Producer Template Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . beans . factory . annotation . Autowired Ċ import Ġorg . springframework . beans . factory . annotation . Value Ċ import Ġorg . springframework . context . annotation . Profile Ċ import Ġorg . springframework . stereotype . Component Ċ Ċ @ Profile (" ! test ") Ċ @ Component Ċ class ĠRun Plan Status Dispatcher Ġ@ Autowired Ġconstructor ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġproducer Template : ĠProducer Template Ċ ){ ĊĊĠĠĠ Ġ@ Value ("\ ${ od mp . pulsar . namespace }") ĊĠĠĠ Ġlateinit Ġvar Ġpul sar Namespace : ĠString ĊĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( java Class ) ĊĊĠĠĠ Ġprivate Ġval Ġmapper Ġ= Ġj ack son ObjectMapper () ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġmapper . find And Register Modules () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcollect EndPoint (): ĠString Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" pulsar : persistent :// $ pulsar Namespace / run plan _ collect _ status ? producer Name = od mp Producer " ĊĊĠĠĠ Ġfun Ġfailure EndPoint (): ĠString Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" pulsar : persistent :// $ pulsar Namespace / run plan _ failure ? producer Name = od mp Producer " ĊĠĠĠ Ġfun Ġstart Failure EndPoint Url () Ġ: ĠString Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" pulsar : persistent :// $ pulsar Namespace / run plan _ start _ failure " ĊĊĠĠĠ Ġprivate Ġfun ĠsendMessage ( msg : ĠAny , Ġendpoint : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjson Data Ġ= Ġmapper . writeValue AsString ( msg ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Template . send Body ( endpoint , Ġjson Data ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( j pe : ĠJson Processing Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Error Ġconverting Ġmessage ", Ġj pe ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Error Ġsending Ġmessage ", Ġex ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsend Start Run Plan Failure Message ( msg : ĠRun Plan Start Failure Message ) Ġ{ ĊĠĠĠĠĠĠĠ ĠsendMessage ( msg , Ġstart Failure EndPoint Url ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsend Collection Complete ( msg : ĠCollection Complete Message ) Ġ{ ĊĠĠĠĠĠĠĠ ĠsendMessage ( msg , Ġcollect EndPoint ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsend Failure Message ( msg : ĠRun Plan Failure Message ) Ġ{ ĊĠĠĠĠĠĠĠ ĠsendMessage ( msg , Ġfailure EndPoint ()) ĊĠĠĠ Ġ} Ċ Ċ }