/* Ċ Ġ* Ġ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 Ġhu . ak arn ok d . kotlin . flow . impl . Sp sc Array Queue Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . flow . Flow Collector Ċ import Ġjava . util . concurrent . atomic . Atomic Long Ċ Ċ /** Ċ Ġ* ĠA Ġcollector Ġthat Ġcan Ġbuffer Ġa Ġfixed Ġnumber Ġof Ċ Ġ* Ġitems Ġand Ġdrain Ġit Ġwith Ġmuch Ġless Ġneed Ġto Ċ Ġ* Ġcoordinate Ġthe Ġres umption Ġper Ġitems . Ċ Ġ*/ Ċ open Ġclass ĠBuffer ing Res um able Collector < T > Ġ@ Suppress (" UN CHECK ED _ CAST ") Ġconstructor ( capacity : ĠInt ) Ġ: ĠRes um able () Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġqueue Ġ: ĠSp sc Array Queue < T > Ġ= ĠSp sc Array Queue ( capacity ) ĊĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġdone : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġvar Ġerror Ġ: ĠThrowable ? Ġ= Ġnull ĊĊĠĠĠ Ġprivate Ġval Ġavailable Ġ= ĠAtomic Long () ĊĊĠĠĠ Ġprivate Ġval Ġvalue Ready Ġ= ĠRes um able () ĊĊĠĠĠ Ġprivate Ġval Ġoutput Ġ: ĠArray < Any ?> Ġ= ĠArray ( 1 ) Ġ{ Ġnull Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġlimit Ġ: ĠInt Ġ= Ġcapacity Ġ- Ġ( capacity Ġshr Ġ 2 ) ĊĊĠĠĠ Ġ@ Volatile ĊĠĠĠ Ġprivate Ġvar Ġcancelled Ġ= Ġfalse ĊĊĠĠĠ Ġsuspend Ġfun Ġnext ( value : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ(! cancelled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( queue . offer ( value )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( available . get And Increment () Ġ== Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ready . resume () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġawait () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( cancelled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠC ancellation Exception () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġerror ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġerror Ġ= Ġex ĊĠĠĠĠĠĠĠ Ġdone Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġvalue Ready . resume () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcomplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġdone Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġvalue Ready . resume () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġdrain ( collector : ĠFlow Collector < T >, Ġon Crash : Ġ(( Buffer ing Res um able Collector < T >) Ġ-> ĠUnit )? Ġ= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġq Ġ= Ġqueue ĊĠĠĠĠĠĠĠ Ġval Ġavail Ġ= Ġavailable ĊĠĠĠĠĠĠĠ Ġval Ġready Ġ= Ġvalue Ready ĊĠĠĠĠĠĠĠ Ġvar Ġconsumed Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġval Ġlimit Ġ= Ġthis . limit . to Long () ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġd Ġ= Ġdone ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġe Ġ= Ġ! q . poll ( output ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( d Ġ&& Ġe ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġex Ġ= Ġerror ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ex Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! e ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . emit ( output [ 0 ] Ġas ĠT ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Crash ?. invoke ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancelled Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresume () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( consumed ++ Ġ== Ġlimit ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġavail . add AndGet (- consumed ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumed Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresume () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontinue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( avail . add AndGet (- consumed ) Ġ== Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresume () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġready . await () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumed Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }