package Ġio . small ry e . mut iny . coroutines Ċ Ċ import Ġio . small ry e . mut iny . Multi Ċ import Ġio . small ry e . mut iny . subscription . Multi Emitter Ċ import Ġio . small ry e . mut iny . subscription . Multi Subscriber Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . channel Flow Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġkotlinx . coroutines . sync . Mutex Ċ import Ġorg . reactiv est ream s . Subscription Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ import Ġkotlin . coroutines . coroutine Context Ċ Ċ /** Ċ Ġ* ĠSubscribe Ġto Ġthis Ġ[ Multi ] Ġand Ġprovide Ġthe Ġitems Ġas Ġ[ Flow ]. Ċ Ġ* Ċ Ġ* ĠIf Ġthis Ġ[ Multi ] Ġemits Ġa Ġfailure , Ġit Ġis Ġthrown Ġfrom Ġthe ĠFlow Ġwhen Ġoperating Ġon Ġit . Ċ Ġ* Ċ Ġ* ĠThe Ġsubscription Ġto Ġthis Ġ[ Multi ] Ġgets Ġautomatically Ġcancelled Ġwhen Ġthe Ġsurrounding Ġcoroutine Ġfails Ġor Ġgets Ġcancelled . Ċ Ġ*/ Ċ @ Experimental Cor outines Api Ċ suspend Ġfun Ġ< T > ĠMulti < T >. as Flow (): ĠFlow < T > Ġ= Ġchannel Flow < T > Ġ{ ĊĠĠĠ Ġval Ġparent Ctx Ġ= Ġcoroutine Context ĊĠĠĠ Ġval Ġtermination Failure Ġ= ĠAtomic Reference < Throwable ? >( null ) ĊĠĠĠ Ġval Ġtermination Lock Ġ= ĠMutex ( locked Ġ= Ġtrue ) ĊĊĠĠĠ Ġval Ġsubscriber Ġ= Ġobject Ġ: ĠMulti Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsubscription Ġ= ĠAtomic Reference < Subscription ?> () ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinvoke On Close Ġ{ Ġreason Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( reason Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcoroutine Ġwas Ġcancelled Ġor Ġis Ġfailed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription . get ()?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( subscription . compare AndSet ( null , Ġs )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . request ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Item ( item : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking ( parent Ctx ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . send ( item ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCor outine Ġwas Ġcancelled Ġeither Ġregular ly Ġor Ġby Ġfailure ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġin Ġcase Ġof Ġa Ġregular Ġcancellation , Ġe . g . Ġby ĠFlow Ġabort ion Ġ( happ ens Ġalso Ġby ĠFlow . first ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġin Ġcase Ġof Ġa Ġfailure , Ġ' invoke On Close ' Ġis Ġcalled Ġwith Ġa Ġreason Ġ- Ġsetting Ġa Ġtermin at ino Failure Ġis Ġnot Ġnecessary Ġhere ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( termination Lock . is Locked ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription . get ()?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtermination Lock . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( failure : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtermination Failure . set ( failure ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtermination Lock . unlock () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Completion () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( termination Lock . is Locked ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtermination Lock . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsubscribe (). with Subscriber ( subscriber ) ĊĠĠĠ Ġ// Ġwait Ġ( sus pending ) Ġfor Ġcompletion Ġor Ġfailure Ġof ĠMulti Ġto Ġterminate Ġthe ĠFlow ĊĠĠĠ Ġtermination Lock . lock () ĊĠĠĠ Ġtermination Failure . get ()?. also Ġ{ Ġthrow Ġit Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠProvide Ġthis Ġ[ Flow ] s Ġitems Ġor Ġfailure Ġas Ġ[ Multi ]. Ċ Ġ* Ċ Ġ* ĠThe ĠMulti Emitter Ġis Ġscheduled Ġas Ġchild Ġof Ġcall ings Ġcoroutine Ġcontext . Ċ Ġ* Ċ Ġ* ĠPlease Ġnote , Ġthat Ġthe Ġ[ Flow ] s Ġvalues Ġare Ġemitted Ġimmediately , Ċ Ġ* Ġwithout Ġrespect ing Ġthe Ġrequested Ġamount Ġof Ġthe Ġsubscriber . Ċ Ġ*/ Ċ suspend Ġfun Ġ< T > ĠFlow < T >. as Multi (): ĠMulti < T > Ġ{ ĊĠĠĠ Ġval Ġparent Ctx Ġ= Ġcoroutine Context ĊĠĠĠ Ġreturn ĠMulti . createFrom (). emitter Ġ{ Ġem : ĠMulti Emitter < in ĠT > Ġ-> ĊĠĠĠĠĠĠĠ ĠCor outine Scope ( parent Ctx ). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollect Ġ{ Ġitem Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( em . is Cancelled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠC ancellation Exception (" Multi Ġsubscription Ġcancelled ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . emit ( item ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . complete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( th : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . fail ( th ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ