package Ġcom . ins id ion . ax on . open admin Ċ Ċ import Ġcom . ins id ion . ax on . open admin . events . Event T ailing Service Ċ import Ġcom . ins id ion . ax on . open admin . model . Token Information DTO Ċ import Ġcom . ins id ion . ax on . open admin . processors . Processor Status Service Ċ import Ġcom . ins id ion . ax on . open admin . tokens . Token Information Service Ċ import Ġorg . ax on framework . config . Event Processing Module Ċ import Ġorg . ax on framework . event handling . Tracking Event Processor Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . http . Response Entity Ċ import Ġorg . springframework . stereotype . Controller Ċ import Ġorg . springframework . web . bind . annotation . GetMapping Ċ import Ġorg . springframework . web . bind . annotation . PathVariable Ċ import Ġorg . springframework . web . bind . annotation . PostMapping Ċ import Ġorg . springframework . web . bind . annotation . RequestMapping Ċ import Ġorg . springframework . web . bind . annotation . RestController Ċ Ċ @ RestController Ċ @ RequestMapping ("\ ${ ax on . admin . base - url : ax on - admin }") Ċ class ĠAx on Open Admin Endpoint ( ĊĠĠĠ Ġprivate Ġval Ġax on Open Admin Token Store : ĠToken Information Service , ĊĠĠĠ Ġprivate Ġval Ġprocessor Status Service : ĠProcessor Status Service , ĊĠĠĠ Ġprivate Ġval Ġevent Processing Module : ĠEvent Processing Module , ĊĠĠĠ Ġprivate Ġval Ġevent T ailing Service : ĠEvent T ailing Service , Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( this :: class . java ) ĊĊĠĠĠ Ġ@ GetMapping ("/ tokens ") ĊĠĠĠ Ġfun Ġget Tokens (): ĠToken Information DTO Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġax on Open Admin Token Store . get Processors () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ GetMapping ("/ processors ") ĊĠĠĠ Ġfun Ġget Processors () Ġ= Ġprocessor Status Service . getStatus () ĊĊĠĠĠ Ġ@ GetMapping ("/ events ") ĊĠĠĠ Ġfun Ġget Events () Ġ= Ġevent T ailing Service . get Events () ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ split /{ segment Id }") ĊĠĠĠ Ġfun Ġsplit (@ PathVariable Ġprocessor Name : ĠString , Ġ@ PathVariable Ġsegment Id : ĠInt ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstatus Ġ= Ġit . processing Status ()[ segment Id ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreplay After Ġ= Ġstatus ?. is Replay ing Ġ== Ġtrue Ġ// ĠWhen Ġsplitting , Ġax on Ġre verts Ġto Ġa Ġnormal Ġtoken Ġfor Ġsome Ġreason ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! it . isRunning ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Will Ġnot Ġsplit Ġsegment Ġsince Ġit Ġis Ġnot Ġrunning Ġon Ġthis Ġnode ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ run On Processor With Response Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . split Segment ( segment Id ). get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( replay After ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . isRunning ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . sh ut Down () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . reset Tokens () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ merge /{ segment Id }") ĊĠĠĠ Ġfun Ġmerge (@ PathVariable Ġprocessor Name : ĠString , Ġ@ PathVariable Ġsegment Id : ĠInt ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Starting Ġmerge ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! it . isRunning ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Will Ġnot Ġmerge Ġsegment Ġsince Ġit Ġis Ġnot Ġrunning Ġon Ġthis Ġnode ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ run On Processor With Response Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" At tempting Ġmerge !") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstatus Ġ= Ġit . processing Status ()[ segment Id ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( status Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus . segment . merge able Segment Id () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . merge Segment ( segment Id ). get () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Merge Ġsuccessful ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Merge Ġfailed ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ run On Processor With Response Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ release /{ segment Id }") ĊĠĠĠ Ġfun Ġrelease (@ PathVariable Ġprocessor Name : ĠString , Ġ@ PathVariable Ġsegment Id : ĠInt ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Starting Ġrelease ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . processing Status ()[ segment Id ] Ġ?: Ġreturn @ run On Processor With Response Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . release Segment ( segment Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ stop ") ĊĠĠĠ Ġfun Ġstop (@ PathVariable Ġprocessor Name : ĠString ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . sh ut Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ start ") ĊĠĠĠ Ġfun Ġstart (@ PathVariable Ġprocessor Name : ĠString ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ processor /{ processor Name }/ reset ") ĊĠĠĠ Ġfun Ġreset (@ PathVariable Ġprocessor Name : ĠString ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġrun On Processor With Response ( processor Name ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . isRunning ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . sh ut Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . reset Tokens () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġrun On Processor With Response ( processor Name : ĠString , Ġblock : Ġ( Tracking Event Processor ) Ġ-> ĠBoolean ): ĠResponseEntity < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġevent Processor Ġ= Ġevent Processing Module . event Processor ( processor Name , ĠTracking Event Processor :: class . java ) ĊĠĠĠĠĠĠĠ Ġif (! event Processor . isPresent ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠResponseEntity . status ( 4 0 9 ). build () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġevent Processor ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map ( block ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse ( false ) ĊĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( result ) ĠResponseEntity . ok (). build () Ġelse ĠResponseEntity . status ( 5 0 0 ). build () ĊĠĠĠ Ġ} Ċ } Ċ