/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 - 2 0 2 0 Ġ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 Collector Ċ import Ġjava . util . concurrent . Concurrent Linked Queue Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ /** Ċ Ġ* ĠBu ffers Ġitems Ġuntil Ġa Ġsingle Ġcollector Ġcollect s Ġthem . Ċ Ġ* Ċ Ġ* Ġ@ param Ġ< T > Ġthe Ġitem Ġtype Ċ Ġ*/ Ċ @ Flow Preview Ċ class ĠUn icast Work Subject < T > Ġ: ĠAbstract Flow < T >(), ĠSubject API < T > Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġterminated Ġ= ĠThrowable (" No Ġmore Ġelements ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġqueue Ġ= ĠConcurrent Linked Queue < T >() ĊĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġvar Ġterminal Ġ: ĠThrowable ? Ġ= Ġnull ĊĊĠĠĠ Ġval Ġres um able Ġ= ĠRes um able () ĊĊĠĠĠ Ġval Ġcurrent Ġ= ĠAtomic Reference < Flow Collector < T >>() ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcollect S af ely ( collector : ĠFlow Collector < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcurr Ġ= Ġcurrent . get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( curr Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Only Ġone Ġcollector Ġallowed ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( current . compare AndSet ( curr , Ġcollector )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġt Ġ= Ġterminal ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġv Ġ= Ġqueue . poll () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( t Ġ!= Ġnull Ġ&& Ġv Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent . get AndSet ( null ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( t Ġ!= Ġterminated ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġt ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( v Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . emit ( v ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent . get AndSet ( null ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġres um able . await () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit Error ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġterminal Ġ= Ġex ĊĠĠĠĠĠĠĠ Ġres um able . resume () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcomplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġterminal Ġ= Ġterminated ĊĠĠĠĠĠĠĠ Ġres um able . resume () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġhas Collectors (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcurr Ġ= Ġcurrent . get () ĊĠĠĠĠĠĠĠ Ġreturn Ġcurr Ġ!= Ġnull ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcollector Count (): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( has Collectors ()) Ġ 1 Ġelse Ġ 0 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit ( value : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġqueue . offer ( value ) ĊĠĠĠĠĠĠĠ Ġres um able . resume () ĊĠĠĠ Ġ} Ċ }