/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 ĠDavid ĠK arn ok Ċ Ġ* Ċ Ġ* Ġ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 Ġhu . ak arn ok d . kotlin . flow Ċ Ċ import Ġkotlinx . coroutines . Flow Preview Ċ import Ġkotlinx . coroutines . flow . Abstract Flow Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Flow Collector Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ /** Ċ Ġ* ĠMult icast s Ġitems Ġto Ġany Ġnumber Ġof Ġcollect ors Ġwhen Ġthey Ġare Ġready Ġto Ġreceive . Ċ Ġ* Ċ Ġ* Ġ@ param Ġ< T > Ġthe Ġelement Ġtype Ġof Ġthe Ġ[ Flow ] Ċ Ġ*/ Ċ @ Flow Preview Ċ class ĠPublish Subject < T > Ġ: ĠAbstract Flow < T >(), ĠSubject API < T > Ġ Ġ{ ĊĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠEMPTY Ġ= ĠarrayOf < Res um able Collector < Any >>() ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠTER MIN ATED Ġ= ĠarrayOf < Res um able Collector < Any >>() ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprivate Ġval Ġcollect ors Ġ= ĠAtomic Reference ( EMPTY Ġas ĠArray < Res um able Collector < T >>) ĊĊĠĠĠ Ġprivate Ġvar Ġerror Ġ: ĠThrowable ? Ġ= Ġnull ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠReturns Ġtrue Ġif Ġthis ĠPublish Subject Ġhas Ġany Ġcollect ors . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġfun Ġhas Collectors () Ġ: ĠBoolean Ġ= Ġcollect ors . get (). isNotEmpty () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠReturns Ġthe Ġcurrent Ġnumber Ġof Ġcollect ors . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġfun Ġcollector Count () Ġ: ĠInt Ġ= Ġcollect ors . get (). size ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠEmit Ġthe Ġvalue Ġto Ġall Ġcurrent Ġcollect ors , Ġwaiting Ġfor Ġeach Ġof Ġthem ĊĠĠĠĠ Ġ* Ġto Ġbe Ġready Ġfor Ġconsuming Ġit . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit ( value : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ( collector Ġin Ġcollect ors . get ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . next ( value ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThrow Ġan Ġerror Ġon Ġthe Ġconsumer Ġside . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit Error ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( this . error Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . error Ġ= Ġex ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( collector Ġin Ġcollect ors . get AndSet ( TERMIN ATED Ġas ĠArray < Res um able Collector < T >> )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . error ( ex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠIndic ate Ġno Ġfurther Ġitems Ġwill Ġbe Ġemitted ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcomplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġfor Ġ( collector Ġin Ġcollect ors . get AndSet ( TERMIN ATED Ġas ĠArray < Res um able Collector < T >> )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . complete () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ", Ġ"") ĊĠĠĠ Ġprivate Ġfun Ġadd ( inner : ĠRes um able Collector < T >) Ġ: ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġa Ġ= Ġcollect ors . get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( a Ġas ĠAny Ġ== ĠTER MIN ATED Ġas ĠAny ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġa . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġb Ġ= Ġa . copyOf ( n Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġb [ n ] Ġ= Ġinner ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( collect ors . compare AndSet ( a , Ġb Ġas ĠArray < Res um able Collector < T >> )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġprivate Ġfun Ġremove ( inner : ĠRes um able Collector < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġa Ġ= Ġcollect ors . get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġa . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( n Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġj Ġ= Ġa . indexOf ( inner ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( j Ġ< Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġb Ġ= ĠEMPTY Ġas ĠArray < Res um able Collector < T > ?> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( n Ġ!= Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġb Ġ= ĠArray ( n Ġ- Ġ 1 ) Ġ{ Ġnull Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSystem . arraycopy ( a , Ġ 0 , Ġb , Ġ 0 , Ġj ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSystem . arraycopy ( a , Ġj Ġ+ Ġ 1 , Ġb , Ġj , Ġn Ġ- Ġj Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( collect ors . compare AndSet ( a , Ġb Ġas ĠArray < Res um able Collector < T >> )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠStart Ġcollecting Ġsignals Ġfrom Ġthis ĠPublish Subject . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Flow Preview ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcollect S af ely ( collector : ĠFlow Collector < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġinner Ġ= ĠRes um able Collector < T >() ĊĠĠĠĠĠĠĠ Ġif Ġ( add ( inner )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinner . drain ( collector ) Ġ{ Ġremove ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġex Ġ= Ġerror ĊĠĠĠĠĠĠĠ Ġif Ġ( ex Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }