package Ġcom . r 3 . logic apps . processing Ċ Ċ import Ġcom . r 3 . logic apps . Bus Request Ċ import Ġcom . r 3 . logic apps . Bus Request . Invoke Flow With Input States Ċ import Ġcom . r 3 . logic apps . Bus Request . Invoke Flow Without Input States Ċ import Ġcom . r 3 . logic apps . Bus Request . Query Flow State Ċ import Ġcom . r 3 . logic apps . Bus Response Ċ import Ġcom . r 3 . logic apps . Bus Response . Confirmation Ċ import Ġcom . r 3 . logic apps . Bus Response . Error . Flow Error Ċ import Ġcom . r 3 . logic apps . Bus Response . Flow Output Ċ import Ġcom . r 3 . logic apps . Bus Response . Invocation State Ċ import Ġcom . r 3 . logic apps . Bus Response . State Output Ċ import Ġcom . r 3 . logic apps . Inv oc able Ċ import Ġcom . r 3 . logic apps . service bus . Service bus Client Ċ import Ġio . github . class graph . Class Graph Ċ import Ġnet . cord a . core . contracts . Unique Identifier Ċ import Ġnet . cord a . core . flows . Flow Logic Ċ import Ġnet . cord a . core . identity . C ord a X 5 0 0 Name Ċ import Ġnet . cord a . core . internal . p ooled Scan Ċ import Ġnet . cord a . core . node . services . Identity Service Ċ import Ġnet . cord a . core . utilities . context Logger Ċ import Ġjava . util . UUID Ċ Ċ open Ġclass ĠMessage Processor Impl ( ĊĠĠĠ Ġprivate Ġval Ġstart Flow Delegate : Ġ( Flow Logic <* >, ĠService bus Client , ĠUUID ) Ġ-> ĠFlow Invocation Result , ĊĠĠĠ Ġprivate Ġval Ġretrieve State Delegate : Ġ( Unique Identifier ) Ġ-> ĠState QueryResult , ĊĠĠĠ Ġprivate Ġval Ġidentity Service : ĠIdentity Service ? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġowner : ĠC ord a X 5 0 0 Name Ċ ) Ġ: ĠMessage Processor Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġinvoke ( message : ĠBus Request , Ġclient : ĠService bus Client , Ġmessage Lock Token Id : ĠUUID ): ĠList < Bus Response > Ġ= ĊĠĠĠĠĠĠĠ Ġwhen Ġ( message ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvoke Flow Without Input States Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Invocation Message ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= Ġmessage . requestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinv oc able Ġ= Ġmessage , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis New Ġ= Ġtrue , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Ġ= Ġclient , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Lock Token Id Ġ= Ġmessage Lock Token Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠInvoke Flow With Input States ĠĠĠ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Invocation Message ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= Ġmessage . requestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġmessage . linear Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinv oc able Ġ= Ġmessage , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis New Ġ= Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Ġ= Ġclient , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Lock Token Id Ġ= Ġmessage Lock Token Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠQuery Flow State ĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠMessage Ġcan Ġbe Ġsafely ĠACK d ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Ack nowled ging Ġmessage Ġwith Ġlock Token Id Ġ$ message Lock Token Id ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient . ack nowledge ( message Lock Token Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Query Message ( message . requestId , Ġmessage . linear Id )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġprocess Invocation Message ( ĊĠĠĠĠĠĠĠ ĠrequestId : ĠString , ĊĠĠĠĠĠĠĠ Ġlinear Id : ĠUnique Identifier ?, ĊĠĠĠĠĠĠĠ Ġinv oc able : ĠInv oc able , ĊĠĠĠĠĠĠĠ Ġis New : ĠBoolean , ĊĠĠĠĠĠĠĠ Ġclient : ĠService bus Client , ĊĠĠĠĠĠĠĠ Ġmessage Lock Token Id : ĠUUID ĊĠĠĠ Ġ): ĠList < Bus Response > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġingress Type Ġ= Ġinv oc able :: class ĊĠĠĠĠĠĠĠ Ġvar Ġflow Logic : ĠFlow Logic < *> ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlinear Id Parameter Ġ= Ġlinear Id ?. let Ġ{ Ġmap Of (" linear Id " Ġto Ġlinear Id . toString ()) Ġ} Ġ?: Ġempty Map () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Logic Ġ= Ġderive Flow Logic ( inv oc able . workflow Name , Ġinv oc able . parameters Ġ+ Ġlinear Id Parameter ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCatch Ġexception Ġthrown Ġby Ġflow Ġlogic Ġcreation Ġand ĠACK Ġthe Ġmessage Ġto Ġavoid Ġdouble ĠACK . ĠIn Ġcase Ġof Ġflow Ġinvocation Ġerrors , Ġthey ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcan Ġappear Ġafter Ġthe Ġhandle Ġis Ġcreated , Ġat Ġwhich Ġpoint Ġthe Ġmessage Ġis Ġalready ĠACK d . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Exception Ġduring Ġflow Ġlogic Ġcreation ", Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Ack nowled ging Ġmessage Ġwith Ġlock Token Id Ġ$ message Lock Token Id ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . ack nowledge ( message Lock Token Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠlistOf ( Flow Error ( ingress Type , ĠrequestId , Ġexception , Ġlinear Id )) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġstart Flow Delegate ( flow Logic !! , Ġclient , Ġmessage Lock Token Id ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlid Ġ= Ġresult . linear Id Ġ?: Ġlinear Id Ġ?: Ġerror (" Unable Ġto Ġderive Ġlinear ĠID Ġafter Ġflow Ġinvocation ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtransaction Hash Ġ= Ġresult . hash Ġ?: Ġerror (" Unable Ġto Ġderive Ġtransaction Ġhash Ġafter Ġflow Ġinvocation ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠInvocation State ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparameters Ġ= Ġinv oc able . parameters , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcaller Ġ= Ġowner , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow Class Ġ= Ġflow Logic :: class , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction Hash Ġ= Ġtransaction Hash ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlow Output ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġingress Type Ġ= Ġingress Type , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfields Ġ= Ġresult . fields , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis New Contract Ġ= Ġis New , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom Name Ġ= Ġowner , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Names Ġ= Ġresult . to Names , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction Hash Ġ= Ġtransaction Hash ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfirm ation . Commit ted ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġingress Type Ġ= Ġingress Type ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfirm ation . Submitted ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġingress Type Ġ= Ġingress Type ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Exception Ġduring Ġflow Ġinvocation ", Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( Flow Error ( ingress Type , ĠrequestId , Ġexception , Ġlinear Id )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġprocess Query Message ( requestId : ĠString , Ġlinear Id : ĠUnique Identifier ): ĠBus Response Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġretrieve State Delegate ( linear Id ). let Ġ{ Ġresult Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Output ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlinear Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfields Ġ= Ġresult . fields , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis New Contract Ġ= Ġresult . isNew Contract ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġcatch Ġ( exception : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ ĠFlow Error ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġingress Type Ġ= ĠQuery Flow State :: class , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠrequestId Ġ= ĠrequestId , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcause Ġ= Ġexception , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlinear Id Ġ= Ġlinear Id ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġderive Flow Logic ( flow Name : ĠString , Ġparameters : ĠMap < String , ĠString >): ĠFlow Logic < *> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġflow Classes Ġ= Ġflow Class List ĊĠĠĠĠĠĠĠ Ġval Ġfull Flow Name Ġ= Ġflow Classes . first OrNull Ġ{ Ġit . endsWith (" .$ flow Name ") Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġclazz Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠClass . forName ( full Flow Name Ġ?: Ġflow Name ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ(_ : ĠClass NotFoundException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠClass NotFoundException (" Unable Ġto Ġfind Ġ'$ flow Name ' Ġon Ġthe Ġclass Ġpath ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġflow Logic Ġ= Ġclazz . as Subclass ( Flow Logic :: class . java ) Ġ?: Ġerror ("$ flow Name Ġis Ġnot Ġa Ġsubclass Ġof ĠFlow Logic ") ĊĠĠĠĠĠĠĠ Ġval Ġstatement Ġ= ĠFlow Invoker . get Flow Statement FromString ( flow Logic , Ġparameters , Ġidentity Service ) ĊĠĠĠĠĠĠĠ Ġif Ġ( statement . errors . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException ( statement . errors . join ToString ("; Ġ")) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġctor Ġ= Ġstatement . ctor Ġ?: Ġerror (" Unable Ġto Ġderive Ġapplicable Ġconstructor Ġfor Ġ$ flow Name ") ĊĠĠĠĠĠĠĠ Ġval Ġarguments Ġ= Ġstatement . arguments Ġ?: Ġerror (" Unable Ġto Ġderive Ġarguments Ġfor Ġ$ flow Name ") ĊĠĠĠĠĠĠĠ Ġreturn Ġctor . newInstance (* arguments ) Ġas ? ĠFlow Logic < *> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġerror (" Unable Ġto Ġinstantiate Ġ$ flow Name Ġwith Ġprovided Ġarguments ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġlog Ġ= Ġcontext Logger () ĊĠĠĠĠĠĠĠ Ġval Ġflow Class List : ĠList < String > Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġscan Result Ġ= ĠClass Graph (). enable All Info (). p ooled Scan () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Classes Ġ= Ġscan Result . getSub classes ( Flow Logic :: class . java . name ). map Ġ{ Ġit . name Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Classes ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }