package Ġid h . c 3 cloud . tis . pilot Ċ Ċ import Ġid h . c 3 cloud . tis .* Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . new Single Thread Context Ċ import Ġkotlinx . coroutines . reactive . await Single Ċ import Ġorg . bson . types . ObjectId Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . http . HttpStatus Ċ import Ġorg . springframework . http . Response Entity Ċ import Ġorg . springframework . web . bind . annotation .* Ċ import Ġreactor . core . publisher . Mono Ċ import Ġjava . lang . Exception Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ Ċ @ RestController Ċ @ RequestMapping ("/ api / c 3 cloud / import ") Ċ class ĠC 3 dp Controller ( private Ġval Ġtask Executor : ĠTask Executor , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġtask Repository : ĠTask Repository , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġtask Execution Repository : ĠTask Execution Repository , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġpatient Repository : ĠPatient Repository , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġc 3 dp : ĠC 3 dp ): ĠCor outine Scope Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( C 3 dp Controller :: class . java ) ĊĊĠĠĠ Ġoverride Ġval Ġcoroutine Context Ġ= Ġnew Single Thread Context (" C 3 DP - Notification ") ĊĊĠĠĠ Ġprivate Ġval Ġnot ifier Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnotified Ġ= Ġmutable Set Of < ObjectId >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( job Ġin Ġjobs ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstatus Ġ= Ġmutable Map Of < String , ĠAny >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexecution Ġ= Ġtask Execution Repository . findById ( job ). await Single () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( execution . state Ġ== ĠTask Execution . State . Success ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" success "] Ġ= Ġ" Data Ġsynchronization Ġcomplete " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġif Ġ( execution . state Ġ== ĠTask Execution . State . Error ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" error "] Ġ= Ġexecution . details [" error "] Ġ?: Ġ" None " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" message "] Ġ= Ġexecution . details [" message "] Ġ?: Ġ" None " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" timestamp "] Ġ= Ġexecution . endTime !! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ( e Ġ: ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Getting ĠTask Execution Ġ${ job } Ġfor ĠC 3 DP Ġnotification ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" error "] Ġ= Ġe . toString () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus [" message "] Ġ= Ġ" Task Execution Ġdatabase Ġerror " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// c 3 dp . notify Test ( job , Ġstatus ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġc 3 dp . notify ( job , Ġstatus ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcatch ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Notify ing ĠC 3 DP Ġof Ġjob Ġ${ job } Ġcompletion ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnotified . add ( job ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjobs . removeAll ( not ified ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġjobs Ġ= ĠConcurrent HashMap . new Key Set < ObjectId >() ĊĊĠĠĠ Ġprivate Ġval Ġtasks Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" provider 1 " Ġto Ġ" Provider 1 - CD A - DB P - Import - v 1 ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" provider 2 " Ġto Ġ" Provider 2 - Open Services - Import - v 1 ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" provider 3 " Ġto Ġ" Provider 3 - Import - v 1 ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" helloworld " Ġto Ġ" HelloWorld - v 1 " ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ@ GetMapping ("/ { site }/{ patient }") ĊĠĠĠ Ġfun Ġimport Single Patient (@ PathVariable Ġsite : ĠString , Ġ@ PathVariable Ġpatient : ĠString ): ĠMono < Response Entity < Map < String , ĠString >>> Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠtaskId Ġ= Ġtasks [ site . toLowerCase ()] ĊĠĠĠĠĠĠĠ Ġif Ġ( taskId Ġ== Ġnull ) Ġreturn ĠMono . just ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . bad Request (). body ( map Of (" error " Ġto Ġ"' ${ site }' Ġnot Ġfound ")) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġreturn Ġpatient Repository ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ById ( patient ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit Ġ== Ġfalse Ġ} ĠĠĠ Ġ// Ġpatient Ġnot Ġfound ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . bad Request (). body ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of (" error " Ġto Ġ" Patient ĠID Ġ${ patient } Ġnot Ġfound ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. switch If Empty ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Repository . findById ( taskId ) ĠĠĠĠ Ġ// Ġpatient Ġfound ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġtask Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtask Parameters Ġ= Ġtask . parameters . to Mutable Map () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Parameters [" patient "] Ġ= Ġpatient ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Id Ġ= Ġtask Executor . execute ( taskId , Ġtask Parameters ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . accepted (). body ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" success " Ġto Ġ" Data Ġsynchronization Ġtriggered ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" job Ġid " Ġto Ġ"${ jobId }" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ GetMapping ("/ { site }") ĊĠĠĠ Ġfun Ġimport Multiple Pat ients (@ PathVariable Ġsite : ĠString , Ġ@ RequestParam Ġpatient : ĠString ): ĠMono < Response Entity < Map < String , ĠString >>> Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠtaskId Ġ= Ġtasks [ site . toLowerCase ()] ĊĠĠĠĠĠĠĠ Ġif Ġ( taskId Ġ== Ġnull ) Ġreturn ĠMono . just ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . bad Request (). body ( map Of (" error " Ġto Ġ"' ${ site }' Ġnot Ġfound ")) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġval Ġpatient Ids Ġ= Ġpatient . split ("," ). map ( String :: trim ) ĊĠĠĠĠĠĠĠ Ġreturn Ġpatient Repository ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. findAll ById ( patient Ids ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map ( Patient :: id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. collect List () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġfound Ġ-> Ġpatient Ids Ġ- Ġfound Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . isNotEmpty () Ġ} ĠĠĠĠ Ġ// Ġsome Ġpatients Ġnot Ġfound ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . bad Request (). body ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of (" error " Ġto Ġ" Patient ĠID Ġ${ it . join ToString ()} Ġnot Ġfound ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. switch If Empty ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Repository . findById ( taskId ) ĠĠĠĠ Ġ// Ġall Ġpatients Ġfound ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġtask Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtask Parameters Ġ= Ġtask . parameters . to Mutable Map () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Parameters [" patient "] Ġ= Ġpatient Ids . join ToString () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Id Ġ= Ġtask Executor . execute ( taskId , Ġtask Parameters ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠResponseEntity . accepted (). body ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" success " Ġto Ġ" Data Ġsynchronization Ġtriggered ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" job Ġid " Ġto Ġ"${ jobId }" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ GetMapping ("/ job / subscription /{ id }") ĊĠĠĠ Ġ@ ResponseStatus ( HttpStatus . ACCEPT ED ) ĊĠĠĠ Ġfun Ġsubscribe Job Completion Notification (@ PathVariable Ġid : ĠObjectId ) Ġ{ ĊĠĠĠĠĠĠĠ Ġjobs . add ( id ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ PostMapping ("/ job / event / api ") ĊĠĠĠ Ġ@ ResponseStatus ( HttpStatus . OK ) ĊĠĠĠ Ġfun Ġon Job Completion Notification (@ RequestBody Ġbody : ĠMap < String , ĠString >) Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln ( body ) ĊĠĠĠ Ġ} Ċ Ċ }