/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 - 2 0 1 8 ĠJetBrains Ġs . r . o . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ package Ġkotlinx . coroutines . experimental . channels Ċ Ċ import Ġkotlinx . coroutines . experimental .* Ċ import Ġkotlinx . coroutines . experimental . internal .* Ċ import Ġkotlinx . coroutines . experimental . select s .* Ċ Ċ /** Ċ Ġ* ĠBroadcast Ġchannel Ġwith Ġarray Ġbuffer Ġof Ġa Ġfixed Ġ[ capacity ]. Ċ Ġ* ĠSender Ġsus p ends Ġonly Ġwhen Ġbuffer Ġis Ġfull Ġdue Ġto Ġone Ġof Ġthe Ġreceives Ġbeing Ġslow Ġto Ġconsume Ġand Ċ Ġ* Ġreceiver Ġsus p ends Ġonly Ġwhen Ġbuffer Ġis Ġempty . Ċ Ġ* Ċ Ġ* Ġ** Note **, Ġthat Ġelements Ġthat Ġare Ġsent Ġto Ġthis Ġchannel Ġwhile Ġthere Ġare Ġno Ċ Ġ* Ġ[ open Subscription ] Ġsubscribers Ġare Ġimmediately Ġlost . Ċ Ġ* Ċ Ġ* ĠThis Ġchannel Ġis Ġcreated Ġby Ġ` Broadcast Channel ( capacity )` Ġfactory Ġfunction Ġinvocation . Ċ Ġ* Ċ Ġ* ĠThis Ġimplementation Ġuses Ġlock Ġto Ġprotect Ġthe Ġbuffer , Ġwhich Ġis Ġheld Ġonly Ġduring Ġvery Ġshort Ġbuffer - update Ġoperations . Ċ Ġ* ĠThe Ġlock Ġat Ġeach Ġsubscription Ġis Ġalso Ġused Ġto Ġmanage Ġconcurrent Ġattempts Ġto Ġreceive Ġfrom Ġthe Ġsame Ġsubscriber . Ċ Ġ* ĠThe Ġlists Ġof Ġsus pended Ġsend ers Ġor Ġreceiv ers Ġare Ġlock - free . Ċ Ġ* Ċ Ġ* Ġ@ suppress Ġ** This Ġan Ġinternal ĠAPI Ġand Ġshould Ġnot Ġbe Ġused Ġfrom Ġgeneral Ġcode .** Ċ Ġ*/ Ċ @ Internal Cor outines Api Ċ public Ġclass ĠArray Broadcast Channel < E > Ċ @ Deprecated ( ĊĠĠĠ Ġ" Replace Ġwith ĠBroadcast Channel Ġfactory Ġfunction ", ĊĠĠĠ Ġreplace With Ġ= ĠReplace With (" Broadcast Channel ( capacity )") Ċ ) Ċ constructor ( ĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠBuffer Ġcapacity . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġcapacity : ĠInt Ċ ) Ġ: ĠAbstract Send Channel < E >(), ĠBroadcast Channel < E > Ġ{ ĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire ( capacity Ġ>= Ġ 1 ) Ġ{ Ġ" Array Broadcast Channel Ġcapacity Ġmust Ġbe Ġat Ġleast Ġ 1 , Ġbut Ġ$ capacity Ġwas Ġspecified " Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/* ĊĠĠĠĠ Ġ* Ġ ĠWrites Ġto Ġbuffer Ġare Ġgu arded Ġby Ġbuffer Lock , Ġbut Ġreads Ġfrom Ġbuffer Ġare Ġconcurrent Ġwith Ġwrites ĊĠĠĠĠ Ġ* ĠĠĠ Ġ- ĠWrite Ġelement Ġto Ġbuffer Ġthen Ġwrite Ġ" tail " Ġ( volatile ) ĊĠĠĠĠ Ġ* ĠĠĠ Ġ- ĠRead Ġ" tail " Ġ( volatile ), Ġthen Ġread Ġelement Ġfrom Ġbuffer ĊĠĠĠĠ Ġ* Ġ ĠSo Ġread / writes Ġto Ġbuffer Ġneed Ġnot Ġbe Ġvolatile ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġbuffer Lock Ġ= ĠRe entrant Lock () ĊĠĠĠ Ġprivate Ġval Ġbuffer Ġ= ĠarrayOf Null s < Any ? >( capacity ) ĊĊĠĠĠ Ġ// Ġhead Ġ& Ġtail Ġare ĠLong Ġ( 6 4 Ġbits ) Ġand Ġwe Ġassume Ġthat Ġthey Ġnever Ġwrap Ġaround ĊĠĠĠ Ġ// Ġhead , Ġtail , Ġand Ġsize Ġare Ġgu arded Ġby Ġbuffer Lock ĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġhead : ĠLong Ġ= Ġ 0 Ġ// Ġdo Ġmodulo Ġon Ġuse Ġof Ġhead ĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġtail : ĠLong Ġ= Ġ 0 Ġ// Ġdo Ġmodulo Ġon Ġuse Ġof Ġtail ĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġsize : ĠInt Ġ= Ġ 0 ĊĊĠĠĠ Ġprivate Ġval Ġsubscribers Ġ= Ġsubscriber List < Subscriber < E >>() ĊĊĠĠĠ Ġoverride Ġval Ġis Buffer Always Full : ĠBoolean Ġget () Ġ= Ġfalse ĊĠĠĠ Ġoverride Ġval Ġis Buffer Full : ĠBoolean Ġget () Ġ= Ġsize Ġ>= Ġcapacity ĊĊĠĠĠ Ġpublic Ġoverride Ġfun Ġopen Subscription (): ĠReceive Channel < E > Ġ= ĊĠĠĠĠĠĠĠ ĠSubscriber ( this ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Head ( add Sub Ġ= Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġoverride Ġfun Ġclose ( cause : ĠThrowable ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! super . close ( cause )) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġcheck Sub Offers () ĊĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġoverride Ġfun Ġcancel ( cause : ĠThrowable ?): ĠBoolean Ġ= ĊĠĠĠĠĠĠĠ Ġclose ( cause ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( sub Ġin Ġsubscribers ) Ġsub . cancel ( cause ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġresult Ġis Ġ` OFF ER _ SUCCESS Ġ| ĠO FFER _ FAILED Ġ| ĠClosed ` ĊĠĠĠ Ġoverride Ġfun Ġoffer Internal ( element : ĠE ): ĠAny Ġ{ ĊĠĠĠĠĠĠĠ Ġbuffer Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcheck Ġif Ġclosed Ġfor Ġsend Ġ( under Ġlock , Ġso Ġsize Ġcannot Ġchange ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclosed For Send ?. let Ġ{ Ġreturn Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsize Ġ= Ġthis . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( size Ġ>= Ġcapacity ) Ġreturn ĠO FFER _ FAILED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtail Ġ= Ġthis . tail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbuffer [( tail Ġ% Ġcapacity ). toInt ()] Ġ= Ġelement ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . size Ġ= Ġsize Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . tail Ġ= Ġtail Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// Ġif Ġoffered Ġsuccessfully , Ġthen Ġcheck Ġsubscribers Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠ Ġcheck Sub Offers () ĊĠĠĠĠĠĠĠ Ġreturn ĠO FFER _ SUCCESS ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġresult Ġis Ġ` AL READY _ SELECTED Ġ| ĠO FFER _ SUCCESS Ġ| ĠO FFER _ FAILED Ġ| ĠClosed ` ĊĠĠĠ Ġoverride Ġfun Ġoffer Select Internal ( element : ĠE , Ġselect : ĠSelect Instance < *> ): ĠAny Ġ{ ĊĠĠĠĠĠĠĠ Ġbuffer Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcheck Ġif Ġclosed Ġfor Ġsend Ġ( under Ġlock , Ġso Ġsize Ġcannot Ġchange ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclosed For Send ?. let Ġ{ Ġreturn Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsize Ġ= Ġthis . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( size Ġ>= Ġcapacity ) Ġreturn ĠO FFER _ FAILED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġlet 's Ġtry Ġto Ġselect Ġsending Ġthis Ġelement Ġto Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! select . try Select ( null )) Ġ{ Ġ// Ġ: todo : Ġmove Ġtry Select Ġcompletion Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠAL READY _ SELECTED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtail Ġ= Ġthis . tail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbuffer [( tail Ġ% Ġcapacity ). toInt ()] Ġ= Ġelement ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . size Ġ= Ġsize Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . tail Ġ= Ġtail Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// Ġif Ġoffered Ġsuccessfully , Ġthen Ġcheck Ġsubscribers Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠ Ġcheck Sub Offers () ĊĠĠĠĠĠĠĠ Ġreturn ĠO FFER _ SUCCESS ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcheck Sub Offers () Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġupdated Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġvar Ġhas Sub s Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" Loop To Call Chain ") Ġ// Ġmust Ġinvoke Ġ` check Offer ` Ġon Ġevery Ġsub ĊĠĠĠĠĠĠĠ Ġfor Ġ( sub Ġin Ġsubscribers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhas Sub s Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sub . check Offer ()) Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( updated Ġ|| Ġ! has Sub s ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Head () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġupdates Ġhead Ġif Ġneeded Ġand Ġoptionally Ġadds Ġ/ Ġremoves Ġsubscriber Ġunder Ġthe Ġsame Ġlock ĊĠĠĠ Ġprivate Ġtail rec Ġfun Ġupdate Head ( add Sub : ĠSubscriber < E >? Ġ= Ġnull , Ġremove Sub : ĠSubscriber < E >? Ġ= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġupdate Ġhead Ġin Ġa Ġtail Ġrec Ġloop ĊĠĠĠĠĠĠĠ Ġvar Ġsend : ĠSend ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġtoken : ĠAny ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġbuffer Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( add Sub Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Sub . sub Head Ġ= Ġtail Ġ// Ġstart Ġfrom Ġlast Ġelement ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwas Empty Ġ= Ġsubscribers . isEmpty () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscribers . add ( add Sub ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! was Empty ) Ġreturn Ġ// Ġno Ġneed Ġto Ġupdate Ġwhen Ġadding Ġsecond Ġand Ġetc Ġsub ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( remove Sub Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscribers . remove ( remove Sub ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( head Ġ!= Ġremove Sub . sub Head ) Ġreturn Ġ// Ġno Ġneed Ġto Ġupdate ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmin Head Ġ= Ġcompute Min Head () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtail Ġ= Ġthis . tail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġhead Ġ= Ġthis . head ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtarget Head Ġ= Ġmin Head . coerce At Most ( tail ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( target Head Ġ<= Ġhead ) Ġreturn Ġ// Ġnothing Ġto Ġdo Ġ-- Ġhead Ġwas Ġalready Ġmoved ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsize Ġ= Ġthis . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġclean Ġup Ġremoved Ġ( on Ġnot Ġneed Ġif Ġwe Ġdon 't Ġhave Ġany Ġsubscribers Ġanymore ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( head Ġ< Ġtarget Head ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbuffer [( head Ġ% Ġcapacity ). toInt ()] Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwas Full Ġ= Ġsize Ġ>= Ġcapacity ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġupdate Ġthe Ġsize Ġbefore Ġchecking Ġqueue Ġ( no Ġmore Ġsend ers Ġcan Ġqueue Ġup ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . head Ġ= Ġ++ head ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . size Ġ= Ġ-- size ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( was Full ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Ġ= Ġtake First Send Or Peek Closed () Ġ?: Ġbreak Ġ// Ġwhen Ġwhen Ġno Ġsender ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( send Ġis ĠClosed <* >) Ġbreak Ġ// Ġbreak Ġwhen Ġclosed Ġfor Ġsend ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken Ġ= Ġsend !!. try Resume Send ( ide mpot ent Ġ= Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( token Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġput Ġsent Ġelement Ġto Ġthe Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbuffer [( tail Ġ% Ġcapacity ). toInt ()] Ġ= Ġ( send Ġas ĠSend ). poll Result ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . size Ġ= Ġsize Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . tail Ġ= Ġtail Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ with Lock Ġ// Ġgo Ġout Ġof Ġlock Ġto Ġwake up Ġthis Ġsender ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ// Ġdone Ġupdating Ġhere Ġ-> Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// Ġwe Ġonly Ġget Ġout Ġof Ġthe Ġlock Ġnormally Ġwhen Ġthere Ġis Ġa Ġsender Ġto Ġresume ĊĠĠĠĠĠĠĠ Ġsend !!. complete Resume Send ( token !! ) ĊĠĠĠĠĠĠĠ Ġ// Ġsince Ġwe 've Ġjust Ġsent Ġan Ġelement , Ġwe Ġmight Ġneed Ġto Ġresume Ġsome Ġreceiv ers ĊĠĠĠĠĠĠĠ Ġcheck Sub Offers () ĊĠĠĠĠĠĠĠ Ġ// Ġtail rec Ġcall Ġto Ġre check ĊĠĠĠĠĠĠĠ Ġupdate Head () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcompute Min Head (): ĠLong Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġmin Head Ġ= ĠLong . MAX _ VALUE ĊĠĠĠĠĠĠĠ Ġfor Ġ( sub Ġin Ġsubscribers ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmin Head Ġ= Ġmin Head . coerce At Most ( sub . sub Head ) Ġ// Ġvolatile Ġ( atomic ) Ġreads Ġof Ġsub Head ĊĠĠĠĠĠĠĠ Ġreturn Ġmin Head ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprivate Ġfun Ġelement At ( index : ĠLong ): ĠE Ġ= Ġbuffer [( index Ġ% Ġcapacity ). toInt ()] Ġas ĠE ĊĊĠĠĠ Ġprivate Ġclass ĠSubscriber < E >( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġbroadcast Channel : ĠArray Broadcast Channel < E > ĊĠĠĠ Ġ) Ġ: ĠAbstract Channel < E >(), ĠReceive Channel < E >, ĠSubscription Receive Channel < E > Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsub Lock Ġ= ĠRe entrant Lock () ĊĊĠĠĠĠĠĠĠ Ġ@ Volatile ĊĠĠĠĠĠĠĠ Ġ@ Jvm Field ĊĠĠĠĠĠĠĠ Ġvar Ġsub Head : ĠLong Ġ= Ġ 0 Ġ// Ġgu arded Ġby Ġsub Lock ĊĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġis Buffer Always Empty : ĠBoolean Ġget () Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġis Buffer Empty : ĠBoolean Ġget () Ġ= Ġsub Head Ġ>= Ġbroadcast Channel . tail ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġis Buffer Always Full : ĠBoolean Ġget () Ġ= Ġerror (" Should Ġnot Ġbe Ġused ") ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġis Buffer Full : ĠBoolean Ġget () Ġ= Ġerror (" Should Ġnot Ġbe Ġused ") ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġcancel ( cause : ĠThrowable ?): ĠBoolean Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose ( cause ). also Ġ{ Ġclosed Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( closed ) Ġbroadcast Channel . update Head ( remove Sub Ġ= Ġthis ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclear Buffer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġclear Buffer () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsub Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsub Head Ġ= Ġbroadcast Channel . tail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġreturns Ġtrue Ġif Ġsub Head Ġwas Ġupdated Ġand Ġbroadcast Ġchannel 's Ġhead Ġmust Ġbe Ġchecked ĊĠĠĠĠĠĠĠ Ġ// Ġthis Ġmethod Ġis Ġlock - free Ġ( it Ġnever Ġwaits Ġon Ġlock ) ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġfun Ġcheck Offer (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġupdated Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġclosed : ĠClosed < *> ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġloop @ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( need sTo Check Offer Without Lock ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġjust Ġuse Ġ` try Lock ` Ġhere Ġand Ġbreak Ġwhen Ġsome Ġother Ġthread Ġis Ġchecking Ġunder Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġit Ġmeans Ġthat Ġ` check Offer ` Ġmust Ġbe Ġretried Ġafter Ġevery Ġ` unlock ` ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! sub Lock . try Lock ()) Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceive : ĠReceive Or Closed < E >? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtoken : ĠAny ? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġpeek Under Lock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ=== ĠPOL L _ FAILED Ġ-> Ġcontinue @ loop Ġ// Ġmust Ġre test Ġ` need sTo Check Offer Without Lock ` Ġoutside Ġof Ġthe Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġis ĠClosed < *> Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclosed Ġ= Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak @ loop Ġ// Ġwas Ġclosed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġfind Ġa Ġreceiver Ġfor Ġan Ġelement ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceive Ġ= Ġtake First Receive Or Peek Closed () Ġ?: Ġbreak Ġ// Ġbreak Ġwhen Ġno Ġone 's Ġreceiving ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( receive Ġis ĠClosed <* >) Ġbreak Ġ// Ġnot ing Ġmore Ġto Ġdo Ġif Ġthis Ġsub Ġalready Ġclosed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken Ġ= Ġreceive . try Resume Receive ( result Ġas ĠE , Ġide mpot ent Ġ= Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( token Ġ== Ġnull ) Ġcontinue Ġ// Ġbail Ġout Ġhere Ġto Ġnext Ġiteration Ġ( see Ġfor Ġnext Ġreceiver ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsub Head Ġ= Ġthis . sub Head ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . sub Head Ġ= Ġsub Head Ġ+ Ġ 1 Ġ// Ġretrieved Ġelement Ġfor Ġthis Ġsubscriber ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsub Lock . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceive !!. complete Resume Receive ( token !! ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdo Ġclose Ġoutside Ġof Ġlock Ġif Ġneeded ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclosed ?. also Ġ{ Ġclose ( cause Ġ= Ġit . close Cause ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġupdated ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġresult Ġis Ġ` E Ġ| ĠPOL L _ FAILED Ġ| ĠClosed ` ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġpoll Internal (): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġupdated Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġsub Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġpeek Under Lock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġis ĠClosed < *> Ġ-> Ġ{ Ġ/* Ġjust Ġbail Ġout Ġof Ġlock Ġ*/ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ=== ĠPOL L _ FAILED Ġ-> Ġ{ Ġ/* Ġjust Ġbail Ġout Ġof Ġlock Ġ*/ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġupdate Ġsub Head Ġafter Ġretrie iving Ġelement Ġfrom Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsub Head Ġ= Ġthis . sub Head ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . sub Head Ġ= Ġsub Head Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdo Ġclose Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( result Ġas ? ĠClosed <* >) ?. also Ġ{ Ġclose ( cause Ġ= Ġit . close Cause ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthere Ġcould Ġhave Ġbeen Ġcheck Offer Ġattempt Ġwhile Ġwe Ġwere Ġholding Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnow Ġoutside Ġthe Ġlock Ġre check Ġif Ġanything Ġelse Ġto Ġoffer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( check Offer ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġand Ġfinally Ġupdate Ġbroadcast 's Ġchannel Ġhead Ġif Ġneeded ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( updated ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbroadcast Channel . update Head () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġresult Ġis Ġ` AL READY _ SELECTED Ġ| ĠE Ġ| ĠPOL L _ FAILED Ġ| ĠClosed ` ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġpoll Select Internal ( select : ĠSelect Instance < *> ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġupdated Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġsub Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġpeek Under Lock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġis ĠClosed < *> Ġ-> Ġ{ Ġ/* Ġjust Ġbail Ġout Ġof Ġlock Ġ*/ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ=== ĠPOL L _ FAILED Ġ-> Ġ{ Ġ/* Ġjust Ġbail Ġout Ġof Ġlock Ġ*/ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġlet 's Ġtry Ġto Ġselect Ġreceiving Ġthis Ġelement Ġfrom Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! select . try Select ( null )) Ġ{ Ġ// Ġ: todo : Ġmove Ġtry Select Ġcompletion Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= ĠAL READY _ SELECTED ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġupdate Ġsub Head Ġafter Ġretrie iving Ġelement Ġfrom Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsub Head Ġ= Ġthis . sub Head ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . sub Head Ġ= Ġsub Head Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdo Ġclose Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( result Ġas ? ĠClosed <* >) ?. also Ġ{ Ġclose ( cause Ġ= Ġit . close Cause ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthere Ġcould Ġhave Ġbeen Ġcheck Offer Ġattempt Ġwhile Ġwe Ġwere Ġholding Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnow Ġoutside Ġthe Ġlock Ġre check Ġif Ġanything Ġelse Ġto Ġoffer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( check Offer ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdated Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġand Ġfinally Ġupdate Ġbroadcast 's Ġchannel Ġhead Ġif Ġneeded ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( updated ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbroadcast Channel . update Head () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠMust Ġinvoke Ġthis Ġcheck Ġthis Ġafter Ġlock , Ġbecause Ġoffer 's Ġinvocation Ġof Ġ` check Offer ` Ġmight Ġhave Ġfailed ĊĠĠĠĠĠĠĠ Ġ// Ġto Ġ` try Lock ` Ġjust Ġbefore Ġthe Ġlock Ġwas Ġabout Ġto Ġun locked , Ġthus Ġlo osing Ġnotification Ġto Ġthis ĊĠĠĠĠĠĠĠ Ġ// Ġsubscription Ġabout Ġan Ġelement Ġthat Ġwas Ġjust Ġoffered ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġneeds ToCheck Offer Without Lock (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( closed For Receive Ġ!= Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse Ġ// Ġalready Ġclosed Ġ-> Ġnothing Ġto Ġdo ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Buffer Empty Ġ&& Ġbroadcast Channel . closed For Receive Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse Ġ// Ġno Ġdata Ġfor Ġus Ġ&& Ġbroadcast Ġchannel Ġwas Ġnot Ġclosed Ġyet Ġ-> Ġnothing Ġto Ġdo ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue Ġ// Ġcheck Ġotherwise ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġgu arded Ġby Ġlock , Ġreturns : ĊĠĠĠĠĠĠĠ Ġ// ĠĠĠĠĠ ĠE Ġ- Ġthe Ġelement Ġfrom Ġthe Ġbuffer Ġat Ġsub Head ĊĠĠĠĠĠĠĠ Ġ// ĠĠĠĠĠ ĠClosed < *> Ġwhen Ġclosed ; ĊĠĠĠĠĠĠĠ Ġ// ĠĠĠĠĠ ĠPOL L _ FAILED Ġwhen Ġthere Ġseems Ġto Ġbe Ġno Ġdata , Ġbut Ġmust Ġre test Ġ` need sTo Check Offer Without Lock ` Ġoutside Ġof Ġlock ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġpeek Under Lock (): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsub Head Ġ= Ġthis . sub Head Ġ// Ġgu arded Ġread Ġ( can Ġbe Ġnon - volatile Ġread ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnote : Ġfrom Ġthe Ġbroadcast Channel Ġwe Ġmust Ġread Ġclosed Ġtoken Ġfirst , Ġthen Ġread Ġits Ġtail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbecause Ġit Ġis ĠOk Ġif Ġtail Ġmoves Ġin Ġbetween Ġthe Ġreads Ġ( we Ġmake Ġdecision Ġbased Ġon Ġtail Ġfirst ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclosed Broadcast Ġ= Ġbroadcast Channel . closed For Receive Ġ// Ġun gu arded Ġvolatile Ġread ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtail Ġ= Ġbroadcast Channel . tail Ġ// Ġun gu arded Ġvolatile Ġread ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sub Head Ġ>= Ġtail ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġno Ġelements Ġto Ġpoll Ġfrom Ġthe Ġqueue Ġ-- Ġcheck Ġif Ġclosed Ġbroad s Ġ& Ġclosed Ġthis Ġsub ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġmust Ġre test Ġ` need sTo Check Offer Without Lock ` Ġoutside Ġof Ġthe Ġlock ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġclosed Broadcast Ġ?: Ġthis . closed For Receive Ġ?: ĠPOL L _ FAILED ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGet Ġtent ative Ġresult . ĠThis Ġresult Ġmay Ġbe Ġwrong Ġ( comp let ely Ġinvalid Ġvalue , Ġincluding Ġnull ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbecause Ġthis Ġsubscription Ġmight Ġget Ġclosed , Ġmoving Ġchannel 's Ġhead Ġpast Ġthis Ġsubscription 's Ġhead . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġbroadcast Channel . elementAt ( sub Head ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnow Ġcheck Ġif Ġthis Ġsubscription Ġwas Ġclosed ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclosed Sub Ġ= Ġthis . closed For Receive ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( closed Sub Ġ!= Ġnull ) Ġreturn Ġclosed Sub ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġknow Ġthe Ġsubscription Ġwas Ġnot Ġclosed , Ġso Ġthis Ġtent ative Ġresult Ġis ĠOk Ġto Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġ------ Ġdebug Ġ------ ĊĊĠĠĠ Ġoverride Ġval Ġbuffer Debug String : ĠString ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġ"( buffer : capacity =${ buffer . size }, size =$ size )" Ċ }