package Ġarrow . fx . coroutines Ċ Ċ import Ġarrow . core . Tuple 2 Ċ import Ġarrow . core . to T Ċ import Ġarrow . fx . coroutines . Default Concurrent Var . Companion . State . WaitFor Put Ċ import Ġarrow . fx . coroutines . Default Concurrent Var . Companion . State . WaitFor Take Ċ import Ġkotlinx . atomic fu . atomic Ċ Ċ /** Ċ Ġ* Ġ[ Concurrent Var ] Ġis Ġa Ġmutable Ġconcurrent Ġsafe Ġvariable Ġwhich Ġis Ġeither Ġ` empty ` Ġor Ġcontains Ġa Ġ` single Ġvalue ` Ġof Ġtype Ġ[ A ]. Ċ Ġ* ĠIt Ġbehaves Ġas Ġa Ġsingle Ġelement Ġ[ arrow . fx . coroutines . stream . concurrent . Queue ]. Ċ Ġ* ĠWhen Ġtrying Ġto Ġ[ put ] Ġor Ġ[ take ], Ġit Ġwill Ġsuspend Ġwhen Ġit Ġis Ġrespectively Ġ[ isEmpty ] Ġor Ġ[ isNotEmpty ]. Ċ Ġ* Ċ Ġ* ĠThere Ġare Ġalso Ġoperators Ġthat Ġreturn Ġimmediately , Ġ[ try Take ] Ġ& Ġ[ try Put ], Ċ Ġ* Ġsince Ġchecking Ġ[ isEmpty ] Ġcould Ġbe Ġoutdated Ġimmediately . Ċ Ġ* Ċ Ġ* Ġ[ Concurrent Var ] Ġis Ġappropriate Ġfor Ġbuilding Ġsynchronization Ġprimitives Ġand Ġperforming Ġsimple Ġinter - thread Ġcommunications . Ċ Ġ* Ġi . e . Ġin Ġsituations Ġwhere Ġyou Ġwant Ġto Ġ` suspend ` Ġuntil Ġthe Ġ[ Concurrent Var ] Ġis Ġinitial ised Ġwith Ġa Ġvalue Ġ[ A ]. Ċ Ġ* Ċ Ġ* Ġ``` kotlin : ank : playground Ċ Ġ* Ġimport Ġarrow . fx . coroutines .* Ċ Ġ* Ċ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ Ċ Ġ* ĠĠ Ġval Ġm var Ġ= ĠConcurrent Var . empty < Int >() Ċ Ġ* Ċ Ġ* ĠĠ ĠFork Connected Ġ{ Ċ Ġ* ĠĠĠĠ Ġsleep ( 3 . seconds ) Ċ Ġ* ĠĠĠĠ Ġm var . put ( 5 ) Ċ Ġ* ĠĠ Ġ} Ċ Ġ* Ċ Ġ* Ġ Ġval Ġr Ġ= Ġm var . take () Ġ// Ġsuspend Ġuntil ĠFork Ġputs Ġresult Ġin ĠConcurrent Var Ċ Ġ* Ġ Ġprintln ( r ) Ċ Ġ* Ġ} Ċ Ġ* Ġ``` Ċ Ġ* Ċ Ġ* Ġ## ĠUsing Ġ[ Concurrent Var ] Ġas Ġa Ġlock Ġsafely Ċ Ġ* Ċ Ġ* Ġ[ Concurrent Var ] Ġcan Ġalso Ġbe Ġused Ġas Ġa Ġlock Ġif Ġevery Ġoperation Ġcalls Ġ[ take ], Ġdoes Ġwork Ġand Ġthen Ġ[ put ]' s Ġthe Ġvalue Ġback . Ċ Ġ* ĠHowever Ġthis Ġis Ġquite Ġunsafe Ġif Ġoperations Ġcan Ġbe Ġcancelled Ġor Ġcan Ġthrow Ġexception Ġwhile Ġthey Ġhold Ġa Ġlock . Ċ Ġ* ĠThe Ġbest Ġapproach Ġto Ġover come Ġthis Ġis Ġto Ġuse Ġ[ bracket Case ] Ġhowever Ġsince Ġthis Ġis Ġa Ġrather Ġcommon Ġpattern , Ġit Ġis Ġmade Ġavailable Ġwith Ġ[ with Concurrent Var ], Ġ[ modify ] Ġand Ġ[ modify _ ]. Ċ Ġ* Ċ Ġ* Ġ> ĠNote Ġthat Ġthis Ġonly Ġworks Ġif Ġall Ġoperations Ġover Ġthe Ġ[ Concurrent Var ] Ġfollow Ġthe Ġpattern Ġof Ġfirst Ġtaking Ġand Ġthen Ġputting Ġback Ġboth Ġexactly Ġonce Ġand Ġin Ġorder . Ċ Ġ* Ġ ĠOr Ġuse Ġthe Ġhelpers Ġto Ġalso Ġbe Ġsafe Ġin Ġcase Ġof Ġexceptions Ġand Ġcancellation . Ċ Ġ*/ Ċ @ Deprecated (" Concurrent Var Ġis Ġdeprecated Ġin Ġfavor Ġof Ġkotlinx . coroutines . channels . Channel ( 1 ). ") Ċ interface ĠConcurrent Var < A > Ġ{ ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠReturns Ġtrue Ġif Ġthere Ġare Ġno Ġelements . ĠOtherwise Ġfalse . ĊĠĠ Ġ* ĠThis Ġmay Ġbe Ġoutdated Ġimmediately ; Ġuse Ġ[ try Put ] Ġor Ġ[ try Take ] Ġto Ġ[ put ] Ġ& Ġ[ take ] Ġwithout Ġsus pending . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġempty Ġ= ĠConcurrent Var . empty < Int >(). isEmpty () ĊĠĠ Ġ* ĠĠ Ġval Ġfull Ġ= ĠConcurrent Var ( 1 0 ). isEmpty () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" empty : Ġ$ empty , Ġfull : Ġ$ full ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ* ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun ĠisEmpty (): ĠBoolean ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠReturns Ġtrue Ġif Ġthere Ġno Ġelements . ĠOtherwise Ġfalse . ĊĠĠ Ġ* ĠThis Ġmay Ġbe Ġoutdated Ġimmediately ; Ġuse Ġ[ try Put ] Ġor Ġ[ try Take ] Ġto Ġ[ put ] Ġ& Ġ[ take ] Ġwithout Ġsus pending . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġempty Ġ= ĠConcurrent Var . empty < Int >(). isNotEmpty () ĊĠĠ Ġ* ĠĠ Ġval Ġfull Ġ= ĠConcurrent Var ( 1 0 ). isNotEmpty () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" empty : Ġ$ empty , Ġfull : Ġ$ full ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġis NotEmpty (): ĠBoolean ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠPut s Ġ[ A ] Ġin Ġthe Ġ[ Concurrent Var ] Ġif Ġit Ġis Ġempty , Ġor Ġsus p ends Ġif Ġfull Ġuntil Ġthe Ġgiven Ġvalue Ġis Ġnext Ġin Ġline ĊĠĠ Ġ* Ġto Ġbe Ġconsumed Ġby Ġ[ take ]. ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġm var Ġ= ĠConcurrent Var . empty < Int >() ĊĠĠ Ġ* ĠĠ Ġm var . put ( 5 ) ĊĠĠ Ġ* ĠĠ Ġval Ġnone Ġ= Ġtime Out OrNull ( 1 . seconds ) Ġ{ ĊĠĠ Ġ* ĠĠĠĠ Ġm var . put ( 1 0 ) ĊĠĠ Ġ* ĠĠ Ġ} ĊĠĠ Ġ* ĠĠ Ġval Ġres Ġ= Ġm var . take () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" none : Ġ$ none , Ġres : Ġ$ res ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ see Ġ[ try Put ] Ġfor Ġan Ġoperator Ġthat Ġdoesn 't Ġsuspend Ġif Ġthe Ġ[ Concurrent Var ] Ġis Ġfilled . ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġput ( a : ĠA ): ĠUnit ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠT ries Ġto Ġput Ġ[ A ] Ġin Ġthe Ġ[ Concurrent Var ] Ġif Ġit Ġis Ġempty , ĊĠĠ Ġ* Ġreturns Ġimmediately Ġwith Ġtrue Ġif Ġsuccessfully Ġput Ġthe Ġvalue Ġin Ġthe Ġ[ Concurrent Var ] Ġor Ġfalse Ġotherwise . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġm var Ġ= ĠConcurrent Var . empty < Int >() ĊĠĠ Ġ* ĠĠ Ġval Ġsucceed Ġ= Ġm var . try Put ( 5 ) ĊĠĠ Ġ* ĠĠ Ġval Ġfailed Ġ= Ġm var . try Put ( 1 0 ) ĊĠĠ Ġ* ĠĠ Ġval Ġres Ġ= Ġm var . take () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" suc ceed : Ġ$ suc ceed , Ġfailed : Ġ$ failed , Ġres : Ġ$ res ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ see Ġ[ try Put ] Ġfor Ġan Ġoperator Ġthat Ġdoesn 't Ġsuspend Ġif Ġthe Ġ[ Concurrent Var ] Ġis Ġfilled . ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġtry Put ( a : ĠA ): ĠBoolean ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠE mpt ies Ġthe Ġ[ Concurrent Var ] Ġif Ġfull , Ġreturning Ġthe Ġvalue , ĊĠĠ Ġ* Ġor Ġsuspend Ġuntil Ġa Ġvalue Ġis Ġavailable . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* Ġ// sample Start ĊĠĠ Ġ* Ġval Ġm var Ġ= ĠConcurrent Var ( 5 ) ĊĠĠ Ġ* Ġval Ġfive Ġ= Ġm var . take () ĊĠĠ Ġ* Ġval Ġnone Ġ= Ġtime Out OrNull ( 1 . seconds ) Ġ{ ĊĠĠ Ġ* ĠĠ Ġm var . take () ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* Ġ// sample End ĊĠĠ Ġ* Ġprintln (" five : Ġ$ five , Ġnone : Ġ$ none ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġtake (): ĠA ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠT ries Ġto Ġtake Ġthe Ġvalue Ġof Ġ[ Concurrent Var ], Ġreturns Ġa Ġvalue Ġimmediately Ġif Ġthe Ġ[ Concurrent Var ] Ġis Ġnot ĊĠĠ Ġ* Ġempty , Ġor Ġnull Ġotherwise . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġm var Ġ= ĠConcurrent Var ( 5 ) ĊĠĠ Ġ* ĠĠ Ġval Ġvalue Ġ= Ġm var . try Take () ĊĠĠ Ġ* ĠĠ Ġval Ġempty Ġ= Ġm var . try Take () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" value : Ġ$ value , Ġempty : Ġ$ empty ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġtry Take (): ĠA ? ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠReads Ġthe Ġcurrent Ġvalue Ġwithout Ġempty ing Ġthe Ġ[ Concurrent Var ], Ġassuming Ġthere Ġis Ġone , Ġor Ġotherwise ĊĠĠ Ġ* Ġit Ġsus p ends Ġuntil Ġthere Ġis Ġa Ġvalue Ġavailable . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ``` kotlin : ank : playground ĊĠĠ Ġ* Ġimport Ġarrow . fx . coroutines .* ĊĠĠ Ġ* ĊĠĠ Ġ* Ġsuspend Ġfun Ġmain (): ĠUnit Ġ{ ĊĠĠ Ġ* ĠĠ Ġ// sample Start ĊĠĠ Ġ* ĠĠ Ġval Ġm var Ġ= ĠConcurrent Var . empty < Int >() ĊĠĠ Ġ* ĊĠĠ Ġ* ĠĠ Ġval Ġnone Ġ= Ġtime Out OrNull ( 1 . seconds ) Ġ{ ĊĠĠ Ġ* ĠĠĠĠ Ġm var . read () ĊĠĠ Ġ* ĠĠ Ġ} ĊĠĠ Ġ* ĠĠ Ġm var . put ( 1 0 ) ĊĠĠ Ġ* ĠĠ Ġval Ġread 1 Ġ= Ġm var . read () ĊĠĠ Ġ* ĠĠ Ġval Ġread 2 Ġ= m var . read () ĊĠĠ Ġ* ĠĠ Ġ// sample End ĊĠĠ Ġ* ĠĠ Ġprintln (" none : Ġ$ none , Ġread 1 : Ġ$ read 1 , Ġread 2 : Ġ$ read 2 ") ĊĠĠ Ġ* Ġ} ĊĠĠ Ġ* ``` ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġread (): ĠA ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠException - Ġand ĠC ancellation - safe Ġwrapper Ġfor Ġoperating Ġon Ġthe Ġcontents Ġof Ġa Ġ[ Concurrent Var ]. ĊĠĠ Ġ* ĊĠĠ Ġ* ĠShould Ġan Ġexception Ġoccur Ġduring Ġ[ f ]' s Ġexecution , Ġor Ġif Ġit Ġis Ġcancelled , Ġthe Ġvalue Ġwill Ġalways Ġbe Ġput Ġback . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġoperation Ġis Ġonly Ġatomic Ġif Ġthere Ġare Ġno Ġother Ġpro ducers Ġfor Ġthis Ġ[ Concurrent Var ]. ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġ< B > Ġwith Concurrent Var ( f : Ġsuspend Ġ( A ) Ġ-> ĠB ): ĠB ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠException - Ġand ĠC ancellation - safe Ġwrapper Ġfor Ġmodifying Ġthe Ġcontents Ġof Ġa Ġ[ Concurrent Var ]. ĊĠĠ Ġ* ĊĠĠ Ġ* ĠShould Ġan Ġexception Ġoccur Ġduring Ġ[ f ]' s Ġexecution , Ġor Ġif Ġit Ġis Ġcancelled , Ġthe Ġinitial Ġvalue Ġwill Ġbe Ġput Ġback . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġoperation Ġis Ġonly Ġatomic Ġif Ġthere Ġare Ġno Ġother Ġpro ducers Ġfor Ġthis Ġ[ Concurrent Var ]. ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ see Ġ[ modify _] ĠA Ġversion Ġthat Ġreturns Ġunit Ġand Ġdoes Ġnot Ġexpect Ġa ĠTuple ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġ< B > Ġmodify ( f : Ġsuspend Ġ( A ) Ġ-> ĠTuple 2 < A , ĠB >): ĠB ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠException - Ġand ĠC ancellation - safe Ġwrapper Ġfor Ġmodifying Ġthe Ġcontents Ġof Ġa Ġ[ Concurrent Var ]. ĊĠĠ Ġ* ĊĠĠ Ġ* ĠShould Ġan Ġexception Ġoccur Ġduring Ġ[ f ]' s Ġexecution , Ġor Ġif Ġit Ġis Ġcancelled , Ġthe Ġinitial Ġvalue Ġwill Ġbe Ġput Ġback . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġoperation Ġis Ġonly Ġatomic Ġif Ġthere Ġare Ġno Ġother Ġpro ducers Ġfor Ġthis Ġ[ Concurrent Var ]. ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ see Ġ[ modify ] ĠA Ġversion Ġthat Ġallows Ġa Ġcustom Ġreturn Ġvalue Ġinstead Ġof Ġunit . ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġmodify _( f : Ġsuspend Ġ( A ) Ġ-> ĠA ): ĠUnit Ġ= Ġmodify Ġ{ Ġf ( it ) Ġto T ĠUnit Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġ/** ĠBuilds Ġa Ġ[ Concurrent Var ] Ġinstance Ġwith Ġan Ġ[ initial ] Ġvalue . Ġ*/ ĊĠĠĠ Ġsuspend Ġoperator Ġfun Ġ< A > Ġinvoke ( initial : ĠA ): ĠConcurrent Var < A > Ġ= ĊĠĠĠĠĠ ĠDefault Concurrent Var ( Default Concurrent Var . Companion . State ( initial )) ĊĊĠĠĠ Ġ/** ĠReturns Ġan Ġempty Ġ[ Concurrent Var ] Ġinstance . Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġ< A > Ġempty (): ĠConcurrent Var < A > Ġ= ĊĠĠĠĠĠ ĠDefault Concurrent Var ( Default Concurrent Var . Companion . State . empty ()) ĊĊĠĠĠ Ġfun Ġ< A > Ġunsafe ( initial : ĠA ): ĠConcurrent Var < A > Ġ= ĊĠĠĠĠĠ ĠDefault Concurrent Var ( Default Concurrent Var . Companion . State ( initial )) ĊĊĠĠĠ Ġfun Ġ< A > Ġunsafe Empty (): ĠConcurrent Var < A > Ġ= ĊĠĠĠĠĠ ĠDefault Concurrent Var ( Default Concurrent Var . Companion . State . empty ()) ĊĠ Ġ} Ċ } Ċ Ċ internal Ġsealed Ġclass ĠMaybe < out ĠA > Ġ{ ĊĠ Ġobject ĠNone Ġ: ĠMaybe < Nothing >() ĊĠ Ġdata Ġclass ĠJust < A >( val Ġvalue : ĠA ) Ġ: ĠMaybe < A >() Ċ } Ċ Ċ internal Ġfun Ġ< A > ĠMaybe < A >. orElse ( f : Ġ() Ġ-> ĠMaybe < A >): ĠMaybe < A > Ġ= ĊĠ Ġwhen Ġ( this ) Ġ{ ĊĠĠĠ ĠMaybe . None Ġ-> Ġf () ĊĠĠĠ Ġis ĠMaybe . Just Ġ-> Ġthis ĊĠ Ġ} Ċ Ċ internal Ġfun Ġ< A > ĠMaybe < A >. getOrElse ( f : Ġ() Ġ-> ĠA ): ĠA Ġ= ĊĠ Ġwhen Ġ( this ) Ġ{ ĊĠĠĠ ĠMaybe . None Ġ-> Ġf () ĊĠĠĠ Ġis ĠMaybe . Just Ġ-> Ġthis . value ĊĠ Ġ} Ċ Ċ internal Ġfun Ġ< A > ĠA ?. to Maybe (): ĠMaybe < A > Ġ= ĊĠ Ġif Ġ( this Ġ== Ġnull ) ĠMaybe . None Ġelse ĠMaybe . Just ( this ) Ġas ĠMaybe < A > Ċ Ċ private Ġclass ĠDefault Concurrent Var < A > Ġconstructor ( initial : ĠState < A >) Ġ: ĠConcurrent Var < A > Ġ{ ĊĊĠ Ġprivate Ġval Ġstate Ġ= Ġatomic ( initial ) ĊĊĠ Ġoverride Ġsuspend Ġfun ĠisEmpty (): ĠBoolean Ġ= ĊĠĠĠ Ġwhen Ġ( state . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġtrue ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġfalse ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġis NotEmpty (): ĠBoolean Ġ= ĊĠĠĠ Ġwhen Ġ( state . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġfalse ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġtrue ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġput ( a : ĠA ): ĠUnit Ġ= ĊĠĠĠ Ġif Ġ( try Put ( a )) ĠUnit ĊĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠ Ġcancel Boundary () ĊĠĠĠĠĠ Ġcancell able F Ġ{ Ġcb Ġ-> Ġunsafe Put ( a , Ġcb ) Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġtry Put ( a : ĠA ): ĠBoolean Ġ= ĊĠĠĠ Ġunsafe Try Put ( a ) ĊĊĠ Ġoverride Ġsuspend Ġfun Ġtake (): ĠA Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġm Ġ= Ġunsafe Try Take ()) Ġ{ ĊĠĠĠĠĠ ĠMaybe . None Ġ-> Ġcancell able F Ġ{ Ġunsafe Take ( it ) Ġ} ĊĠĠĠĠĠ Ġis ĠMaybe . Just Ġ-> Ġm . value ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġtry Take (): ĠA ? Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġm Ġ= Ġunsafe Try Take ()) Ġ{ ĊĠĠĠĠĠ ĠMaybe . None Ġ-> Ġnull ĊĠĠĠĠĠ Ġis ĠMaybe . Just Ġ-> Ġm . value ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġread (): ĠA Ġ= ĊĠĠĠ Ġcancell able (:: unsafe Read ) ĊĊĠ Ġoverride Ġsuspend Ġfun Ġ< B > Ġwith Concurrent Var ( f : Ġsuspend Ġ( A ) Ġ-> ĠB ): ĠB Ġ= ĊĠĠĠ Ġbracket Case ( ĊĠĠĠĠĠ Ġacquire Ġ= Ġ:: take , ĊĠĠĠĠĠ Ġuse Ġ= Ġf , ĊĠĠĠĠĠ Ġrelease Ġ= Ġ{ Ġa , Ġ_ Ġ-> Ġput ( a ) Ġ} ĊĠĠĠ Ġ) ĊĊĠ Ġoverride Ġsuspend Ġfun Ġ< B > Ġmodify ( f : Ġsuspend Ġ( A ) Ġ-> ĠTuple 2 < A , ĠB >): ĠB Ġ{ ĊĠĠĠ Ġ// Ġugly . ĠIs Ġthere Ġa Ġbetter Ġway ? ĊĠĠĠ Ġvar Ġres : ĠA ? Ġ= Ġnull ĊĠĠĠ Ġreturn Ġbracket Case ( ĊĠĠĠĠĠ Ġacquire Ġ= Ġ:: take , ĊĠĠĠĠĠ Ġuse Ġ= Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġ( a , Ġb ) Ġ= Ġf ( it ) ĊĠĠĠĠĠĠĠ Ġres Ġ= Ġa ĊĠĠĠĠĠĠĠ Ġb ĊĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠ Ġrelease Ġ= Ġ{ Ġa , Ġexit Ġ-> ĊĠĠĠĠĠĠĠ Ġwhen Ġ( exit ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġis ĠExit Case . Failure Ġ-> Ġput ( a ) ĊĠĠĠĠĠĠĠĠĠ Ġis ĠExit Case . Cancelled Ġ-> Ġput ( a ) ĊĠĠĠĠĠĠĠĠĠ Ġis ĠExit Case . Completed Ġ-> Ġput ( res !! ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ) ĊĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġunsafe Try Put ( a : ĠA ): ĠBoolean Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġfalse ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġfirst : ĠListener < A >? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġupdate : ĠState < A > Ġ= Ġif Ġ( current . t akes . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ ĠState ( a ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġfirst Ġ= Ġcurrent . t akes . values . first () ĊĠĠĠĠĠĠĠĠĠ Ġval Ġrest Ġ= Ġcurrent . t akes . entries . drop ( 1 ) ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( rest . isEmpty ()) ĠState . empty () ĊĠĠĠĠĠĠĠĠĠ Ġelse ĠWaitFor Put ( empty Map (), Ġrest . toMap ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! state . compare AndSet ( current , Ġupdate )) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġunsafe Try Put ( a ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( first Ġ!= Ġnull Ġ|| Ġcurrent . reads . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġcall Put And All Read ers ( a , Ġfirst , Ġcurrent . reads ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġtrue ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġunsafe Put ( a : ĠA , Ġon Put : ĠListener < Unit >): ĠCancel Token Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= ĠToken () ĊĠĠĠĠĠĠĠ Ġval Ġnew Map Ġ= Ġcurrent . listeners Ġ+ ĠPair ( id , ĠPair ( a , Ġon Put )) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠWaitFor Take ( current . value , Ġnew Map ))) ĠCancel Token Ġ{ Ġunsafe Cancel Put ( id ) Ġ} ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Put ( a , Ġon Put ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġfirst : ĠListener < A >? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= Ġif Ġ( current . t akes . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ ĠState ( a ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġfirst Ġ= Ġcurrent . t akes . values . first () ĊĠĠĠĠĠĠĠĠĠ Ġval Ġrest Ġ= Ġcurrent . t akes . entries . drop ( 1 ) ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( rest . isEmpty ()) ĠState . empty () ĊĠĠĠĠĠĠĠĠĠ Ġelse ĠWaitFor Put ( empty Map (), Ġrest . toMap ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , Ġupdate )) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( first Ġ!= Ġnull Ġ|| Ġcurrent . reads . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcall Put And All Read ers ( a , Ġfirst , Ġcurrent . reads ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Put ( Result . success ( Unit )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCancel Token . unit ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Put ( Result . success ( Unit )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCancel Token . unit ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġunsafe Put ( a , Ġon Put ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġfun Ġunsafe Cancel Put ( id : ĠToken ): ĠUnit Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= Ġcurrent . copy ( listeners Ġ= Ġcurrent . listeners Ġ- Ġid ) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , Ġupdate )) ĠUnit ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Cancel Put ( id ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> ĠUnit ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġunsafe Try Take (): ĠMaybe < A > Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( current . listeners . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠState . empty ())) ĠMaybe . Just ( current . value ) ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġunsafe Try Take () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġ( ax , Ġnotify ) Ġ= Ġcurrent . listeners . values . first () ĊĠĠĠĠĠĠĠĠĠ Ġval Ġxs Ġ= Ġcurrent . listeners . entries . drop ( 1 ) ĊĠĠĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= ĠWaitFor Take ( ax , Ġxs . toMap ()) ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , Ġupdate )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnotify ( Result . success ( Unit )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMaybe . Just ( current . value ) ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Try Take () ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> ĠMaybe . None ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġsuspend Ġfun Ġunsafe Take ( on Take : ĠListener < A >): ĠCancel Token Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( current . listeners . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠState . empty ())) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Take ( Result . success ( current . value )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCancel Token . unit ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġunsafe Take ( on Take ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġ( ax , Ġnotify ) Ġ= Ġcurrent . listeners . values . first () ĊĠĠĠĠĠĠĠĠĠ Ġval Ġxs Ġ= Ġcurrent . listeners . entries . drop ( 1 ) ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠWaitFor Take ( ax , Ġxs . toMap ()))) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnotify ( Result . success ( Unit )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Take ( Result . success ( current . value )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCancel Token . unit ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġunsafe Take ( on Take ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= ĠToken () ĊĠĠĠĠĠĠĠ Ġval Ġnew Queue Ġ= Ġcurrent . t akes Ġ+ ĠPair ( id , Ġon Take ) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠWaitFor Put ( current . reads , Ġnew Queue ))) ĠCancel Token Ġ{ Ġunsafe Cancel Take ( id ) Ġ} ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Take ( on Take ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġfun Ġunsafe Cancel Take ( id : ĠToken ): ĠUnit Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnew Map Ġ= Ġcurrent . t akes Ġ- Ġid ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= ĠWaitFor Put ( current . reads , Ġnew Map ) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , Ġupdate )) ĠUnit ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Cancel Take ( id ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> ĠUnit ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġfun Ġunsafe Read ( on Read : ĠListener < A >): ĠCancel Token Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġon Read ( Result . success ( current . value )) ĊĠĠĠĠĠĠĠ ĠCancel Token . unit ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= ĠToken () ĊĠĠĠĠĠĠĠ Ġval Ġnew Reads Ġ= Ġcurrent . reads Ġ+ ĠPair ( id , Ġon Read ) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , ĠWaitFor Put ( new Reads , Ġcurrent . t akes ))) ĠCancel Token Ġ{ Ġunsafe Cancel Read ( id ) Ġ} ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Read ( on Read ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġtail rec Ġfun Ġunsafe Cancel Read ( id : ĠToken ): ĠUnit Ġ= ĊĠĠĠ Ġwhen Ġ( val Ġcurrent Ġ= Ġstate . value ) Ġ{ ĊĠĠĠĠĠ Ġis ĠWaitFor Put Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnew Map Ġ= Ġcurrent . reads Ġ- Ġid ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= ĠWaitFor Put ( new Map , Ġcurrent . t akes ) ĊĠĠĠĠĠĠĠ Ġif Ġ( state . compare AndSet ( current , Ġupdate )) ĠUnit ĊĠĠĠĠĠĠĠ Ġelse Ġunsafe Cancel Read ( id ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠWaitFor Take Ġ-> ĠUnit ĊĠĠĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġcall Put And All Read ers ( ĊĠĠĠ Ġa : ĠA , ĊĠĠĠ Ġput : ĠListener < A > ?, ĊĠĠĠ Ġreads : ĠMap < Token , ĠListener < A >> ĊĠ Ġ): ĠBoolean Ġ{ ĊĠĠĠ Ġval Ġvalue Ġ= ĠResult . success ( a ) ĊĠĠĠ Ġreads . values . call All ( value ) ĊĊĠĠĠ Ġreturn Ġif Ġ( put Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠ Ġput ( value ) ĊĠĠĠĠĠ Ġtrue ĊĠĠĠ Ġ} Ġelse Ġtrue ĊĠ Ġ} ĊĊĠ Ġ// ĠFor Ġstreaming Ġa Ġvalue Ġto Ġa Ġwhole Ġ` reads ` Ġcollection ĊĠ Ġprivate Ġsuspend Ġfun ĠIterable < Listener < A >> . call All ( value : ĠResult < A >): ĠUnit Ġ= ĊĠĠĠ ĠforEach Ġ{ Ġcb Ġ-> Ġcb ( value ) Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġinternal Ġsealed Ġclass ĠState < out ĠA > Ġ{ ĊĠĠĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġref Ġ= ĠWaitFor Put < Any >( empty Map (), Ġempty Map ()) ĊĠĠĠĠĠĠĠ Ġoperator Ġfun Ġ< A > Ġinvoke ( a : ĠA ): ĠState < A > Ġ= ĠWaitFor Take ( a , Ġempty Map ()) ĊĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġfun Ġ< A > Ġempty (): ĠState < A > Ġ= Ġref Ġas ĠState < A > ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġdata Ġclass ĠWaitFor Put < A >( val Ġreads : ĠMap < Token , ĠListener < A >>, Ġval Ġtakes : ĠMap < Token , ĠListener < A >>) Ġ: ĠState < A >() ĊĠĠĠĠĠ Ġdata Ġclass ĠWaitFor Take < A >( val Ġvalue : ĠA , Ġval Ġlisteners : ĠMap < Token , ĠPair < A , ĠListener < Unit >> >) Ġ: ĠState < A >() ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ internal Ġtype alias ĠListener < A > Ġ= Ġ( Result < A >) Ġ-> ĠUnit Ċ Ċ internal Ġfun Ġ< K , ĠV > ĠList < Map . Entry < K , ĠV >> . toMap (): ĠMap < K , ĠV > Ġ= ĊĠ Ġwhen Ġ( size ) Ġ{ ĊĠĠĠĠ 0 Ġ-> Ġempty Map () ĊĠĠĠ Ġelse Ġ-> ĠLinkedHashMap < K , ĠV >( size ). apply Ġ{ ĊĠĠĠĠĠ Ġfor Ġ(( key , Ġvalue ) Ġin Ġthis @ toMap ) Ġ{ ĊĠĠĠĠĠĠĠ Ġput ( key , Ġvalue ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ