/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 - 2 0 2 1 ĠJetBrains Ġs . r . o . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ package Ġkotlinx . coroutines . flow . internal Ċ Ċ import Ġkotlinx . coroutines . channels .* Ċ import Ġkotlinx . coroutines . flow .* Ċ import Ġkotlinx . coroutines . internal .* Ċ import Ġkotlin . coroutines .* Ċ import Ġkotlin . jvm .* Ċ import Ġkotlin . native . concurrent .* Ċ Ċ @ Jvm Field Ċ @ Shared Immutable Ċ internal Ġval ĠEMPTY _ RES UM ES Ġ= ĠarrayOf Null s < Continuation < Unit >? >( 0 ) Ċ Ċ internal Ġabstract Ġclass ĠAbstract Shared Flow Slot < F > Ġ{ ĊĠĠĠ Ġabstract Ġfun Ġallocate Locked ( flow : ĠF ): ĠBoolean ĊĠĠĠ Ġabstract Ġfun Ġfree Locked ( flow : ĠF ): ĠArray < Continuation < Unit > ?> Ġ// Ġreturns Ġcontinu ations Ġto Ġresume Ġafter Ġlock Ċ } Ċ Ċ internal Ġabstract Ġclass ĠAbstract Shared Flow < S Ġ: ĠAbstract Shared Flow Slot <* >> Ġ: ĠSynchron ized Object () Ġ{ ĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprotected Ġvar Ġslots : ĠArray < S ?> ? Ġ= Ġnull Ġ// Ġallocated Ġwhen Ġneeded ĊĠĠĠĠĠĠĠ Ġprivate Ġset ĊĠĠĠ Ġprotected Ġvar Ġn Collectors Ġ= Ġ 0 Ġ// Ġnumber Ġof Ġallocated Ġ(! free ) Ġslots ĊĠĠĠĠĠĠĠ Ġprivate Ġset ĊĠĠĠ Ġprivate Ġvar Ġnext Index Ġ= Ġ 0 Ġ// Ġoracle Ġfor Ġthe Ġnext Ġfree Ġslot Ġindex ĊĠĠĠ Ġprivate Ġvar Ġ_ subscription Count : ĠSubscription Count State Flow ? Ġ= Ġnull Ġ// Ġinit Ġon Ġfirst Ġneed ĊĊĠĠĠ Ġval Ġsubscription Count : ĠState Flow < Int > ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġallocate Ġunder Ġlock Ġin Ġsync Ġwith Ġn Collectors Ġvariable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ_ subscription Count Ġ?: ĠSubscription Count State Flow ( n Collectors ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ_ subscription Count Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġabstract Ġfun Ġcreate Slot (): ĠS ĊĊĠĠĠ Ġprotected Ġabstract Ġfun Ġcreate Slot Array ( size : ĠInt ): ĠArray < S ?> ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprotected Ġfun Ġallocate Slot (): ĠS Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠActually Ġcreate Ġslot Ġunder Ġlock ĊĠĠĠĠĠĠĠ Ġvar Ġsubscription Count : ĠSubscription Count State Flow ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġslot Ġ= Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġslots Ġ= Ġwhen Ġ( val Ġcur Slots Ġ= Ġslots ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull Ġ-> Ġcreate Slot Array ( 2 ). also Ġ{ Ġslots Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġif Ġ( n Collectors Ġ>= Ġcur Slots . size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcur Slots . copyOf ( 2 Ġ* Ġcur Slots . size ). also Ġ{ Ġslots Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcur Slots ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġindex Ġ= Ġnext Index ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġslot : ĠS ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġslot Ġ= Ġslots [ index ] Ġ?: Ġcreate Slot (). also Ġ{ Ġslots [ index ] Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġindex ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( index Ġ>= Ġslots . size ) Ġindex Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(( slot Ġas ĠAbstract Shared Flow Slot < Any >). allocate Locked ( this )) Ġbreak Ġ// Ġbreak Ġwhen Ġfound Ġand Ġallocated Ġfree Ġslot ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnext Index Ġ= Ġindex ĊĠĠĠĠĠĠĠĠĠĠĠ Ġn Collectors ++ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription Count Ġ= Ġ_ subscription Count Ġ// Ġretrieve Ġunder Ġlock Ġif Ġinitialized ĊĠĠĠĠĠĠĠĠĠĠĠ Ġslot ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// Ġincre ments Ġsubscription Ġcount ĊĠĠĠĠĠĠĠ Ġsubscription Count ?. increment ( 1 ) ĊĠĠĠĠĠĠĠ Ġreturn Ġslot ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprotected Ġfun Ġfree Slot ( slot : ĠS ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠRelease Ġslot Ġunder Ġlock ĊĠĠĠĠĠĠĠ Ġvar Ġsubscription Count : ĠSubscription Count State Flow ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġresume s Ġ= Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġn Collectors -- ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription Count Ġ= Ġ_ subscription Count Ġ// Ġretrieve Ġunder Ġlock Ġif Ġinitialized ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠReset Ġnext Ġindex Ġoracle Ġif Ġwe Ġhave Ġno Ġmore Ġactive Ġcollect ors Ġfor Ġmore Ġpredict able Ġbehavior Ġnext Ġtime ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( n Collectors Ġ== Ġ 0 ) Ġnext Index Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( slot Ġas ĠAbstract Shared Flow Slot < Any >). free Locked ( this ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ/* ĊĠĠĠĠĠĠĠĠ Ġ* ĠResume Ġsus pended Ġcor outines . ĊĠĠĠĠĠĠĠĠ Ġ* ĠThis Ġcan Ġhappen Ġwhen Ġthe Ġsubscriber Ġthat Ġwas Ġfreed Ġwas Ġa Ġslow Ġone Ġand Ġwas Ġholding Ġup Ġbuffer . ĊĠĠĠĠĠĠĠĠ Ġ* ĠWhen Ġthis Ġsubscriber Ġwas Ġfreed , Ġpreviously Ġqueued Ġemitted Ġcan Ġnow Ġwake Ġup Ġand Ġare Ġres umed Ġhere . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġfor Ġ( cont Ġin Ġresume s ) Ġcont ?. resume ( Unit ) ĊĠĠĠĠĠĠĠ Ġ// Ġdecrement Ġsubscription Ġcount ĊĠĠĠĠĠĠĠ Ġsubscription Count ?. increment (- 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġinline Ġfun ĠforEach Slot Locked ( block : Ġ( S ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( n Collectors Ġ== Ġ 0 ) Ġreturn ĊĠĠĠĠĠĠĠ Ġslots ?. forEach Ġ{ Ġslot Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( slot Ġ!= Ġnull ) Ġblock ( slot ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* Ġ[ State Flow ] Ġthat Ġrepresents Ġthe Ġnumber Ġof Ġsubscriptions . Ċ Ġ* Ċ Ġ* ĠIt Ġis Ġexposed Ġas Ġa Ġregular Ġ[ State Flow ] Ġin Ġour Ġpublic ĠAPI , Ġbut Ġit Ġis Ġimplemented Ġas Ġ[ Shared Flow ] Ġunder cover Ġto Ċ Ġ* Ġavoid Ġcon fl ations Ġof Ġconsecutive Ġupdates Ġbecause Ġthe Ġsubscription Ġcount Ġis Ġvery Ġsensitive Ġto Ġit . Ċ Ġ* Ċ Ġ* ĠThe Ġimportance Ġof Ġnon - con fl ating Ġcan Ġbe Ġdemonstr ated Ġwith Ġthe Ġfollowing Ġexample : Ċ Ġ* Ġ``` Ċ Ġ* Ġval Ġshared Ġ= Ġflow Of ( 2 3 9 ). state In ( this , ĠSh aring Started . L az ily , Ġ 4 2 ) Ġ// Ġstate In Ġfor Ġthe Ġsake Ġof Ġthe Ġinitial Ġvalue Ċ Ġ* Ġprintln ( shared . first ()) Ċ Ġ* Ġyield () Ċ Ġ* Ġprintln ( shared . first ()) Ċ Ġ* Ġ``` Ċ Ġ* ĠIf Ġthe Ġflow Ġis Ġshared Ġwithin Ġthe Ġsame Ġdispatcher Ġ( e . g . ĠMain ) Ġor Ġwith Ġa Ġslow / th rott led Ġone , Ċ Ġ* Ġthe Ġ` Sharing Started . L az ily ` Ġwill Ġnever Ġbe Ġable Ġto Ġstart Ġthe Ġsource : Ġ` first ` Ġsees Ġthe Ġinitial Ġvalue Ġand Ġimmediately Ċ Ġ* Ġun sub scri bes , Ġleaving Ġthe Ġasynchronous Ġ` Sharing Started ` Ġwith Ġcon fl ated Ġzero . Ċ Ġ* Ċ Ġ* ĠTo Ġavoid Ġthat Ġ( espec ially Ġin Ġa Ġmore Ġcomplex Ġscenarios ), Ġwe Ġdo Ġnot Ġconf late Ġsubscription Ġupdates . Ċ Ġ*/ Ċ private Ġclass ĠSubscription Count State Flow ( initial Value : ĠInt ) Ġ: ĠState Flow < Int >, ĊĠĠĠ ĠShared Flow Impl < Int >( 1 , ĠInt . MAX _ VALUE , ĠBuffer Overflow . DROP _ OL DEST ) Ċ { ĊĠĠĠ Ġinit Ġ{ Ġtry Emit ( initial Value ) Ġ} ĊĊĠĠĠ Ġoverride Ġval Ġvalue : ĠInt ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġsynchronized ( this ) Ġ{ Ġlast Replay ed Locked Ġ} ĊĊĠĠĠ Ġfun Ġincrement ( delta : ĠInt ) Ġ= Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Emit ( last Replay ed Locked Ġ+ Ġdelta ) ĊĠĠĠ Ġ} Ċ } Ċ