/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 ĠSquare ĠInc . Ċ Ġ* Ċ Ġ* Ġ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 Ġcom . squareup . workflow Ċ Ċ import Ġcom . squareup . workflow . Workflow Host . Factory Ċ import Ġcom . squareup . workflow . Workflow Host . Update Ċ import Ġcom . squareup . workflow . internal . Workflow Id Ċ import Ġcom . squareup . workflow . internal . Workflow Node Ċ import Ġcom . squareup . workflow . internal . id Ċ import Ġkotlinx . coroutines . experimental . cancel Ċ import Ġkotlinx . coroutines . experimental . channels . Receive Channel Ċ import Ġkotlinx . coroutines . experimental . channels . produce Ċ import Ġkotlinx . coroutines . experimental . isActive Ċ import Ġkotlinx . coroutines . experimental . select s . select Ċ import Ġorg . jetbrains . annotations . Test Only Ċ import Ġkotlin . coroutines . experimental . Coroutine Context Ċ import Ġkotlin . coroutines . experimental . Empty Coroutine Context Ċ Ċ /** Ċ Ġ* ĠProvides Ġa Ġstream Ġof Ġ[ updates ][ Update ] Ġfrom Ġa Ġtree Ġof Ġ[ Workflow ] s . Ċ Ġ* Ċ Ġ* ĠCreate Ġthese Ġby Ġinject ing Ġa Ġ[ Factory ] Ġand Ġcalling Ġ[ run ][ Factory . run ]. Ċ Ġ*/ Ċ interface ĠWorkflow Host < out ĠOutput T Ġ: ĠAny , Ġout ĠRendering T Ġ: ĠAny > Ġ{ ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠOutput Ġfrom Ġa Ġ[ Workflow Host ]. ĠE mitted Ġfrom Ġ[ Workflow Host . updates ] Ġafter Ġevery ĊĠĠ Ġ* Ġ[ compose ][ Workflow . compose ] Ġpass . ĊĠĠ Ġ*/ ĊĠ Ġdata Ġclass ĠUpdate < out ĠOutput T Ġ: ĠAny , Ġout ĠRendering T Ġ: ĠAny >( ĊĠĠĠ Ġval Ġrendering : ĠRendering T , ĊĠĠĠ Ġval Ġsnapshot : ĠSnapshot , ĊĠĠĠ Ġval Ġoutput : ĠOutput T ? Ġ= Ġnull ĊĠ Ġ) ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠStream Ġof Ġ[ updates ][ Update ] Ġfrom Ġthe Ġroot Ġworkflow . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġis Ġ* not * Ġa Ġbroadcast Ġchannel , Ġso Ġit Ġshould Ġonly Ġbe Ġread Ġby Ġa Ġsingle Ġconsumer . ĊĠĠ Ġ*/ ĊĠ Ġval Ġupdates : ĠReceive Channel < Update < Output T , ĠRendering T >> ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠInject Ġone Ġof Ġthese Ġto Ġstart Ġroot Ġ[ Workflow ] s . ĊĠĠ Ġ*/ ĊĠ Ġclass ĠFactory ( private Ġval Ġbase Context : ĠCor outine Context ) Ġ{ ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCreates Ġa Ġ[ Workflow Host ] Ġto Ġrun Ġ[ workflow ]. ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThe Ġworkflow 's Ġinitial Ġstate Ġis Ġdetermined Ġby Ġpassing Ġ[ input ] Ġto Ġ[ Workflow . initialState ]. ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġworkflow ĠThe Ġworkflow Ġto Ġstart . ĊĠĠĠĠ Ġ* Ġ@ param Ġinput ĠP assed Ġto Ġ[ Workflow . initialState ] Ġto Ġdetermine Ġthe Ġroot Ġworkflow 's Ġinitial Ġstate . ĊĠĠĠĠ Ġ* ĠIf Ġ[ Input T ] Ġis Ġ` Unit `, Ġyou Ġcan Ġjust Ġomit Ġthis Ġargument . ĊĠĠĠĠ Ġ* Ġ@ param Ġsnapshot ĠIf Ġnot Ġnull , Ġused Ġto Ġrestore Ġthe Ġworkflow . ĊĠĠĠĠ Ġ* Ġ@ param Ġcontext ĠThe Ġ[ Coroutine Context ] Ġused Ġto Ġrun Ġthe Ġworkflow Ġtree . ĠAdded Ġto Ġthe Ġ[ Factory ]' s ĊĠĠĠĠ Ġ* Ġcontext . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġ< Input T Ġ: ĠAny , ĠState T Ġ: ĠAny , ĠOutput T Ġ: ĠAny , ĠRendering T Ġ: ĠAny > Ġrun ( ĊĠĠĠĠĠ Ġworkflow : ĠWorkflow < Input T , ĠState T , ĠOutput T , ĠRendering T >, ĊĠĠĠĠĠ Ġinput : ĠInput T , ĊĠĠĠĠĠ Ġsnapshot : ĠSnapshot ? Ġ= Ġnull , ĊĠĠĠĠĠ Ġcontext : ĠCor outine Context Ġ= ĠEmpty Coroutine Context ĊĠĠĠ Ġ): ĠWorkflow Host < Output T , ĠRendering T > Ġ= Ġrun ( workflow . id (), Ġworkflow , Ġinput , Ġsnapshot , Ġcontext ) ĊĊĠĠĠ Ġfun Ġ< S Ġ: ĠAny , ĠO Ġ: ĠAny , ĠR Ġ: ĠAny > Ġrun ( ĊĠĠĠĠĠ Ġworkflow : ĠWorkflow < Unit , ĠS , ĠO , ĠR >, ĊĠĠĠĠĠ Ġsnapshot : ĠSnapshot ? Ġ= Ġnull , ĊĠĠĠĠĠ Ġcontext : ĠCor outine Context Ġ= ĠEmpty Coroutine Context ĊĠĠĠ Ġ): ĠWorkflow Host < O , ĠR > Ġ= Ġrun ( workflow . id (), Ġworkflow , ĠUnit , Ġsnapshot , Ġcontext ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCreates Ġa Ġ[ Workflow Host ] Ġthat Ġruns Ġ[ workflow ] Ġstarting Ġfrom Ġ[ initialState ]. ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ** Don 't Ġcall Ġthis Ġdirectly .** ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠInstead , Ġyour Ġmodule Ġshould Ġhave Ġa Ġtest Ġdependency Ġon Ġ` pure - v 2 - testing ` Ġand Ġyou Ġshould Ġcall Ġthe ĊĠĠĠĠ Ġ* Ġtesting Ġextension Ġmethod Ġdefined Ġthere Ġon Ġyour Ġworkflow Ġitself . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test Only ĊĠĠĠ Ġfun Ġ< Input T Ġ: ĠAny , ĠState T Ġ: ĠAny , ĠOutput T Ġ: ĠAny , ĠRendering T Ġ: ĠAny > Ġrun Test From State ( ĊĠĠĠĠĠ Ġworkflow : ĠWorkflow < Input T , ĠState T , ĠOutput T , ĠRendering T >, ĊĠĠĠĠĠ Ġinput : ĠInput T , ĊĠĠĠĠĠ ĠinitialState : ĠState T ĊĠĠĠ Ġ): ĠWorkflow Host < Output T , ĠRendering T > Ġ{ ĊĠĠĠĠĠ Ġval Ġworkflow Id Ġ= Ġworkflow . id () ĊĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠWorkflow Host < Output T , ĠRendering T > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnode Ġ= ĠWorkflow Node ( workflow Id , Ġworkflow , Ġinput , Ġnull , Ġbase Context , ĠinitialState ) ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġupdates : ĠReceive Channel < Update < Output T , ĠRendering T >> Ġ= Ġnode . start ( workflow , Ġinput ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġ< Input T Ġ: ĠAny , ĠState T Ġ: ĠAny , ĠOutput T Ġ: ĠAny , ĠRendering T Ġ: ĠAny > Ġrun ( ĊĠĠĠĠĠ Ġid : ĠWorkflow Id < Input T , ĠState T , ĠOutput T , ĠRendering T >, ĊĠĠĠĠĠ Ġworkflow : ĠWorkflow < Input T , ĠState T , ĠOutput T , ĠRendering T >, ĊĠĠĠĠĠ Ġinput : ĠInput T , ĊĠĠĠĠĠ Ġsnapshot : ĠSnapshot ?, ĊĠĠĠĠĠ Ġcontext : ĠCor outine Context ĊĠĠĠ Ġ): ĠWorkflow Host < Output T , ĠRendering T > Ġ= Ġobject Ġ: ĠWorkflow Host < Output T , ĠRendering T > Ġ{ ĊĠĠĠĠĠ Ġval Ġnode Ġ= ĠWorkflow Node ( id , Ġworkflow , Ġinput , Ġsnapshot , Ġbase Context Ġ+ Ġcontext ) ĊĠĠĠĠĠ Ġoverride Ġval Ġupdates : ĠReceive Channel < Update < Output T , ĠRendering T >> Ġ= Ġnode . start ( workflow , Ġinput ) ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠStarts Ġthe Ġcoroutine Ġthat Ġruns Ġthe Ġcoroutine Ġloop . Ċ Ġ*/ Ċ internal Ġfun Ġ< I Ġ: ĠAny , ĠS Ġ: ĠAny , ĠO Ġ: ĠAny , ĠR Ġ: ĠAny > ĠWorkflow Node < I , ĠS , ĠO , ĠR >. start ( ĊĠ Ġworkflow : ĠWorkflow < I , ĠS , ĠO , ĠR >, ĊĠ Ġinput : ĠI Ċ ): ĠReceive Channel < Update < O , ĠR >> Ġ= Ġproduce ( capacity Ġ= Ġ 0 ) Ġ{ ĊĠ Ġtry Ġ{ ĊĠĠĠ Ġvar Ġoutput : ĠO ? Ġ= Ġnull ĊĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠ Ġval Ġrendering Ġ= Ġcompose ( workflow , Ġinput ) ĊĠĠĠĠĠ Ġval Ġsnapshot Ġ= Ġsnapshot ( workflow ) ĊĠĠĠĠĠ Ġsend ( Update ( rendering , Ġsnapshot , Ġoutput )) ĊĠĠĠĠĠ Ġ// ĠTick Ġ_ m ight _ Ġreturn Ġan Ġoutput , Ġbut Ġif Ġit Ġreturns Ġnull , Ġit Ġmeans Ġthe Ġstate Ġor Ġa Ġchild ĊĠĠĠĠĠ Ġ// Ġprobably Ġchanged , Ġso Ġwe Ġshould Ġre - compose / snapshot Ġand Ġemit Ġagain . ĊĠĠĠĠĠ Ġoutput Ġ= Ġselect Ġ{ ĊĠĠĠĠĠĠĠ Ġtick ( this ) Ġ{ Ġit Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ġcatch Ġ( e : ĠThrowable ) Ġ{ ĊĠĠĠ Ġ// ĠFor Ġsome Ġreason Ġthe Ġexception Ġgets Ġmasked Ġif Ġwe Ġdon 't Ġexplicitly Ġpass Ġit Ġto Ġcancel Ġthe ĊĠĠĠ Ġ// Ġproducer Ġcoroutine Ġourselves Ġhere . ĊĠĠĠ Ġcoroutine Context . cancel ( e ) ĊĠĠĠ Ġthrow Ġe ĊĠ Ġ} Ġfinally Ġ{ ĊĠĠĠ Ġ// ĠThere 's Ġa Ġpotential Ġrace Ġcondition Ġif Ġthe Ġproducer Ġcoroutine Ġis Ġcancelled Ġbefore Ġit Ġhas Ġa Ġchance ĊĠĠĠ Ġ// Ġto Ġenter Ġthe Ġtry Ġblock , Ġsince Ġwe Ġcan 't Ġuse ĠCor outine Start . ATOMIC . ĠHowever , Ġuntil Ġwe Ġactually ĊĠĠĠ Ġ// Ġsee Ġthis Ġcause Ġproblems , ĠI 'm Ġnot Ġtoo Ġwor ried Ġabout Ġit . ĊĠĠĠ Ġcancel () ĊĠ Ġ} Ċ } Ċ