package Ġnet . cord a . node . services . persistence Ċ Ċ import Ġnet . cord a . core . context . Invocation Context Ċ import Ġnet . cord a . core . context . Invocation Origin Ċ import Ġnet . cord a . core . flows . StateMachine Run Id Ċ import Ġnet . cord a . core . internal . PLATFORM _ VERSION Ċ import Ġnet . cord a . core . internal . Visible ForTesting Ċ import Ġnet . cord a . core . internal . unchecked Cast Ċ import Ġnet . cord a . core . flows . Result Serialization Exception Ċ import Ġnet . cord a . core . serialization . Serialization Defaults Ċ import Ġnet . cord a . core . serialization . Serialized Bytes Ċ import Ġnet . cord a . core . serialization . deserialize Ċ import Ġnet . cord a . core . serialization . internal . Missing Serializer Exception Ċ import Ġnet . cord a . core . serialization . serialize Ċ import Ġnet . cord a . core . utilities . context Logger Ċ import Ġnet . cord a . node . services . api . Checkpoint Storage Ċ import Ġnet . cord a . node . services . state machine . Checkpoint Ċ import Ġnet . cord a . node . services . state machine . Checkpoint . Flow Status Ċ import Ġnet . cord a . node . services . state machine . Checkpoint State Ċ import Ġnet . cord a . node . services . state machine . Error State Ċ import Ġnet . cord a . node . services . state machine . Flow Result Metadata Ċ import Ġnet . cord a . node . services . state machine . Flow State Ċ import Ġnet . cord a . node . services . state machine . Sub Flow Version Ċ import Ġnet . cord a . node api . internal . persistence . NODE _ DATABASE _ PREFIX Ċ import Ġnet . cord a . node api . internal . persistence . current DB Session Ċ import Ġorg . apache . commons . lang 3 . Array Utils . EMPTY _ BYTE _ ARRAY Ċ import Ġorg . apache . commons . lang 3 . exception . Exception Utils Ċ import Ġorg . hibernate . annotations . Type Ċ import Ġjava . sql . Connection Ċ import Ġjava . sql . SQLException Ċ import Ġjava . time . Clock Ċ import Ġjava . time . Instant Ċ import Ġjava . util .* Ċ import Ġjava . util . stream . Stream Ċ import Ġjavax . persistence . Column Ċ import Ġjavax . persistence . Entity Ċ import Ġjavax . persistence . Fetch Type Ċ import Ġjavax . persistence . Id Ċ import Ġjavax . persistence . One ToOne Ċ import Ġjavax . persistence . PrimaryKey JoinColumn Ċ Ċ /** Ċ Ġ* ĠSimple Ġcheckpoint Ġkey Ġvalue Ġstorage Ġin ĠDB . Ċ Ġ*/ Ċ @ Suppress (" Too Many Functions ") Ċ class ĠDB Checkpoint Storage ( ĊĠĠĠ Ġprivate Ġval Ġcheckpoint Performance Recorder : ĠCheck point Performance Recorder , ĊĠĠĠ Ġprivate Ġval Ġclock : ĠClock Ċ ) Ġ: ĠCheck point Storage Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġlog Ġ= Ġcontext Logger () ĊĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠHMAC _ SIZE _ BYTES Ġ= Ġ 1 6 ĊĊĠĠĠĠĠĠĠ Ġ@ Visible ForTesting ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠMAX _ STACK TRACE _ LENGTH Ġ= Ġ 2 0 0 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠMAX _ EX C _ MSG _ LENGTH Ġ= Ġ 2 0 0 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠMAX _ EX C _ TYPE _ LENGTH Ġ= Ġ 2 5 6 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠMAX _ FLOW _ NAME _ LENGTH Ġ= Ġ 1 2 8 ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠMAX _ PROGRESS _ STEP _ LENGTH Ġ= Ġ 2 5 6 ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠMAX _ CLIENT _ ID _ LENGTH Ġ= Ġ 5 1 2 ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠRUN N ABLE _ CHECK POINTS Ġ= Ġset Of ( Flow Status . RUN N ABLE , ĠFlow Status . HO SP ITAL IZED ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠThis Ġis Ġa Ġdummy Ġ[ DB Flow Metadata ] Ġobject Ġwhich Ġhelp Ġus Ġwhenever Ġwe Ġwant Ġto Ġpersist Ġa Ġ[ DB Flow Checkpoint ], Ġbut Ġnot Ġpersist Ġits Ġ[ DB Flow Metadata ]. ĊĠĠĠĠĠĠĠ Ġ// Ġ[ DB Flow Checkpoint ] Ġneeds Ġto Ġalways Ġreference Ġa Ġ[ DB Flow Metadata ] Ġ([ DB Flow Checkpoint . flow Metadata ] Ġis Ġnot Ġnullable ). ĊĠĠĠĠĠĠĠ Ġ// ĠHowever , Ġsince Ġwe Ġdo Ġnot Ġ- hibernate - Ġcascade , Ġit Ġdoes Ġnot Ġget Ġpersisted Ġinto Ġthe Ġdatabase . ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġdummy DB Flow Metadata : ĠDB Flow Metadata Ġ= ĠDB Flow Metadata ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id Ġ= Ġ" dummy Flow Id ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinvocation Id Ġ= Ġ" dummy Invocation Id ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Name Ġ= Ġ" dummy Flow Name ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġuser Supp lied Identifier Ġ= Ġ" dummy User Supp lied Identifier ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Type Ġ= ĠStart Reason . INIT I ATED , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Parameters Ġ= ĠByteArray ( 0 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ing C ord app Ġ= Ġ" dummy Launch ing C ord app ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplatform Version Ġ= Ġ- 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstarted By Ġ= Ġ" dummy Started By ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinvocation Instant Ġ= ĠInstant . now (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Instant Ġ= ĠInstant . now (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfinish Instant Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThis Ġneeds Ġto Ġrun Ġbefore ĠHibernate Ġis Ġinitial ised . ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* ĠNo Ġneed Ġto Ġset Ġup Ġ[ DB Checkpoint Storage ] Ġfully Ġfor Ġthis Ġfunction ĊĠĠĠĠĠĠĠĠ Ġ* ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ param Ġconnection ĠThe ĠSQL ĠConnection . ĊĠĠĠĠĠĠĠĠ Ġ* Ġ@ return Ġthe Ġnumber Ġof Ġcheck points Ġstored Ġin Ġthe Ġdatabase . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġfun Ġget Checkpoint Count ( connection : ĠConnection ): ĠLong Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNo Ġneed Ġto Ġset Ġup Ġ[ DB Checkpoint Storage ] Ġfully Ġfor Ġthis Ġfunction ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . prepareStatement (" select Ġcount (*) Ġfrom Ġnode _ check points "). use Ġ{ Ġps Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġps . executeQuery (). use Ġ{ Ġrs Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrs . next () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrs . getLong ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠSQLException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠH app ens Ġwhen Ġthe Ġtable Ġwas Ġnot Ġcreated Ġyet . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġenum Ġclass ĠStart Reason Ġ{ ĊĠĠĠĠĠĠĠ ĠRPC , ĠSERVICE , ĠS CH ED UL ED , ĠINIT I ATED ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } check points ") ĊĠĠĠ Ġdata Ġclass ĠDB Flow Checkpoint ( ĊĠĠĠĠĠĠĠ Ġ@ Id ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ id ", Ġlength Ġ= Ġ 6 4 , Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow Id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ One ToOne ( fetch Ġ= ĠFetch Type . LA ZY , Ġoptional Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġ@ PrimaryKey JoinColumn ĊĠĠĠĠĠĠĠ Ġvar Ġblob : ĠDB Flow Checkpoint Blob ?, ĊĊĠĠĠĠĠĠĠ Ġ@ One ToOne ( fetch Ġ= ĠFetch Type . LA ZY , Ġoptional Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġ@ PrimaryKey JoinColumn ĊĠĠĠĠĠĠĠ Ġvar Ġresult : ĠDB Flow Result ?, ĊĊĠĠĠĠĠĠĠ Ġ@ One ToOne ( fetch Ġ= ĠFetch Type . LA ZY , Ġoptional Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġ@ PrimaryKey JoinColumn ĊĠĠĠĠĠĠĠ Ġvar Ġexception Details : ĠDB Flow Exception ?, ĊĊĠĠĠĠĠĠĠ Ġ@ One ToOne ( fetch Ġ= ĠFetch Type . LA ZY ) ĊĠĠĠĠĠĠĠ Ġ@ PrimaryKey JoinColumn ĊĠĠĠĠĠĠĠ Ġvar Ġflow Metadata : ĠDB Flow Metadata , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" status ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġstatus : ĠFlow Status , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" compatible ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġcompatible : ĠBoolean , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" progress _ step ") ĊĠĠĠĠĠĠĠ Ġvar Ġprogress Step : ĠString ?, ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ io _ request ") ĊĠĠĠĠĠĠĠ Ġvar Ġio RequestType : ĠString ?, ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" timestamp ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġcheckpoint Instant : ĠInstant ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } checkpoint _ blobs ") ĊĠĠĠ Ġdata Ġclass ĠDB Flow Checkpoint Blob ( ĊĠĠĠĠĠĠĠ Ġ@ Id ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ id ", Ġlength Ġ= Ġ 6 4 , Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow Id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - blob ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" checkpoint _ value ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġcheckpoint : ĠByteArray Ġ= ĠEMPTY _ BYTE _ ARRAY , ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - blob ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ state ", Ġnullable Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow Stack : ĠByteArray ?, ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - wrapper - binary ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" hmac ") ĊĠĠĠĠĠĠĠ Ġvar Ġhmac : ĠByteArray , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" timestamp ") ĊĠĠĠĠĠĠĠ Ġvar Ġpersisted Instant : ĠInstant ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother ?. java Class ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġother Ġas ĠDB Flow Checkpoint Blob ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flow Id Ġ!= Ġother . flow Id ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! checkpoint . content Equals ( other . checkpoint )) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(!( flow Stack Ġ?: ĠEMPTY _ BYTE _ ARRAY ) !!. content Equals ( other . flow Stack Ġ?: ĠEMPTY _ BYTE _ ARRAY )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! hmac . content Equals ( other . hmac )) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( pers isted Instant Ġ!= Ġother . pers isted Instant ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġflow Id . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġcheckpoint . content HashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( flow Stack ?. content HashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġhmac . content HashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġpersisted Instant . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } flow _ results ") ĊĠĠĠ Ġdata Ġclass ĠDB Flow Result ( ĊĠĠĠĠĠĠĠ Ġ@ Id ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ id ", Ġlength Ġ= Ġ 6 4 , Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow _ id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - blob ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" result _ value ", Ġnullable Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġvar Ġvalue : ĠByteArray ? Ġ= Ġnull , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" timestamp ") ĊĠĠĠĠĠĠĠ Ġval Ġpersisted Instant : ĠInstant ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" Complex Method ") ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother ?. java Class ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġother Ġas ĠDB Flow Result ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flow _ id Ġ!= Ġother . flow _ id ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= Ġvalue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġother Value Ġ= Ġother . value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( value Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( other Value Ġ== Ġnull ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! value . content Equals ( other Value )) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( other Value Ġ!= Ġnull ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( pers isted Instant Ġ!= Ġother . pers isted Instant ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġflow _ id . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( value ?. content HashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġpersisted Instant . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } flow _ exceptions ") ĊĠĠĠ Ġdata Ġclass ĠDB Flow Exception ( ĊĠĠĠĠĠĠĠ Ġ@ Id ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ id ", Ġlength Ġ= Ġ 6 4 , Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow _ id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" type ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġtype : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" exception _ message ") ĊĠĠĠĠĠĠĠ Ġvar Ġmessage : ĠString ? Ġ= Ġnull , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" stack _ trace ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġstack Trace : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - blob ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" exception _ value ") ĊĠĠĠĠĠĠĠ Ġvar Ġvalue : ĠByteArray ? Ġ= Ġnull , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" timestamp ") ĊĠĠĠĠĠĠĠ Ġval Ġpersisted Instant : ĠInstant ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother ?. java Class ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġother Ġas ĠDB Flow Exception ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flow _ id Ġ!= Ġother . flow _ id ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( type Ġ!= Ġother . type ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( message Ġ!= Ġother . message ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( stack Trace Ġ!= Ġother . stack Trace ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(!( value Ġ?: ĠEMPTY _ BYTE _ ARRAY ) !!. content Equals ( other . value Ġ?: ĠEMPTY _ BYTE _ ARRAY )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( pers isted Instant Ġ!= Ġother . pers isted Instant ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġflow _ id . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġtype . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( message ?. hashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġstack Trace . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( value ?. content HashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġpersisted Instant . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Entity ĊĠĠĠ Ġ@ javax . persistence . Table ( name Ġ= Ġ"${ NODE _ DATABASE _ PREFIX } flow _ metadata ") ĊĠĠĠ Ġdata Ġclass ĠDB Flow Metadata ( ĊĠĠĠĠĠĠĠ Ġ@ Id ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ id ", Ġlength Ġ= Ġ 6 4 , Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow Id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" invocation _ id ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġinvocation Id : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ name ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġflow Name : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ identifier ", Ġnullable Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġvar Ġuser Supp lied Identifier : ĠString ?, ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" started _ type ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġstart Type : ĠStart Reason , ĊĊĠĠĠĠĠĠĠ Ġ@ Type ( type Ġ= Ġ" cord a - blob ") ĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" flow _ parameters ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġinitial Parameters : ĠByteArray Ġ= ĠEMPTY _ BYTE _ ARRAY , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" cord app _ name ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġlaunch ing C ord app : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" platform _ version ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġplatform Version : ĠInt , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" started _ by ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġstarted By : ĠString , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" invocation _ time ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġinvocation Instant : ĠInstant , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" start _ time ", Ġnullable Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġvar Ġstart Instant : ĠInstant , ĊĊĠĠĠĠĠĠĠ Ġ@ Column ( name Ġ= Ġ" finish _ time ", Ġnullable Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġvar Ġfinish Instant : ĠInstant ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" Complex Method ") ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother ?. java Class ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġother Ġas ĠDB Flow Metadata ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flow Id Ġ!= Ġother . flow Id ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( invocation Id Ġ!= Ġother . invocation Id ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flow Name Ġ!= Ġother . flow Name ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( user Supp lied Identifier Ġ!= Ġother . user Supp lied Identifier ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( start Type Ġ!= Ġother . start Type ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! initial Parameters . content Equals ( other . initial Parameters )) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( launch ing C ord app Ġ!= Ġother . launch ing C ord app ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( platform Version Ġ!= Ġother . platform Version ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( started By Ġ!= Ġother . started By ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( invocation Instant Ġ!= Ġother . invocation Instant ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( start Instant Ġ!= Ġother . start Instant ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( finish Instant Ġ!= Ġother . finish Instant ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġflow Id . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġinvocation Id . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġflow Name . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( user Supp lied Identifier ?. hashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġstart Type . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġinitial Parameters . content HashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġlaunch ing C ord app . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġplatform Version ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġstarted By . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġinvocation Instant . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġstart Instant . hashCode () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġ 3 1 Ġ* Ġresult Ġ+ Ġ( finish Instant ?. hashCode () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġadd Checkpoint ( ĊĠĠĠĠĠĠĠ Ġid : ĠState Machine Run Id , ĊĠĠĠĠĠĠĠ Ġcheckpoint : ĠCheck point , ĊĠĠĠĠĠĠĠ Ġserialized Flow State : ĠSerialized Bytes < Flow State >, ĊĠĠĠĠĠĠĠ Ġserialized Checkpoint State : ĠSerialized Bytes < Checkpoint State > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnow Ġ= Ġclock . instant () ĊĠĠĠĠĠĠĠ Ġval Ġflow Id Ġ= Ġid . uuid . toString () ĊĊĠĠĠĠĠĠĠ Ġcheckpoint Performance Recorder . record ( serialized Checkpoint State , Ġserialized Flow State ) ĊĊĠĠĠĠĠĠĠ Ġval Ġblob Ġ= Ġcreate DB Checkpoint Blob ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Checkpoint State , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Flow State , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnow ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġcreate DB Flow Metadata ( flow Id , Ġcheckpoint , Ġnow ) ĊĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Exception Ġ= Ġif Ġ( checkpoint . status Ġ== ĠFlow Status . FAILED Ġ|| Ġcheckpoint . status Ġ== ĠFlow Status . HO SP ITAL IZED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġerro red Ġ= Ġcheckpoint . error State Ġas ? ĠError State . Er rored ĊĠĠĠĠĠĠĠĠĠĠĠ Ġerro red ?. let Ġ{ Ġcreate DB Flow Exception ( flow Id , Ġit , Ġnow ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠIllegalStateException (" Found Ġ'${ checkpoint . status }' Ġcheckpoint Ġwhose Ġerror Ġstate Ġis Ġnot Ġ${ Error State . Er rored :: class . java . simple Name }") ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠMost Ġfields Ġare Ġnull Ġas Ġthey Ġcannot Ġhave Ġbeen Ġset Ġwhen Ġcreating Ġthe Ġinitial Ġcheckpoint ĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Checkpoint Ġ= ĠDB Flow Checkpoint ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblob Ġ= Ġblob , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexception Details Ġ= Ġdb Flow Exception , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Metadata Ġ= Ġmetadata , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstatus Ġ= Ġcheckpoint . status , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompatible Ġ= Ġcheckpoint . compatible , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprogress Step Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġio RequestType Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint Instant Ġ= Ġnow ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). save ( db Flow Checkpoint ) ĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). save ( blob ) ĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). save ( metadata ) ĊĠĠĠĠĠĠĠ Ġdb Flow Exception ?. let Ġ{ Ġcurrent DB Session (). save ( it ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" Complex Method ") ĊĠĠĠ Ġoverride Ġfun Ġupdate Checkpoint ( ĊĠĠĠĠĠĠĠ Ġid : ĠState Machine Run Id , ĊĠĠĠĠĠĠĠ Ġcheckpoint : ĠCheck point , ĊĠĠĠĠĠĠĠ Ġserialized Flow State : ĠSerialized Bytes < Flow State > ?, ĊĠĠĠĠĠĠĠ Ġserialized Checkpoint State : ĠSerialized Bytes < Checkpoint State > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnow Ġ= Ġclock . instant () ĊĠĠĠĠĠĠĠ Ġval Ġflow Id Ġ= Ġid . uuid . toString () ĊĊĠĠĠĠĠĠĠ Ġval Ġblob Ġ= Ġif Ġ( checkpoint . status Ġ== ĠFlow Status . HO SP ITAL IZED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDo Ġnot Ġupdate Ġ' checkpoint State ' Ġor Ġ' flow State ' Ġif Ġflow Ġhospital ized ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( checkpoint . status Ġ== ĠFlow Status . FAILED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġneed Ġto Ġupdate Ġonly Ġthe Ġ' flow State ' Ġto Ġnull , Ġand Ġwe Ġdon 't Ġwant Ġto Ġupdate Ġthe Ġcheckpoint Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbecause Ġwe Ġwant Ġto Ġretain Ġthe Ġlast Ġclean Ġcheckpoint Ġstate , Ġtherefore Ġjust Ġuse Ġa Ġquery Ġfor Ġthat Ġupdate . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsql Query Ġ= Ġ" Update Ġ${ NODE _ DATABASE _ PREFIX } checkpoint _ blobs Ġset Ġflow _ state Ġ= Ġnull Ġwhere Ġflow _ id Ġ= Ġ'$ flow Id '" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġcurrent DB Session (). create Native Query ( sql Query ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġquery . executeUpdate () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint Performance Recorder . record ( serialized Checkpoint State , Ġserialized Flow State ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate DB Checkpoint Blob ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Checkpoint State , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Flow State , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnow ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Result Ġ= Ġif Ġ( checkpoint . status Ġ== ĠFlow Status . COMPLETED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate DB Flow Result ( flow Id , Ġcheckpoint . result , Ġnow ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠMissing Serializer Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠResult Serialization Exception ( e ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Exception Ġ= Ġif Ġ( checkpoint . status Ġ== ĠFlow Status . FAILED Ġ|| Ġcheckpoint . status Ġ== ĠFlow Status . HO SP ITAL IZED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġerro red Ġ= Ġcheckpoint . error State Ġas ? ĠError State . Er rored ĊĠĠĠĠĠĠĠĠĠĠĠ Ġerro red ?. let Ġ{ Ġcreate DB Flow Exception ( flow Id , Ġit , Ġnow ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠIllegalStateException (" Found Ġ'${ checkpoint . status }' Ġcheckpoint Ġwhose Ġerror Ġstate Ġis Ġnot Ġ${ Error State . Er rored :: class . java . simple Name }") ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠUpdates Ġto Ġchildren Ġentities Ġ([ DB Flow Checkpoint Blob ], Ġ[ DB Flow Result ], Ġ[ DB Flow Exception ], Ġ[ DB Flow Metadata ]) Ġare Ġnot Ġcas ca ded Ġto Ġchildren Ġtables . ĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Checkpoint Ġ= ĠDB Flow Checkpoint ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblob Ġ= Ġblob , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġdb Flow Result , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexception Details Ġ= Ġdb Flow Exception , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Metadata Ġ= Ġdummy DB Flow Metadata , Ġ// Ġ[ DB Flow Metadata ] Ġwill Ġonly Ġupdate Ġits Ġ' finish _ time ' Ġwhen Ġa Ġcheckpoint Ġfinishes ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstatus Ġ= Ġcheckpoint . status , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompatible Ġ= Ġcheckpoint . compatible , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprogress Step Ġ= Ġcheckpoint . progress Step ?. take ( MAX _ PROGRESS _ STEP _ LENGTH ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġio RequestType Ġ= Ġcheckpoint . flow Io Request , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint Instant Ġ= Ġnow ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). update ( db Flow Checkpoint ) ĊĠĠĠĠĠĠĠ Ġblob ?. let Ġ{ Ġcurrent DB Session (). update ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġdb Flow Result ?. let Ġ{ Ġcurrent DB Session (). save ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġdb Flow Exception ?. let Ġ{ Ġcurrent DB Session (). save ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( checkpoint . is Finished ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset DB Flow Metadata Finish Time ( flow Id , Ġnow ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġmark All Paused () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġrunnable Ordinal s Ġ= ĠRUN N ABLE _ CHECK POINTS . map Ġ{ Ġ"${ it . ordinal }" Ġ}. join ToString Ġ{ Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġsql Query Ġ= Ġ" Update Ġ${ NODE _ DATABASE _ PREFIX } check points Ġset Ġstatus Ġ= Ġ${ Flow Status . PA USED . ordinal } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" where Ġstatus Ġin Ġ($ run nable Ordinal s )" ĊĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġsession . create Native Query ( sql Query ) ĊĠĠĠĠĠĠĠ Ġquery . executeUpdate () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" Magic Number ") ĊĠĠĠ Ġoverride Ġfun Ġremove Checkpoint ( id : ĠState Machine Run Id , Ġmay Have Persistent Results : ĠBoolean ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġdeleted Rows Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġval Ġflow Id Ġ= Ġid . uuid . toString () ĊĠĠĠĠĠĠĠ Ġdeleted Rows Ġ+= Ġdelete Row ( DB Flow Checkpoint :: class . java , ĠDB Flow Checkpoint :: flow Id . name , Ġflow Id ) ĊĠĠĠĠĠĠĠ Ġdeleted Rows Ġ+= Ġdelete Row ( DB Flow Checkpoint Blob :: class . java , ĠDB Flow Checkpoint Blob :: flow Id . name , Ġflow Id ) ĊĠĠĠĠĠĠĠ Ġif Ġ( may Have Persistent Results ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeleted Rows Ġ+= Ġdelete Row ( DB Flow Result :: class . java , ĠDB Flow Result :: flow _ id . name , Ġflow Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeleted Rows Ġ+= Ġdelete Row ( DB Flow Exception :: class . java , ĠDB Flow Exception :: flow _ id . name , Ġflow Id ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġdeleted Rows Ġ+= Ġdelete Row ( DB Flow Metadata :: class . java , ĠDB Flow Metadata :: flow Id . name , Ġflow Id ) ĊĠĠĠĠĠĠĠ Ġreturn Ġdeleted Rows Ġ>= Ġ 2 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġ< T > Ġdelete Row ( clazz : ĠClass < T >, Ġpk : ĠString , Ġvalue : ĠString ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġcriteria Builder Ġ= Ġsession . criteria Builder ĊĠĠĠĠĠĠĠ Ġval Ġdelete Ġ= Ġcriteria Builder . create Criteria Delete ( clazz ) ĊĠĠĠĠĠĠĠ Ġval Ġroot Ġ= Ġdelete . from ( clazz ) ĊĠĠĠĠĠĠĠ Ġdelete . where ( criteria Builder . equal ( root . get < String >( pk ), Ġvalue )) ĊĠĠĠĠĠĠĠ Ġreturn Ġsession . createQuery ( delete ). executeUpdate () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( SQLException :: class ) ĊĠĠĠ Ġoverride Ġfun Ġget Checkpoint ( id : ĠState Machine Run Id ): ĠCheck point . Serialized ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġget DB Checkpoint ( id )?. to Serialized Checkpoint () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Check points ( statuses : ĠCollection < Flow Status >): ĠStream < Pair < StateMachine Run Id , ĠCheck point . Serialized >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġcriteria Builder Ġ= Ġsession . criteria Builder ĊĠĠĠĠĠĠĠ Ġval Ġcriteria Query Ġ= Ġcriteria Builder . createQuery ( DB Flow Checkpoint :: class . java ) ĊĠĠĠĠĠĠĠ Ġval Ġroot Ġ= Ġcriteria Query . from ( DB Flow Checkpoint :: class . java ) ĊĠĠĠĠĠĠĠ Ġcriteria Query . select ( root ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. where ( criteria Builder . isTrue ( root . get < Flow Status >( DB Flow Checkpoint :: status . name ). ` in `( statuses ))) ĊĠĠĠĠĠĠĠ Ġreturn Ġsession . createQuery ( criteria Query ). stream (). map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Machine Run Id ( UUID . fromString ( it . flow Id )) Ġto Ġit . to Serialized Checkpoint () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Check points To Run (): ĠStream < Pair < StateMachine Run Id , ĠCheck point . Serialized >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġget Check points ( RUN N ABLE _ CHECK POINTS ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Visible ForTesting ĊĠĠĠ Ġinternal Ġfun Ġget DB Checkpoint ( id : ĠState Machine Run Id ): ĠDB Flow Checkpoint ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġcurrent DB Session (). find ( DB Flow Checkpoint :: class . java , Ġid . uuid . toString ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget DB Flow Result ( id : ĠState Machine Run Id ): ĠDB Flow Result ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġcurrent DB Session (). find ( DB Flow Result :: class . java , Ġid . uuid . toString ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget DB Flow Exception ( id : ĠState Machine Run Id ): ĠDB Flow Exception ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġcurrent DB Session (). find ( DB Flow Exception :: class . java , Ġid . uuid . toString ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Paused Check points (): ĠStream < Pair < StateMachine Run Id , ĠCheck point . Serialized >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġjp ql Query Ġ= Ġ""" select Ġnew Ġ${ DB Paused Fields :: class . java . name }( checkpoint . id , Ġblob . checkpoint , Ġcheckpoint . status , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . progress Step , Ġcheckpoint . io RequestType , Ġcheckpoint . compatible ) Ġfrom Ġ${ DB Flow Checkpoint :: class . java . name } ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint Ġjoin Ġ${ DB Flow Checkpoint Blob :: class . java . name } Ġblob Ġon Ġcheckpoint . blob Ġ= Ġblob . id Ġwhere ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . status Ġ= Ġ${ Flow Status . PA USED . ordinal } """. trim Indent () ĊĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġsession . createQuery ( jp ql Query , ĠDB Paused Fields :: class . java ) ĊĠĠĠĠĠĠĠ Ġreturn Ġquery . result List . stream (). map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Machine Run Id ( UUID . fromString ( it . id )) Ġto Ġit . to Serialized Checkpoint () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Finished Flows Results Metadata (): ĠStream < Pair < StateMachine Run Id , ĠFlow Result Metadata >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġjp ql Query Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ""" select Ġnew Ġ${ DB Flow Result Metadata Fields :: class . java . name }( checkpoint . id , Ġcheckpoint . status , Ġmetadata . user Supp lied Identifier ) ĠĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom Ġ${ DB Flow Checkpoint :: class . java . name } Ġcheckpoint ĠĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Ġ${ DB Flow Metadata :: class . java . name } Ġmetadata Ġon Ġmetadata . id Ġ= Ġcheckpoint . flow Metadata ĠĠĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhere Ġcheckpoint . status Ġ= Ġ${ Flow Status . COMPLETED . ordinal } Ġor Ġcheckpoint . status Ġ= Ġ${ Flow Status . FAILED . ordinal } """. trim Indent () ĊĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġsession . createQuery ( jp ql Query , ĠDB Flow Result Metadata Fields :: class . java ) ĊĠĠĠĠĠĠĠ Ġreturn Ġquery . result List . stream (). map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Machine Run Id ( UUID . fromString ( it . id )) Ġto ĠFlow Result Metadata ( it . status , Ġit . clientId ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Flow Result ( id : ĠState Machine Run Id , Ġthrow If Missing : ĠBoolean ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Result Ġ= Ġget DB Flow Result ( id ) ĊĠĠĠĠĠĠĠ Ġif Ġ( throw If Missing Ġ&& Ġdb Flow Result Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Flow 's Ġ$ id Ġresult Ġwas Ġnot Ġfound Ġin Ġthe Ġdatabase . ĠSomething Ġis Ġvery Ġwrong .") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġserialized Flow Result Ġ= Ġdb Flow Result ?. value ?. let Ġ{ ĠSerialized Bytes < Any >( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġserialized Flow Result ?. deserialize ( context Ġ= ĠSerialization Defaults . STORAGE _ CONTEXT ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Flow Exception ( id : ĠState Machine Run Id , Ġthrow If Missing : ĠBoolean ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdb Flow Exception Ġ= Ġget DB Flow Exception ( id ) ĊĠĠĠĠĠĠĠ Ġif Ġ( throw If Missing Ġ&& Ġdb Flow Exception Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Flow 's Ġ$ id Ġexception Ġwas Ġnot Ġfound Ġin Ġthe Ġdatabase . ĠSomething Ġis Ġvery Ġwrong .") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġserialized Flow Exception Ġ= Ġdb Flow Exception ?. value ?. let Ġ{ ĠSerialized Bytes < Any >( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġserialized Flow Exception ?. deserialize ( context Ġ= ĠSerialization Defaults . STORAGE _ CONTEXT ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġremove Flow Exception ( id : ĠState Machine Run Id ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġflow Id Ġ= Ġid . uuid . toString () ĊĠĠĠĠĠĠĠ Ġreturn Ġdelete Row ( DB Flow Exception :: class . java , ĠDB Flow Exception :: flow _ id . name , Ġflow Id ) Ġ== Ġ 1 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġupdate Status ( run Id : ĠState Machine Run Id , Ġflow Status : ĠFlow Status ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= Ġ" Update Ġ${ NODE _ DATABASE _ PREFIX } check points Ġset Ġstatus Ġ= Ġ${ flow Status . ordinal } Ġwhere Ġflow _ id Ġ= Ġ'${ run Id . uuid } '" ĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). create Native Query ( update ). executeUpdate () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġupdate Compatible ( run Id : ĠState Machine Run Id , Ġcompatible : ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= Ġ" Update Ġ${ NODE _ DATABASE _ PREFIX } check points Ġset Ġcompatible Ġ= Ġ$ compatible Ġwhere Ġflow _ id Ġ= Ġ'${ run Id . uuid } '" ĊĠĠĠĠĠĠĠ Ġcurrent DB Session (). create Native Query ( update ). executeUpdate () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate DB Flow Metadata ( flow Id : ĠString , Ġcheckpoint : ĠCheck point , Ġnow : ĠInstant ): ĠDB Flow Metadata Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= Ġcheckpoint . checkpoint State . invocation Context ĊĠĠĠĠĠĠĠ Ġval Ġflow Info Ġ= Ġcheckpoint . checkpoint State . sub Flow Stack . first () ĊĠĠĠĠĠĠĠ Ġreturn ĠDB Flow Metadata ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinvocation Id Ġ= Ġcontext . trace . invocation Id . value , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTr uncate Ġthe Ġflow Ġname Ġto Ġfit Ġinto Ġthe Ġdatabase Ġcolumn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFlow Ġnames Ġare Ġunlikely Ġto Ġbe Ġthis Ġlong ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Name Ġ= Ġflow Info . flow Class . name . take ( MAX _ FLOW _ NAME _ LENGTH ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġuser Supp lied Identifier Ġ= Ġcontext . clientId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstart Type Ġ= Ġcontext . get Started Type (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Parameters Ġ= Ġcontext . get Flow Parameters (). storage Serialize (). bytes , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ing C ord app Ġ= Ġ( flow Info . sub Flow Version Ġas ? ĠSub Flow Version . Cor D app Flow )?. cor D appName Ġ?: Ġ" Core Ġflow ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġplatform Version Ġ= ĠPLATFORM _ VERSION , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstarted By Ġ= Ġcontext . principal (). name , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinvocation Instant Ġ= Ġcontext . trace . invocation Id . timestamp , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstart Instant Ġ= Ġnow , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinish Instant Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate DB Checkpoint Blob ( ĊĠĠĠĠĠĠĠ Ġflow Id : ĠString , ĊĠĠĠĠĠĠĠ Ġserialized Checkpoint State : ĠSerialized Bytes < Checkpoint State >, ĊĠĠĠĠĠĠĠ Ġserialized Flow State : ĠSerialized Bytes < Flow State > ?, ĊĠĠĠĠĠĠĠ Ġnow : ĠInstant ĊĠĠĠ Ġ): ĠDB Flow Checkpoint Blob Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠDB Flow Checkpoint Blob ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint Ġ= Ġserialized Checkpoint State . bytes , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Stack Ġ= Ġserialized Flow State ?. bytes , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhmac Ġ= ĠByteArray ( HMAC _ SIZE _ BYTES ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpersisted Instant Ġ= Ġnow ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate DB Flow Result ( flow Id : ĠString , Ġresult : ĠAny ?, Ġnow : ĠInstant ): ĠDB Flow Result Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠDB Flow Result ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow _ id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġresult ?. storage Serialize ()?. bytes , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpersisted Instant Ġ= Ġnow ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate DB Flow Exception ( flow Id : ĠString , Ġerror State : ĠError State . Er rored , Ġnow : ĠInstant ): ĠDB Flow Exception Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġerror State . errors . last (). exception . let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠDB Flow Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow _ id Ġ= Ġflow Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtype Ġ= Ġit :: class . java . name . truncate ( MAX _ EX C _ TYPE _ LENGTH , Ġtrue ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Ġ= Ġit . message ?. truncate ( MAX _ EX C _ MSG _ LENGTH , Ġfalse ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstack Trace Ġ= Ġit . stack Trace ToString (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġit . storage Serialize (). bytes , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpersisted Instant Ġ= Ġnow ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġset DB Flow Metadata Finish Time ( flow Id : ĠString , Ġnow : ĠInstant ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsession Ġ= Ġcurrent DB Session () ĊĠĠĠĠĠĠĠ Ġval Ġsql Query Ġ= Ġ" Update Ġ${ NODE _ DATABASE _ PREFIX } flow _ metadata Ġset Ġfinish _ time Ġ= Ġ'$ now ' Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" where Ġflow _ id Ġ= Ġ'$ flow Id '" ĊĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġsession . create Native Query ( sql Query ) ĊĠĠĠĠĠĠĠ Ġquery . executeUpdate () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠInvocation Context . get Started Type (): ĠStart Reason Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( origin ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvocation Origin . RPC , Ġis ĠInvocation Origin . Shell Ġ-> ĠStart Reason . RPC ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvocation Origin . Peer Ġ-> ĠStart Reason . INIT I ATED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvocation Origin . Service Ġ-> ĠStart Reason . SERVICE ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvocation Origin . Scheduled Ġ-> ĠStart Reason . SCHEDUL ED ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" Magic Number ") ĊĠĠĠ Ġprivate Ġfun ĠInvocation Context . get Flow Parameters (): ĠList < Any ?> Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠOnly ĠRPC Ġflows Ġhave Ġparameters Ġwhich Ġare Ġfound Ġin Ġindex Ġ 1 Ġor Ġindex Ġ 2 Ġ( if Ġcalled Ġwith Ġclient Ġid ) ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( arguments !!. isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġarguments !!. run Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck ( size Ġ== Ġ 2 Ġ|| Ġsize Ġ== Ġ 3 ) Ġ{ Ġ" Unexpected Ġargument Ġnumber Ġprovided Ġin Ġrpc Ġcall " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunchecked Cast < Any ?, ĠArray < Any ?> >( last ()). toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġempty List () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠDB Flow Checkpoint . to Serialized Checkpoint (): ĠCheck point . Serialized Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġserial ised Flow State Ġ= Ġblob !!. flow Stack ?. let Ġ{ ĠSerialized Bytes < Flow State >( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠCheck point . Serialized ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Checkpoint State Ġ= ĠSerialized Bytes ( blob !!. checkpoint ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Flow State Ġ= Ġserial ised Flow State , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAlways Ġload Ġas Ġa Ġ[ Clean ] Ġcheckpoint Ġto Ġrepresent Ġthat Ġthe Ġcheckpoint Ġis Ġthe Ġlast Ġ_ good _ Ġcheckpoint ĊĠĠĠĠĠĠĠĠĠĠĠ Ġerror State Ġ= ĠError State . Clean , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠA Ġcheckpoint Ġwith Ġa Ġresult Ġshould Ġnot Ġnormally Ġbe Ġloaded Ġ( it Ġshould Ġbe Ġ[ null ] Ġmost Ġof Ġthe Ġtime ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġresult ?. let Ġ{ Ġdb Flow Result Ġ-> Ġdb Flow Result . value ?. let Ġ{ ĠSerialized Bytes < Any >( it ) Ġ} Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstatus Ġ= Ġstatus , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprogress Step Ġ= Ġprogress Step , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Io Request Ġ= Ġio RequestType , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompatible Ġ= Ġcompatible ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠDB Paused Fields ( ĊĠĠĠĠĠĠĠ Ġval Ġid : ĠString , ĊĠĠĠĠĠĠĠ Ġval Ġcheckpoint : ĠByteArray Ġ= ĠEMPTY _ BYTE _ ARRAY , ĊĠĠĠĠĠĠĠ Ġval Ġstatus : ĠFlow Status , ĊĠĠĠĠĠĠĠ Ġval Ġprogress Step : ĠString ?, ĊĠĠĠĠĠĠĠ Ġval Ġio RequestType : ĠString ?, ĊĠĠĠĠĠĠĠ Ġval Ġcompatible : ĠBoolean ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġto Serialized Checkpoint (): ĠCheck point . Serialized Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠCheck point . Serialized ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Checkpoint State Ġ= ĠSerialized Bytes ( checkpoint ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialized Flow State Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAlways Ġload Ġas Ġa Ġ[ Clean ] Ġcheckpoint Ġto Ġrepresent Ġthat Ġthe Ġcheckpoint Ġis Ġthe Ġlast Ġ_ good _ Ġcheckpoint ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror State Ġ= ĠError State . Clean , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstatus Ġ= Ġstatus , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprogress Step Ġ= Ġprogress Step , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Io Request Ġ= Ġio RequestType , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompatible Ġ= Ġcompatible ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠDB Flow Result Metadata Fields ( ĊĠĠĠĠĠĠĠ Ġval Ġid : ĠString , ĊĠĠĠĠĠĠĠ Ġval Ġstatus : ĠFlow Status , ĊĠĠĠĠĠĠĠ Ġval ĠclientId : ĠString ? ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġfun Ġ< T Ġ: ĠAny > ĠT . storage Serialize (): ĠSerialized Bytes < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġserialize ( context Ġ= ĠSerialization Defaults . STORAGE _ CONTEXT ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠCheck point . is Finished () Ġ= Ġwhen Ġ( status ) Ġ{ ĊĠĠĠĠĠĠĠ ĠFlow Status . COMPLETED , ĠFlow Status . K IL LED , ĠFlow Status . FAILED Ġ-> Ġtrue ĊĠĠĠĠĠĠĠ Ġelse Ġ-> Ġfalse ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠString . truncate ( maxLength : ĠInt , Ġwith Warnings : ĠBoolean ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġstr Ġ= Ġthis ĊĠĠĠĠĠĠĠ Ġif Ġ( length Ġ> ĠmaxLength ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( with Warnings ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Trunc ating Ġlong Ġstring Ġbefore Ġstoring Ġit Ġinto Ġthe Ġdatabase . ĠString : Ġ$ str .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstr Ġ= Ġstr . substring ( 0 , ĠmaxLength ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġstr ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠThrowable . stack Trace ToString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġstack Trace Str Ġ= ĠException Utils . get StackTrace ( this ) ĊĠĠĠĠĠĠĠ Ġif Ġ( stack Trace Str . length Ġ> ĠMAX _ STACK TRACE _ LENGTH ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcut Ġoff Ġthe Ġlast Ġline , Ġwhich Ġwill Ġbe Ġa Ġhalf Ġline ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġline Break Ġ= ĠSystem . getProperty (" line . separator ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtruncate Index Ġ= Ġstack Trace Str . lastIndexOf ( line Break , ĠMAX _ STACK TRACE _ LENGTH Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstack Trace Str Ġ= Ġstack Trace Str . substring ( 0 , Ġtruncate Index Ġ+ Ġline Break . length ) Ġ// Ġinclude Ġlast Ġline Ġbreak Ġin ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġstack Trace Str ĊĠĠĠ Ġ} Ċ } Ċ