package Ġnet . cord a . node . services . events Ċ Ċ import Ġco . parallel universe . fib ers . Suspend able Ċ import Ġcom . google . common . util . concurrent . List enable Future Ċ import Ġnet . cord a . core . concurrent . C ord a Future Ċ import Ġnet . cord a . core . context . Invocation Context Ċ import Ġnet . cord a . core . context . Invocation Origin Ċ import Ġnet . cord a . core . contracts . S chedul able State Ċ import Ġnet . cord a . core . contracts . Scheduled Activity Ċ import Ġnet . cord a . core . contracts . Scheduled State Ref Ċ import Ġnet . cord a . core . contracts . State Ref Ċ import Ġnet . cord a . core . flows . Flow Logic Ċ import Ġnet . cord a . core . flows . Flow Logic Ref Factory Ċ import Ġnet . cord a . core . flows . StateMachine Run Id Ċ import Ġnet . cord a . core . internal .* Ċ import Ġnet . cord a . core . internal . concurrent . flatMap Ċ import Ġnet . cord a . core . internal . concurrent . open Future Ċ import Ġnet . cord a . core . node . Services For Resolution Ċ import Ġnet . cord a . core . schemas . Persistent State Ref Ċ import Ġnet . cord a . core . serialization . Singleton Serialize As Token Ċ import Ġnet . cord a . core . utilities . context Logger Ċ import Ġnet . cord a . core . utilities . seconds Ċ import Ġnet . cord a . core . utilities . trace Ċ import Ġnet . cord a . node . C ord aC lock Ċ import Ġnet . cord a . node . Mutable Clock Ċ import Ġnet . cord a . node . services . api . Flow Starter Ċ import Ġnet . cord a . node . services . api . Node Properties Store Ċ import Ġnet . cord a . node . services . api . Scheduler Service Ċ import Ġnet . cord a . node . services . messaging . Ded up lication Handler Ċ import Ġnet . cord a . node . services . state machine . External Event Ċ import Ġnet . cord a . node api . internal . persistence . C ord a Persistence Ċ import Ġnet . cord a . node api . internal . persistence . NODE _ DATABASE _ PREFIX Ċ import Ġnet . cord a . node api . internal . persistence . context Transaction Ċ import Ġorg . apache . activ emq . art emis . utils . Re usable Latch Ċ import Ġorg . slf 4 j . Logger Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent .* Ċ import Ġjavax . annotation . concurrent . Thread Safe Ċ import Ġjavax . persistence . Column Ċ import Ġjavax . persistence . Embedded Id Ċ import Ġjavax . persistence . Entity Ċ import Ġco . parallel universe . str ands . Set table Future Ġas ĠQu asar Set table Future Ċ import Ġcom . google . common . util . concurrent . Set table Future Ġas ĠGu ava Set table Future Ċ Ċ /** Ċ Ġ* ĠA Ġfirst Ġpass Ġof Ġa Ġsimple Ġ[ Scheduler Service ] Ġthat Ġworks Ġwith Ġ[ Mutable Clock ] s Ġfor Ġtesting , Ġdemonstr ations Ġand Ġsimulations Ċ Ġ* Ġthat Ġalso Ġen comp asses Ġthe Ġ[ net . cord a . core . node . services . Vault ] Ġobserver Ġfor Ġprocessing Ġtransactions . Ċ Ġ* Ċ Ġ* ĠThis Ġwill Ġobserve Ġtransactions Ġas Ġthey Ġare Ġstored Ġand Ġschedule Ġand Ġun schedule Ġactivities Ġbased Ġon Ġthe ĠStates Ġconsumed Ċ Ġ* Ġor Ġproduced . Ċ Ġ* Ċ Ġ* ĠCurrently Ġdoes Ġnot Ġprovide Ġany Ġsystem Ġstate Ġother Ġthan Ġthe ĠContract State Ġso Ġthe Ġexpectation Ġis Ġthat Ġa Ġtransaction Ċ Ġ* Ġis Ġthe Ġoutcome Ġof Ġthe Ġactivity Ġin Ġorder Ġto Ġschedule Ġanother Ġactivity . Ġ ĠOnce Ġwe Ġhave Ġimplemented Ġmore Ġpersistence Ċ Ġ* Ġin Ġthe Ġnodes , Ġmaybe Ġwe Ġcan Ġconsider Ġmultiple Ġactivities Ġand Ġwhether Ġthe Ġactivities Ġhave Ġbeen Ġcompleted Ġor Ġnot , Ċ Ġ* Ġbut Ġthat Ġstarts Ġto Ġsound Ġa Ġlot Ġlike Ġoff - ledger Ġstate . Ċ Ġ*/ Ċ @ Thread Safe Ċ class ĠNode Scheduler Service ( private Ġval Ġclock : ĠC ord aC lock , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġdatabase : ĠC ord a Persistence , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġflow Starter : ĠFlow Starter , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġservices For Resolution : ĠServices For Resolution , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġflow Logic Ref Factory : ĠFlow Logic Ref Factory , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġnode Properties : ĠNode Properties Store , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġdr aining Mode Poll Period : ĠDuration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog : ĠLogger Ġ= Ġstatic Log , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġun finished Schedules : ĠRe usable Latch Ġ= ĠRe usable Latch (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġscheduler Repo : ĠScheduled Flow Repository Ġ= ĠPersistent Scheduled Flow Repository ( database )) ĊĠĠĠ Ġ: ĠScheduler Service , ĠAuto Closeable , ĠSingleton Serialize As Token () Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġstatic Log Ġget () Ġ= Ġcontext Logger () ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠWait Ġuntil Ġthe Ġgiven Ġ[ Future ] Ġis Ġcomplete Ġor Ġthe Ġdeadline Ġis Ġreached , Ġwith Ġsupport Ġfor Ġ[ Mutable Clock ] Ġimplementations ĊĠĠĠĠĠĠĠĠ Ġ* Ġused Ġin Ġde mos Ġor Ġtesting . Ġ ĠThis Ġwill Ġsubstitute Ġa ĠF iber Ġcompatible ĠFuture Ġso Ġthe Ġcurrent ĊĠĠĠĠĠĠĠĠ Ġ* Ġ[ co . parallel universe . str ands . Str and ] Ġis Ġnot Ġblocked . ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġ[ Future ] Ġis Ġcomplete , Ġfalse Ġif Ġthe Ġdeadline Ġwas Ġreached . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġshould Ġtry Ġto Ġmake Ġthe ĠClock Ġused Ġin Ġour Ġcode Ġinject able Ġ( for Ġtests Ġetc ) Ġand Ġto Ġuse Ġthe Ġextension Ġbelow ĊĠĠĠĠĠĠĠ Ġ// Ġto Ġwait Ġin Ġour Ġcode , Ġrather Ġthan Ġ< code > Thread . sleep ()</ code > Ġor Ġother Ġtime - based Ġpause s . ĊĠĠĠĠĠĠĠ Ġ@ Suspend able ĊĠĠĠĠĠĠĠ Ġ@ Visible ForTesting ĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġspecify Ġfull Ġclasspath Ġon ĠSet table Future Ġto Ġdifferent iate Ġit Ġfrom Ġthe ĠQu asar Ġclass Ġof Ġthe Ġsame Ġname ĊĠĠĠĠĠĠĠ Ġfun Ġawait With Deadline ( clock : ĠC ord aC lock , Ġdeadline : ĠInstant , Ġfuture : ĠFuture < *> Ġ= ĠGu ava Set table Future . create < Any >() ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġnan os : ĠLong ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdo Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoriginal Future Completed Ġ= Ġmake Str and Friendly Set table Future ( future ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscription Ġ= Ġclock . mutations . first (). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoriginal Future Completed . set ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnan os Ġ= Ġ( clock . instant () Ġuntil Ġdeadline ). to Nan os () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( nan os Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThis Ġwill Ġreturn Ġwhen Ġit Ġtimes Ġout , Ġor Ġwhen Ġthe Ġclock Ġmut ates Ġor Ġwhen Ġwhen Ġthe Ġoriginal Ġfuture Ġcompletes . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoriginal Future Completed . get ( nan os , ĠTimeUnit . NAN O SECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠExecution Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNo Ġneed Ġto Ġtake Ġaction Ġas Ġwill Ġfall Ġout Ġof Ġthe Ġloop Ġdue Ġto Ġfuture . is Done ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNo Ġneed Ġto Ġtake Ġaction Ġas Ġwill Ġfall Ġout Ġof Ġthe Ġloop Ġdue Ġto Ġfuture . is Done ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠTimeout Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNo Ġneed Ġto Ġtake Ġaction Ġas Ġwill Ġfall Ġout Ġof Ġthe Ġloop Ġdue Ġto Ġfuture . is Done ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription . unsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoriginal Future Completed . cancel ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġwhile Ġ( nan os Ġ> Ġ 0 Ġ&& Ġ! future . is Done ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfuture . is Done ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠConvert Ġa ĠGu ava Ġ[ List enable Future ] Ġor ĠJDK 8 Ġ[ Comp letableFuture ] Ġto ĠQu asar Ġimplementation Ġand Ġset Ġto Ġtrue Ġwhen Ġa Ġresult ĊĠĠĠĠĠĠĠĠ Ġ* Ġor Ġ[ Throwable ] Ġis Ġavailable Ġin Ġthe Ġoriginal . ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* ĠWe Ġneed Ġthis Ġso Ġthat Ġwe Ġdo Ġnot Ġblock Ġthe Ġactual Ġthread Ġwhen Ġcalling Ġget (), Ġbut Ġinstead Ġallow Ġa ĠQu asar Ġcontext ĊĠĠĠĠĠĠĠĠ Ġ* Ġswitch . Ġ ĠThere 's Ġno Ġneed Ġto Ġcheckpoint Ġour Ġ[ co . parallel universe . fib ers . Fi ber ] s Ġas Ġthere 's Ġno Ġexternal Ġeffect Ġof Ġwaiting . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġ< T Ġ: ĠAny > Ġmake Str and Friendly Set table Future ( future : ĠFuture < T >) Ġ= ĠQu asar Set table Future < Boolean >(). also Ġ{ Ġg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( future ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠList enable Future Ġ-> Ġfuture . addListener ( Runnable Ġ{ Ġg . set ( true ) Ġ}, ĠExecutor Ġ{ Ġit . run () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠCompletion Stage < *> Ġ-> Ġfuture . when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġg . set ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalArgumentException (" Cannot Ġmake Ġfuture Ġ$ future ĠStr and Ġfriendly .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ@ Visible ForTesting ĊĠĠĠĠĠĠĠ Ġinternal Ġval Ġscheduling As Next Format Ġ= Ġ" Scheduling Ġas Ġnext Ġ{} " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } scheduled _ states ") ĊĠĠĠ Ġclass ĠPersistent Scheduled State ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Embedded Id ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġoutput : ĠPersistent State Ref , ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" scheduled _ at ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġscheduled At : ĠInstant Ġ= ĠInstant . now () ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġclass ĠInner State Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġre scheduled : ĠGu ava Set table Future < Boolean >? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġnext Scheduled Action : ĠScheduled State Ref ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġrunning : ĠBoolean Ġ= Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠUsed Ġto Ġde - duplicate Ġflow Ġstarts Ġin Ġcase Ġa Ġflow Ġis Ġstarting Ġbut Ġthe Ġcorresponding Ġentry Ġhasn 't Ġbeen Ġremoved Ġyet ĊĠĠĠ Ġ// Ġfrom Ġthe Ġdatabase ĊĠĠĠ Ġprivate Ġval Ġstarting State Refs : ĠMutable Set < Scheduled State Ref > Ġ= ĠConcurrent HashMap . new Key Set < Scheduled State Ref >() ĊĠĠĠ Ġprivate Ġval Ġmutex Ġ= ĠThread Box ( Inner State ()) ĊĠĠĠ Ġprivate Ġval Ġscheduler Timer Executor Ġ= ĠExec utors . new Single Thread Executor () ĊĊĠĠĠ Ġ// Ġif Ġthere 's Ġnothing Ġto Ġdo , Ġcheck Ġevery Ġminute Ġif Ġsomething Ġf ell Ġthrough Ġthe Ġcr acks . ĊĠĠĠ Ġ// Ġany Ġnew Ġstate Ġshould Ġtrigger Ġa Ġre schedule Ġimmediately Ġif Ġnothing Ġis Ġscheduled , Ġso ĠI Ġwould Ġnot Ġexpect ĊĠĠĠ Ġ// Ġthis Ġto Ġusually Ġtrigger Ġanything . ĊĠĠĠ Ġprivate Ġval Ġidle Wait Seconds Ġ= Ġ 6 0 . seconds ĊĊĠĠĠ Ġ// ĠWe Ġneed Ġthe Ġ[ StateMachine Manager ] Ġto Ġbe Ġconstructed Ġbefore Ġthis Ġis Ġcalled Ġin Ġcase Ġit Ġs chedules Ġa Ġflow . ĊĠĠĠ Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġscheduler Timer Executor . execute Ġ{ Ġrun Loop Function () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġschedule State Activity ( action : ĠScheduled State Ref ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Schedule Ġ$ action " Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠOnly Ġincrease Ġthe Ġnumber Ġof Ġun finished Ġs chedules Ġif Ġthe Ġstate Ġdidn 't Ġalready Ġexist Ġon Ġthe Ġqueue ĊĠĠĠĠĠĠĠ Ġif Ġ(! scheduler Repo . merge ( action )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġun finished Schedules . count Up () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġcontext Transaction . on Commit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmutex . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( action . scheduled At Ġ< Ġnext Scheduled Action ?. scheduled At Ġ?: ĠInstant . MAX ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġare Ġear liest ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre schedule Wake Up () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( action . ref Ġ== Ġnext Scheduled Action ?. ref Ġ&& Ġaction . scheduled At Ġ!= Ġnext Scheduled Action ?. scheduled At ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġwere Ġear liest Ġbut Ġmight Ġnot Ġbe Ġany Ġmore ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre schedule Wake Up () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġun schedule State Activity ( ref : ĠState Ref ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Un schedule Ġ$ ref " Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcount Down Ġ= Ġstarting State Refs . all Ġ{ Ġit . ref Ġ!= Ġref Ġ} Ġ&& Ġscheduler Repo . delete ( ref ) ĊĠĠĠĠĠĠĠ Ġcontext Transaction . on Commit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( count Down ) Ġun finished Schedules . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmutex . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( next Scheduled Action ?. ref Ġ== Ġref ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre schedule Wake Up () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġrun Loop Function () Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( mutex . locked Ġ{ Ġrunning Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( scheduled State , Ġour Res chedule d Future ) Ġ= Ġmutex . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre scheduled Ġ= ĠGu ava Set table Future . create () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// get Ġthe Ġnext Ġscheduled Ġaction Ġthat Ġisn 't Ġcurrently Ġrunning ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġde duplicate Ġ= ĠHashSet ( starting State Refs ) Ġ// ĠTake Ġan Ġimmutable Ġcopy Ġto Ġremove Ġr aces Ġwith Ġafter Database Commit . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnext Scheduled Action Ġ= Ġscheduler Repo . get Latest ( ded uplicate . size Ġ+ Ġ 1 ). first OrNull Ġ{ Ġ! ded uplicate . contains ( it . second ) Ġ} ?. second ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( next Scheduled Action , Ġre scheduled !! ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace ( scheduling As Next Format , Ġscheduled State ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThis Ġwill Ġblock Ġthe Ġscheduler Ġsingle Ġthread Ġuntil Ġthe Ġscheduled Ġtime Ġ( returns Ġfalse ) ĠOR ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthe ĠFuture Ġis Ġcancelled Ġdue Ġto Ġre scheduling Ġ( returns Ġtrue ). ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( scheduled State Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! await With Deadline ( clock , Ġscheduled State . scheduled At , Ġour Res chedule d Future )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Inv oking Ġas Ġnext Ġ$ scheduled State " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Time Reached ( scheduled State ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Res chedule d Ġ$ scheduled State " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġawait With Deadline ( clock , Ġclock . instant () Ġ+ Ġidle Wait Seconds , Ġour Res chedule d Future ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġre schedule Wake Up () Ġ{ ĊĠĠĠĠĠĠĠ Ġmutex . already Locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġre scheduled ?. cancel ( false ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġmutex . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrunning Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġre schedule Wake Up () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġscheduler Timer Executor . join () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Visible ForTesting ĊĠĠĠ Ġinternal Ġfun Ġcancel And Wait () Ġ{ ĊĠĠĠĠĠĠĠ Ġscheduler Timer Executor . shutdown Now () ĊĠĠĠĠĠĠĠ Ġscheduler Timer Executor . join () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinner Ġclass ĠFlow Start Ded up lication Handler ( val Ġscheduled State : ĠScheduled State Ref , Ġoverride Ġval Ġflow Logic : ĠFlow Logic < Any ? >, Ġoverride Ġval Ġcontext : ĠInvocation Context ) Ġ: ĠD ed up lication Handler , ĠExternal Event . External Start Flow Event < Any ?> Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġflow Id : ĠState Machine Run Id Ġ= ĠState Machine Run Id . create Random () ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġexternal Cause : ĠExternal Event ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġthis ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġde dup lication Handler : ĠFlow Start Ded up lication Handler ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġthis ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġinside Database Transaction () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler Repo . delete ( scheduled State . ref ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġafter Database Transaction () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstarting State Refs . remove ( scheduled State ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ"${ java Class . simple Name } ($ scheduled State )" ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġwire Up Future ( flow Future : ĠC ord a Future < out ĠFlow StateMachine Handle < Any ? >>) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ_ future . capture Later ( flow Future ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= Ġ_ future . flatMap Ġ{ Ġit . result Future Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfuture . then Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun finished Schedules . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġ_ future Ġ= Ġopen Future < Flow StateMachine Handle < Any ? >>() ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġfuture : ĠC ord a Future < Flow StateMachine Handle < Any ? >> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġ_ future ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġon Time Reached ( scheduled State : ĠScheduled State Ref ) Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġflow Name : ĠString ? Ġ= Ġ"( unknown )" ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase . transaction Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġscheduled Flow Ġ= Ġget Flow ( scheduled State ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( scheduled Flow Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Name Ġ= Ġscheduled Flow . java Class . name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO Ġrefactor Ġthe Ġscheduler Ġto Ġstore Ġand Ġpropagate Ġthe Ġoriginal Ġinvocation Ġcontext ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= ĠInvocation Context . newInstance ( Invocation Origin . Scheduled ( scheduled State )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstart Flow Event Ġ= ĠFlow Start Ded up lication Handler ( scheduled State , Ġscheduled Flow , Ġcontext ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Starter . start Flow ( start Flow Event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġstart Ġscheduled Ġflow Ġ$ flow Name Ġfor Ġ$ scheduled State Ġdue Ġto Ġan Ġinternal Ġerror ", Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Flow ( scheduled State : ĠScheduled State Ref ): ĠFlow Logic < *> ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġscheduled Activity Ġ= Ġget Scheduled Activity ( scheduled State ) ĊĠĠĠĠĠĠĠ Ġvar Ġscheduled Flow : ĠFlow Logic < *> ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġmutex . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġneed Ġto Ġremove Ġus Ġfrom Ġthose Ġscheduled , Ġbut Ġonly Ġif Ġwe Ġare Ġstill Ġnext ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( next Scheduled Action Ġ!= Ġnull Ġ&& Ġnext Scheduled Action Ġ=== Ġscheduled State ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( scheduled Activity Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Scheduled Ġstate Ġ$ scheduled State Ġhas Ġre scheduled Ġto Ġnever .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun finished Schedules . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler Repo . delete ( scheduled State . ref ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( scheduled Activity . scheduled At . is After ( clock . instant ())) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Scheduled Ġstate Ġ$ scheduled State Ġhas Ġre scheduled Ġto Ġ${ scheduled Activity . scheduled At }. ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠnewState Ġ= ĠScheduled State Ref ( scheduled State . ref , Ġscheduled Activity . scheduled At ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler Repo . merge ( newState ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Logic Ġ= Ġflow Logic Ref Factory . to Flow Logic ( scheduled Activity . logic Ref ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduled Flow Ġ= Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnode Properties . flows Dr aining Mode . isEnabled () Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Ign oring Ġscheduled Ġflow Ġstart Ġbecause Ġof Ġdr aining Ġmode . ĠFlow Logic : Ġ$ flow Logic .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġawait With Deadline ( clock , ĠInstant . now () Ġ+ Ġdr aining Mode Poll Period ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Scheduler Ġstarting ĠFlow Logic Ġ$ flow Logic " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Add Ġthis Ġto Ġthe Ġin Ġmemory Ġlist Ġof Ġstarting Ġrefs Ġso Ġit Ġis Ġnot Ġpicked Ġup Ġon Ġthe Ġnext Ġre schedule Wake Up () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstarting State Refs . add ( scheduled State ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Logic ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġscheduled Flow ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Scheduled Activity ( scheduled State : ĠScheduled State Ref ): ĠScheduled Activity ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtx State Ġ= Ġservices For Resolution . load State ( scheduled State . ref ) ĊĠĠĠĠĠĠĠ Ġval Ġstate Ġ= Ġtx State . data Ġas ĠS chedul able State ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThis Ġcan Ġthrow Ġas Ġrunning Ġcontract Ġcode . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate . next Scheduled Activity ( scheduled State . ref , Ġflow Logic Ref Factory ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Attempt Ġto Ġrun Ġscheduled Ġstate Ġ$ scheduled State Ġresulted Ġin Ġerror .", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ