// Ġ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 . measurement . king dom . deploy . g cloud . spanner . writ ers Ċ Ċ import Ġcom . google . cloud . spanner . Statement Ċ import Ġcom . google . cloud . spanner . Struct Ċ import Ġcom . google . common . base . Optional Ċ import Ġcom . google . type . Date Ċ import Ġjava . time . Clock Ċ import Ġjava . time . Duration Ċ import Ġkotlinx . coroutines . flow . single Ċ import Ġkotlinx . coroutines . flow . single OrNull Ċ import Ġorg . wf an et . measurement . common . to Proto Time Ċ import Ġorg . wf an et . measurement . g cloud . common . to Cloud Date Ċ import Ġorg . wf an et . measurement . g cloud . common . to G cloud Timestamp Ċ import Ġorg . wf an et . measurement . g cloud . spanner . buffer Insert Mutation Ċ import Ġorg . wf an et . measurement . g cloud . spanner . set Ċ import Ġorg . wf an et . measurement . g cloud . spanner . set Json Ċ import Ġorg . wf an et . measurement . g cloud . spanner . statement Ċ import Ġorg . wf an et . measurement . internal . common . Provider Ċ import Ġorg . wf an et . measurement . internal . king dom . Exchange Step Ċ import Ġorg . wf an et . measurement . internal . king dom . Exchange Step Attempt Ċ import Ġorg . wf an et . measurement . internal . king dom . Stream Exchange Steps Request Kt . filter Ċ import Ġorg . wf an et . measurement . internal . king dom . copy Ċ import Ġorg . wf an et . measurement . internal . king dom . exchange Step Attempt Details Ċ import Ġorg . wf an et . measurement . king dom . deploy . g cloud . spanner . queries . Stream Exchange Steps Ċ import Ġorg . wf an et . measurement . king dom . deploy . g cloud . spanner . writ ers . Claim Ready Exchange Step . Result Ċ Ċ private Ġval ĠDEFAULT _ EXPI RATION _ DURATION : ĠDuration Ġ= ĠDuration . of Days ( 1 ) Ċ Ċ class ĠClaim Ready Exchange Step ( ĊĠ Ġprivate Ġval Ġprovider : ĠProvider , ĊĠ Ġprivate Ġval Ġclock : ĠClock , Ċ ) Ġ: ĠSp anner Writer < Optional < Result >, ĠOptional < Result >>() Ġ{ ĊĠ Ġdata Ġclass ĠResult ( val Ġstep : ĠExchange Step , Ġval Ġattempt Index : ĠInt ) ĊĊĠ Ġoverride Ġsuspend Ġfun ĠTransaction Scope . run Transaction (): ĠOptional < Result > Ġ{ ĊĠĠĠ Ġ// ĠGet Ġthe Ġfirst ĠExchange Step Ġwith Ġstatus : ĠREAD Y Ġ| ĠREAD Y _ FOR _ RETRY Ġ Ġby Ġgiven ĠProvider Ġid . ĊĠĠĠ Ġval Ġexchange Step Result Ġ= ĊĠĠĠĠĠ ĠStream Exchange Steps ( ĊĠĠĠĠĠĠĠĠĠ Ġrequest Filter Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfilter Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprincipal Ġ= Ġprovider ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstep Provider Ġ= Ġprovider ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstates Ġ+= ĠExchange Step . State . READY _ FOR _ RETRY ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstates Ġ+= ĠExchange Step . State . READY ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠ Ġlimit Ġ= Ġ 1 ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ. execute ( transaction Context ) ĊĠĠĠĠĠĠĠ Ġ. single OrNull () ĊĠĠĠĠĠĠĠ Ġ?: Ġreturn ĠOptional . abs ent () ĊĊĠĠĠ Ġval Ġexchange Step Ġ= Ġexchange Step Result . exchange Step ĊĠĠĠ Ġval Ġrecur ring Exchange Id Ġ= Ġexchange Step Result . recurring Exchange Id ĊĊĠĠĠ Ġ// ĠCreate Ġan ĠExchange ĠStep ĠAttempt Ġfor Ġthis ĠStep . ĊĠĠĠ Ġval Ġattempt Index Ġ= ĊĠĠĠĠĠ Ġcreate Exchange Step Attempt ( ĊĠĠĠĠĠĠĠ Ġrecur ring Exchange Id Ġ= Ġrecur ring Exchange Id , ĊĠĠĠĠĠĠĠ Ġdate Ġ= Ġexchange Step . date , ĊĠĠĠĠĠĠĠ Ġstep Index Ġ= Ġexchange Step . step Index . to Long () ĊĠĠĠĠĠ Ġ) ĊĊĠĠĠ Ġ// ĠFinally , Ġupdate Ġthe ĠExchange ĠStep Ġstatus Ġto ĠIN _ PROGRESS . ĊĠĠĠ Ġupdate Exchange Step State ( ĊĠĠĠĠĠ Ġexchange Step Ġ= Ġexchange Step , ĊĠĠĠĠĠ Ġrecur ring Exchange Id Ġ= Ġexchange Step Result . recurring Exchange Id , ĊĠĠĠĠĠ Ġstate Ġ= ĠExchange Step . State . IN _ PROGRESS ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġval Ġupdated Step Ġ= Ġexchange Step . copy Ġ{ Ġstate Ġ= ĠExchange Step . State . IN _ PROGRESS Ġ} ĊĊĠĠĠ Ġreturn ĠOptional . of ( Result ( updated Step , Ġattempt Index . toInt ())) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun ĠResult Scope < Optional < Result >> . build Result (): ĠOptional < Result > Ġ{ ĊĠĠĠ Ġrequire NotNull ( transaction Result ) ĊĠĠĠ Ġif Ġ(! transaction Result . isPresent ) Ġ{ ĊĠĠĠĠĠ Ġreturn ĠOptional . abs ent () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġexchange Step With UpdateTime Ġ= ĊĠĠĠĠĠ Ġtransaction Result . get (). step . copy Ġ{ Ġupdate Time Ġ= Ġcommit Timestamp . to Proto () Ġ} ĊĊĠĠĠ Ġreturn ĠOptional . of ( Result ( exchange Step With UpdateTime , Ġtransaction Result . get (). attempt Index )) ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun ĠTransaction Scope . create Exchange Step Attempt ( ĊĠĠĠ Ġrecur ring Exchange Id : ĠLong , ĊĠĠĠ Ġdate : ĠDate , ĊĠĠĠ Ġstep Index : ĠLong ĊĠ Ġ): ĠLong Ġ{ ĊĠĠĠ Ġval Ġnow Ġ= Ġclock . instant () ĊĊĠĠĠ Ġval Ġdetails Ġ= Ġexchange Step Attempt Details Ġ{ ĊĠĠĠĠĠ ĠstartTime Ġ= Ġnow . to Proto Time () ĊĠĠĠĠĠ Ġ// ĠTODO (@ yun y eng ): ĠSet ĠExchange Step Attempt Details Ġwith Ġsuitable Ġfields . ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġattempt Index Ġ= Ġfind Attempt Index ( recurring Exchange Id , Ġdate , Ġstep Index ) ĊĊĠĠĠ Ġtransaction Context . buffer Insert Mutation (" Exchange Step Attempts ") Ġ{ ĊĠĠĠĠĠ Ġset (" Recurring Exchange Id " Ġto Ġrecur ring Exchange Id ) ĊĠĠĠĠĠ Ġset (" Date " Ġto Ġdate . to Cloud Date ()) ĊĠĠĠĠĠ Ġset (" Step Index " Ġto Ġstep Index ) ĊĠĠĠĠĠ Ġset (" Attempt Index " Ġto Ġattempt Index ) ĊĠĠĠĠĠ Ġset (" State " Ġto ĠExchange Step Attempt . State . ACTIVE ) ĊĊĠĠĠĠĠ Ġ// ĠTODO (@ e fox ep stein ): Ġmake Ġthis Ġvariable Ġbased Ġon Ġthe Ġstep Ġtype Ġor Ġsomething . ĊĠĠĠĠĠ Ġset (" Expiration Time " Ġto Ġ( now Ġ+ ĠDEFAULT _ EXPI RATION _ DURATION ). to G cloud Timestamp ()) ĊĊĠĠĠĠĠ Ġset (" Exchange Step Attempt Details " Ġto Ġdetails ) ĊĠĠĠĠĠ Ġset Json (" Exchange Step Attempt Details Json " Ġto Ġdetails ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreturn Ġattempt Index ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun ĠTransaction Scope . find Attempt Index ( ĊĠĠĠ Ġrecur ring Exchange Id : ĠLong , ĊĠĠĠ Ġdate : ĠDate , ĊĠĠĠ Ġstep Index : ĠLong ĊĠ Ġ): ĠLong Ġ{ ĊĠĠĠ Ġval Ġsql Ġ= ĊĠĠĠĠĠ Ġ""" ĊĠĠĠĠĠ ĠSELECT ĠIF NULL ( MAX ( Attempt Index ), Ġ 0 ) ĠAS ĠMax Attempt Index ĊĠĠĠĠĠ ĠFROM ĠExchange Step Attempts ĊĠĠĠĠĠ ĠWHERE ĠExchange Step Attempts . Recurring Exchange Id Ġ= Ġ@ recurring _ exchange _ id ĊĠĠĠĠĠĠĠ ĠAND ĠExchange Step Attempts . Date Ġ= Ġ@ date ĊĠĠĠĠĠĠĠ ĠAND ĠExchange Step Attempts . Step Index Ġ= Ġ@ step _ index ĊĠĠĠĠĠ Ġ""" . trim Indent () ĊĊĠĠĠ Ġval Ġstatement : ĠStatement Ġ= ĊĠĠĠĠĠ Ġstatement ( sql ) Ġ{ ĊĠĠĠĠĠĠĠ Ġbind (" recurring _ exchange _ id "). to ( recurring Exchange Id ) ĊĠĠĠĠĠĠĠ Ġbind (" date "). to ( date . to Cloud Date ()) ĊĠĠĠĠĠĠĠ Ġbind (" step _ index "). to ( step Index ) ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġrow : ĠStruct Ġ= Ġtransaction Context . executeQuery ( statement ). single () ĊĊĠĠĠ Ġreturn Ġrow . getLong (" Max Attempt Index ") Ġ+ Ġ 1 L ĊĠ Ġ} Ċ } Ċ