// ĠCopyright Ġ 2 0 2 1 ĠThe ĠCross - Media ĠMeasurement ĠAuthors Ċ // Ċ // Ġ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 Ġorg . wf an et . panel match . client . launcher Ċ Ċ import Ġcom . google . protobuf . ByteString Ċ import Ġjava . util . logging . Level Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . with Context Ċ import Ġorg . wf an et . measurement . api . v 2 alpha . Exchange Step Attempt Ċ import Ġorg . wf an et . measurement . api . v 2 alpha . Exchange Step Attempt Key Ċ import Ġorg . wf an et . measurement . api . v 2 alpha . Exchange Workflow . Step Ċ import Ġorg . wf an et . measurement . storage . Storage Client Ċ import Ġorg . wf an et . measurement . storage . Storage Client . Blob Ċ import Ġorg . wf an et . measurement . storage . create Blob Ċ import Ġorg . wf an et . panel match . client . common . Exchange Context Ċ import Ġorg . wf an et . panel match . client . ex chan get ask s . Custom IO Exchange Task Ċ import Ġorg . wf an et . panel match . client . ex chan get ask s . Exchange Task Ċ import Ġorg . wf an et . panel match . client . ex chan get ask s . Exchange Task Mapper Ċ import Ġorg . wf an et . panel match . client . launcher . Exchange Step Validator . Validated Exchange Step Ċ import Ġorg . wf an et . panel match . client . logger . Task Log Ċ import Ġorg . wf an et . panel match . client . logger . addTo Task Log Ċ import Ġorg . wf an et . panel match . client . logger . get And Clear Task Log Ċ import Ġorg . wf an et . panel match . client . storage . Private Storage Selector Ċ import Ġorg . wf an et . panel match . common . Timeout Ċ import Ġorg . wf an et . panel match . common . logger For Ċ import Ġorg . wf an et . panel match . common . storage . create Or Replace Blob Ċ Ċ private Ġconst Ġval ĠDONE _ TASK S _ PATH : ĠString Ġ= Ġ" done - tasks " Ċ Ċ /** Ċ Ġ* ĠExec utes Ġthe Ġwork Ġrequired Ġfor Ġ[ Validated Exchange Step ] s . Ċ Ġ* Ċ Ġ* ĠThis Ġinvolves Ġfinding Ġthe Ġappropriate Ġ[ Exchange Task ], Ġreading Ġthe Ġinputs , Ġexecuting Ġthe Ċ Ġ* Ġ[ Exchange Task ], Ġand Ġsaving Ġthe Ġoutputs . Ċ Ġ*/ Ċ class ĠExchange Task Executor ( ĊĠ Ġprivate Ġval ĠapiClient : ĠApi Client , ĊĠ Ġprivate Ġval Ġtimeout : ĠTimeout , ĊĠ Ġprivate Ġval Ġprivate Storage Selector : ĠPrivate Storage Selector , ĊĠ Ġprivate Ġval Ġexchange Task Mapper : ĠExchange Task Mapper Ċ ) Ġ: ĠExchange Step Executor Ġ{ ĊĊĠ Ġoverride Ġsuspend Ġfun Ġexecute ( ĊĠĠĠ Ġvalidated Step : ĠValid ated Exchange Step , ĊĠĠĠ Ġattempt Key : ĠExchange Step Attempt Key ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġval Ġname Ġ= Ġ"${ validated Step . step . step Id }@ ${ attempt Key . to Name () }" ĊĠĠĠ Ġwith Context ( Task Log ( name )) Ġ{ ĊĠĠĠĠĠ Ġval Ġcontext Ġ= ĊĠĠĠĠĠĠĠ ĠExchange Context ( attempt Key , Ġvalidated Step . date , Ġvalidated Step . workflow , Ġvalidated Step . step ) ĊĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġcontext . try Execute () ĊĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . addTo Task Log ( e , ĠLevel . SEVERE ) ĊĠĠĠĠĠĠĠ Ġmark As Finished ( attempt Key , ĠExchange Step Attempt . State . FAILED ) ĊĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġread Inputs ( step : ĠStep , Ġprivate Storage : ĠStorage Client ): ĠMap < String , ĠBlob > Ġ{ ĊĠĠĠ Ġreturn Ġstep . input Labels Map . map Values Ġ{ Ġ( label , Ġblob Key ) Ġ-> ĊĠĠĠĠĠ Ġrequire NotNull ( private Storage . get Blob ( blob Key )) Ġ{ ĊĠĠĠĠĠĠĠ Ġ" Missing Ġblob Ġkey Ġ'$ blob Key ' Ġfor Ġinput Ġlabel Ġ'$ label '" ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġwrite Outputs ( ĊĠĠĠ Ġstep : ĠStep , ĊĠĠĠ Ġtask Output : ĠMap < String , ĠFlow < ByteString >>, ĊĠĠĠ Ġprivate Storage : ĠStorage Client ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġfor Ġ(( generic Label , Ġflow ) Ġin Ġtask Output ) Ġ{ ĊĠĠĠĠĠ Ġval Ġblob Key Ġ= ĊĠĠĠĠĠĠĠ Ġrequire NotNull ( step . output Labels Map [ generic Label ]) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ" Missing Ġ$ generic Label Ġin Ġoutput Labels Ġfor Ġstep : Ġ$ step " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġprivate Storage . create Or Replace Blob ( blob Key , Ġflow ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġmark As Finished ( ĊĠĠĠ Ġattempt Key : ĠExchange Step Attempt Key , ĊĠĠĠ Ġstate : ĠExchange Step Attempt . State ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġlogger . addTo Task Log (" Mark ing Ġattempt Ġstate : Ġ$ state ") ĊĠĠĠ ĠapiClient . finish Exchange Step Attempt ( attempt Key , Ġstate , Ġget And Clear Task Log ()) ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun ĠExchange Context . try Execute () Ġ{ ĊĠĠĠ Ġlogger . addTo Task Log (" Executing Ġ${ step . step Id } Ġwith Ġattempt Ġ$ attempt Key Ġfor Ġ$ this ") ĊĠĠĠ Ġval Ġprivate Storage Client : ĠStorage Client Ġ= ĊĠĠĠĠĠ Ġprivate Storage Selector . get Storage Client ( exchange Date Key ) ĊĠĠĠ Ġif Ġ(! is Already Complete ( step , Ġprivate Storage Client )) Ġ{ ĊĠĠĠĠĠ Ġrun Step ( private Storage Client ) ĊĠĠĠĠĠ Ġwrite Done Blob ( step , Ġprivate Storage Client ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġ// ĠThe ĠKing dom Ġwill Ġbe Ġable Ġto Ġdetect Ġif Ġit 's Ġhand ing Ġout Ġduplicate Ġtasks Ġbecause Ġit Ġwill ĊĠĠĠ Ġ// Ġattempt Ġto Ġtransition Ġan ĠExchange Step Ġfrom Ġsome Ġterminal Ġstate Ġto Ġ` SUCCEEDED `. ĊĠĠĠ Ġmark As Finished ( attempt Key , ĠExchange Step Attempt . State . SUCCEEDED ) ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun ĠExchange Context . run Step ( private Storage : ĠStorage Client ) Ġ{ ĊĠĠĠ Ġtimeout . run WithTimeout Ġ{ ĊĠĠĠĠĠ Ġval Ġexchange Task : ĠExchange Task Ġ= Ġexchange Task Mapper . get Exchange Task For Step ( this @ run Step ) ĊĠĠĠĠĠ Ġval Ġtask Input : ĠMap < String , ĠBlob > Ġ= ĊĠĠĠĠĠĠĠ Ġwhen Ġ( exchange Task ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġis ĠCustom IO Exchange Task Ġ-> Ġempty Map () ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġread Inputs ( step , Ġprivate Storage ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġval Ġtask Output : ĠMap < String , ĠFlow < ByteString >> Ġ= Ġexchange Task . execute ( task Input ) ĊĠĠĠĠĠ Ġwrite Outputs ( step , Ġtask Output , Ġprivate Storage ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġis Already Complete ( step : ĠStep , Ġprivate Storage : ĠStorage Client ): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġprivate Storage . get Blob ("$ DONE _ TASK S _ PATH /${ step . step Id }") Ġ!= Ġnull ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġwrite Done Blob ( step : ĠStep , Ġprivate Storage : ĠStorage Client ) Ġ{ ĊĠĠĠ Ġ// ĠTODO : Ġwrite Ġthe Ġstate Ġinto Ġthe Ġblob . ĊĠĠĠ Ġ// ĠĠ ĠThis Ġwill Ġprevent Ġre - execution Ġof Ġtasks Ġthat Ġfailed . ĊĠĠĠ Ġprivate Storage . create Blob ("$ DONE _ TASK S _ PATH /${ step . step Id }", ĠByte String . EMPTY ) ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger Ġby Ġlogger For () ĊĠ Ġ} Ċ } Ċ