/** Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 1 8 - 2 0 2 0 ĠDr . ĠFl orian ĠSch mid t Ċ Ġ* Ċ Ġ* Ġ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 . dr x . e vo leq . e volution . stubs Ċ Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . channels . Broadcast Channel Ċ import Ġkotlinx . coroutines . delay Ċ import Ġorg . dr x . e vo leq . dsl . E vo leq Dsl Ċ import Ġorg . dr x . e vo leq . e volution . flows . process . Simple Process Flow Ċ import Ġorg . dr x . e vo leq . e vol ving . E vol ving Ċ import Ġorg . e vo leq . math . cat . suspend . morph ism . by Ċ import Ġorg . dr x . e vo leq . e volution . phase . process . Simple Process Phase Ġas ĠPhase ĊĊ Ċ actual Ġabstract Ġclass ĠAction Stub < I , ĠData > Ġactual Ġconstructor ( ĊĠĠĠ Ġprivate Ġval Ġupdate Parent : Ġsuspend Ġ( Update < Data > )-> Unit Ċ ) Ġ: ĠUpdate Stub < Data >( update Parent ) Ġ{ ĊĊĠĠĠ Ġprivate Ġclass ĠStop Exception : ĠException () ĊĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġval Ġinput Broadcast Channel Ġ Ġ= ĠBroadcast Channel < I >( 1 0 _ 0 0 0 ) ĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġval Ġinput Receiver Ġ= Ġinput Broadcast Channel . open Subscription () ĊĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġoverride Ġval Ġflow Ġby Ġlazy { Ġby ( ĊĠĠĠĠĠĠĠ ĠSimple Process Flow ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonStart , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ data Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile ( update Receiver . isEmpty Ġ&& Ġinput Receiver . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! update Receiver . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= Ġby ( update Receiver . receive ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// with ( by ( update Receiver . receive ())) Ġupdate @{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// val Ġupdated Ġ= Ġthis @ update ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġupdated Ġ= Ġupdate ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// console . log (" data Ġ= Ġ$ data ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// console . log (" updated Ġ= Ġ$ updated ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( update Ġis ĠNothing ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġorg . dr x . e vo leq . e volution . phase . process . Simple Process Phase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġif Ġ( updated . data Ġ!= Ġdata ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġorg . dr x . e vo leq . e volution . phase . process . Simple Process Phase . Wait ( on Update ( updated )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġorg . dr x . e vo leq . e volution . phase . process . Simple Process Phase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠStop Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġorg . dr x . e vo leq . e volution . phase . process . Simple Process Phase . Stop ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ//} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ(! input Receiver . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Input ( input Receiver . receive (), Ġdata ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Stop ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ)} ĊĊĊĠĠĠ Ġ// private Ġval Ġupdate Stack Ġby ĠDynamic ArrayList < Update < Data >>( array ListOf ()) ĊĠĠĠ Ġ/* ĊĠĠĠ Ġprivate Ġval Ġupdate Actor Ġ= ĠCor outine Scope ( Job ()). broadcast < Update < Data >>( capacity Ġ= Ġ 1 0 _ 0 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor ( update Ġin Ġchannel ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Stack . add ( update ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ */ ĊĊ Ċ // ĠĠĠ Ġprivate Ġval Ġinput Stack : ĠDynamic ArrayList < I > Ġ= ĠDynamic ArrayList < I >( array ListOf ()) ĊĊĊĊ ĊĊĠĠĠ Ġ/* ĊĠĠĠ Ġprivate Ġval Ġinput Actor Ġ= ĠCor outine Scope ( Job ()). actor < I >( capacity Ġ= Ġ 1 0 _ 0 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor ( input Ġin Ġchannel ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinput Stack . add ( input ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠĠ Ġ*/ Ċ /* ĊĠĠĠ Ġprivate Ġval Ġupdate Stack IsEmpty Ġ= Ġupdate Stack . isEmpty ĊĠĠĠ Ġprivate Ġval Ġinput Stack IsEmpty Ġ= Ġinput Stack . isEmpty ĊĠĠĠ Ġprivate Ġval Ġstacks Are Non Empty Ġ= Ġ!( update Stack IsEmpty Ġand Ġinput Stack IsEmpty ) ĊĊĠĠĠ Ġoverride Ġval Ġflow Ġby Ġlazy { Ġby ( ĊĠĠĠĠĠĠĠ ĠSimple Process Flow ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonStart , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ data Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġblock Until ( stacks Are Non Empty ){ it } ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(!( update Stack IsEmpty . value )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( by ( update Stack . pop ())) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this ( data ). data Ġ!= Ġdata ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Wait ( on Update ( this ( data ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠStop Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Stop ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif (! input Stack IsEmpty . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Input ( input Stack . pop (), Ġdata ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} catch ( exception Ġ: ĠNoSuch ElementException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPhase . Wait ( data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Stop ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ)} Ċ */ ĊĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġoverride Ġval Ġmorph ism : Ġsuspend ĠCor outine Scope .( Data ) Ġ-> ĠE vol ving < Data > Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdata Ġ-> Ġflow ( Phase . Start ( data )) Ġmap Ġ{ Ġit . data Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġabstract Ġval Ġon Input : Ġsuspend ĠCor outine Scope .( I , Data )-> Phase < Data > ĊĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġ@ E vo leq Dsl ĊĠĠĠ Ġactual Ġsuspend Ġfun Ġinput ( input : ĠI ) Ġ{ ĊĠĠĠĠĠĠĠ Ġinput Broadcast Channel . send ( input ) ĊĠĠĠĠĠĠĠ Ġ// input Stack . add ( input ) ĊĠĠĠĠĠĠĠ Ġ// input Actor . send ( input ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġ@ E vo leq Dsl ĊĠĠĠ Ġoverride Ġ Ġfun Ġclose Ports () Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . close Ports () ĊĠĠĠĠĠĠĠ Ġinput Broadcast Channel . close () ĊĠĠĠĠĠĠ Ġ// Ġupdate Actor . close () ĊĠĠĠĠĠ Ġ// Ġ Ġinput Actor . close () ĊĠĠĠ Ġ} Ċ }