/* Ċ Ġ* ĠCopyright Ġ 2 0 2 1 ĠThe ĠAndroid ĠOpen ĠSource ĠProject Ċ Ġ* Ċ Ġ* Ġ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 . Ċ Ġ*/ Ċ Ċ @ file : Restrict To ( Restrict To . Scope . LIBRARY ) Ċ Ċ package Ġandroidx . paging Ċ Ċ import Ġandroidx . annotation . Restrict To Ċ import Ġandroidx . paging . Combine Source . INITIAL Ċ import Ġandroidx . paging . Combine Source . OTHER Ċ import Ġandroidx . paging . Combine Source . RECE IVER Ċ import Ġkotlinx . coroutines . Comp letable Deferred Ċ import Ġkotlinx . coroutines . Job Ċ import Ġkotlinx . coroutines . channels . Send Channel Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Flow Collector Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . flow . collect Latest Ċ import Ġkotlinx . coroutines . flow . emit All Ċ import Ġkotlinx . coroutines . flow . flow Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . sync . Mutex Ċ import Ġkotlinx . coroutines . sync . with Lock Ċ import Ġkotlinx . coroutines . yield Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ Ċ /** Ċ Ġ* ĠThis ĠFile Ġincludes Ġcustom Ġflow Ġoperators Ġthat Ġwe Ġimplement Ġto Ġavoid Ġusing Ġexperimental ĠAPIs Ċ Ġ* Ġfrom Ġcor outines . ĠEvent ually , Ġthis Ġfile Ġshould Ġbe Ġdeleted Ġonce Ġthose ĠAPIs Ġbecome Ġstable . Ċ Ġ*/ Ċ Ċ private Ġval ĠNULL Ġ= ĠAny () Ċ Ċ /** Ċ Ġ* ĠTemporary Ġ` scan ` Ġoperator Ġon ĠFlow Ġwithout Ġexperimental ĠAPIs . Ċ Ġ*/ Ċ internal Ġfun Ġ< T , ĠR > ĠFlow < T >. simple Scan ( ĊĠĠĠ Ġinitial : ĠR , ĊĠĠĠ Ġoperation : Ġsuspend Ġ( accumulator : ĠR , Ġvalue : ĠT ) Ġ-> ĠR Ċ ): ĠFlow < R > Ġ= Ġflow Ġ{ ĊĠĠĠ Ġvar Ġaccumulator : ĠR Ġ= Ġinitial ĊĠĠĠ Ġemit ( accumulator ) ĊĠĠĠ Ġcollect Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠ Ġaccumulator Ġ= Ġoperation ( accumulator , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġemit ( accumulator ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠTemporary Ġ` running Reduce ` Ġoperator Ġon ĠFlow Ġwithout Ġexperimental ĠAPIs . Ċ Ġ*/ Ċ internal Ġfun Ġ< T > ĠFlow < T >. simple Running Reduce ( ĊĠĠĠ Ġoperation : Ġsuspend Ġ( accumulator : ĠT , Ġvalue : ĠT ) Ġ-> ĠT Ċ ): ĠFlow < T > Ġ= Ġflow Ġ{ ĊĠĠĠ Ġvar Ġaccumulator : ĠAny ? Ġ= ĠNULL ĊĠĠĠ Ġcollect Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠ Ġaccumulator Ġ= Ġif Ġ( accumulator Ġ=== ĠNULL ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoperation ( accumulator Ġas ĠT , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġemit ( accumulator Ġas ĠT ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠThis Ġis Ġa Ġsimilar Ġimplementation Ġto Ġtransform Latest Ġusing Ġa Ġchannel ĠFlow . Ċ Ġ*/ Ċ internal Ġfun Ġ< T , ĠR > ĠFlow < T >. simple Transform Latest ( ĊĠĠĠ Ġtransform : Ġsuspend ĠFlow Collector < R >. ( value : ĠT ) Ġ-> ĠUnit Ċ ): ĠFlow < R > Ġ= Ġsimple Channel Flow Ġ{ ĊĠĠĠ Ġval Ġorigin Ġ= Ġthis @ simple Transform Latest ĊĠĠĠ Ġval Ġcollector Ġ= ĠChannel Flow Collector ( this @ simple Channel Flow ) ĊĠĠĠ Ġorigin . collect Latest Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠ Ġcollector . transform ( value ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* Ġflat Map Latest Ġwithout Ġexperimental ĠAPIs Ċ Ġ*/ Ċ internal Ġinline Ġfun Ġ< T , ĠR > ĠFlow < T >. simple Flat Map Latest ( ĊĠĠĠ Ġcross inline Ġtransform : Ġsuspend Ġ( value : ĠT ) Ġ-> ĠFlow < R > Ċ ): ĠFlow < R > Ġ= Ġsimple Transform Latest Ġ{ Ġemit All ( transform ( it )) Ġ} Ċ Ċ /** Ċ Ġ* Ġmap Latest Ġwithout Ġexperimental ĠAPIs Ċ Ġ*/ Ċ internal Ġinline Ġfun Ġ< T , ĠR > ĠFlow < T >. simple Map Latest ( ĊĠĠĠ Ġcross inline Ġtransform : Ġsuspend Ġ( value : ĠT ) Ġ-> ĠR Ċ ): ĠFlow < R > Ġ= Ġsimple Transform Latest Ġ{ Ġemit ( transform ( it )) Ġ} Ċ Ċ internal Ġclass ĠChannel Flow Collector < T >( ĊĠĠĠ Ġval Ġchannel : ĠSend Channel < T > Ċ ) Ġ: ĠFlow Collector < T > Ġ{ ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit ( value : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġchannel . send ( value ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠSimilar Ġto Ġ[ kotlin x . coroutines . flow . combine ], Ġexcept Ġit Ġnever Ġbatches Ġreads Ġfrom Ġits ĠFlow s , Ġso Ċ Ġ* Ġ[ transform ] Ġis Ġalways Ġguaranteed Ġto Ġget Ġcalled Ġfor Ġevery Ġemission Ġfrom Ġeither ĠFlow Ġafter Ġthe Ċ Ġ* Ġinitial Ġcall Ġ( which Ġwaits Ġfor Ġthe Ġfirst Ġemission Ġfrom Ġboth ĠFlow s ). Ċ Ġ* Ċ Ġ* ĠThe Ġemission s Ġfor Ġboth ĠFlow s Ġare Ġalso Ġguaranteed Ġto Ġget Ġbuffered , Ġso Ġif Ġone ĠFlow Ġemits Ċ Ġ* Ġmultiple Ġtimes Ġbefore Ġthe Ġother Ġdoes , Ġ[ transform ] Ġwill Ġget Ġcalled Ġfor Ġeach Ġemission Ġfrom Ġthe Ċ Ġ* Ġfirst ĠFlow Ġinstead Ġof Ġjust Ġonce Ġwith Ġthe Ġlatest Ġvalues . Ċ Ġ* Ċ Ġ* Ġ@ param Ġtransform ĠThe Ġtransform Ġto Ġapply Ġon Ġeach Ġupdate . ĠThis Ġis Ġfirst Ġcalled Ġafter Ġawait ing Ġan Ċ Ġ* Ġinitial Ġemission Ġfrom Ġboth ĠFlow s , Ġand Ġthen Ġis Ġguaranteed Ġto Ġbe Ġcalled Ġfor Ġevery Ġemission Ġfrom Ċ Ġ* Ġeither ĠFlow . Ċ Ġ* Ċ Ġ* ĠFor Ġconvenience , Ġ[ Combine Source ] Ġis Ġalso Ġpassed Ġto Ġthe Ġtransform , Ġwhich Ġindicates Ġthe Ċ Ġ* Ġorigin Ġof Ġthe Ġupdate Ġwith Ġthe Ġfollowing Ġpossible Ġvalues : Ċ Ġ* ĠĠ Ġ* Ġ[ INITIAL ]: ĠInitial Ġemission Ġfrom Ġboth ĠFlow s Ċ Ġ* ĠĠ Ġ* Ġ[ RECE IVER ]: ĠTrigger ed Ġby Ġnew Ġemission Ġfrom Ġreceiver Ċ Ġ* ĠĠ Ġ* Ġ[ OTHER ]: ĠTrigger ed Ġby Ġnew Ġemission Ġfrom Ġ[ other Flow ] Ċ Ġ*/ Ċ internal Ġsuspend Ġinline Ġfun Ġ< T 1 , ĠT 2 , ĠR > ĠFlow < T 1 >. combine Without Batch ing ( ĊĠĠĠ Ġother Flow : ĠFlow < T 2 >, ĊĠĠĠ Ġcross inline Ġtransform : Ġsuspend Ġ( T 1 , ĠT 2 , Ġupdate From : ĠCombine Source ) Ġ-> ĠR , Ċ ): ĠFlow < R > Ġ{ ĊĠĠĠ Ġreturn Ġsimple Channel Flow Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġincomplete Flows Ġ= ĠAtomicInteger ( 2 ) ĊĠĠĠĠĠĠĠ Ġval Ġun bat ched Flow Combin er Ġ= ĠU nb atched Flow Combin er < T 1 , ĠT 2 > Ġ{ Ġt 1 , Ġt 2 , Ġupdate From Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend ( transform ( t 1 , Ġt 2 , Ġupdate From )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġparent Job Ġ= ĠJob () ĊĠĠĠĠĠĠĠ ĠarrayOf ( this @ combine Without Batch ing , Ġother Flow ). forEach Indexed Ġ{ Ġindex , Ġflow Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( parent Job ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . collect Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun bat ched Flow Combin er . onNext ( index , Ġvalue ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠMake Ġthis Ġmore Ġfair , Ġgiving Ġthe Ġother Ġflow Ġa Ġchance Ġto Ġemit . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġyield () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( in complete Flows . decrement AndGet () Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclose () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġawait Close Ġ{ Ġparent Job . cancel () Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠHelper Ġclass Ġfor Ġ[ Un bat ched Flow Combin er ], Ġwhich Ġhandles Ġdispatch ing Ġthe Ġcombined Ġvalues Ġin Ġthe Ċ Ġ* Ġcorrect Ġorder , Ġand Ġwith Ġ[ Combine Source ]. Ċ Ġ* Ċ Ġ* ĠNOTE : ĠThis Ġimplementation Ġrelies Ġon Ġthe Ġfact Ġthat Ġ[ onNext ] Ġis Ġcalled Ġin - order Ġfor Ġemission s Ċ Ġ* Ġfrom Ġthe Ġsame ĠFlow . ĠThis Ġmeans Ġthat Ġconcurrent ly Ġcalling Ġ[ onNext ] Ġwith Ġthe Ġsame Ġindex Ġwill Ġnot Ċ Ġ* Ġwork . Ċ Ġ* Ċ Ġ* Ġ@ see Ġcombine Without Batch ing Ċ Ġ*/ Ċ internal Ġclass ĠU nb atched Flow Combin er < T 1 , ĠT 2 >( ĊĠĠĠ Ġprivate Ġval Ġsend : Ġsuspend Ġ( t 1 : ĠT 1 , Ġt 2 : ĠT 2 , Ġupdate From : ĠCombine Source ) Ġ-> ĠUnit Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġinitial Dis patched Ġ= ĠComp letable Deferred < Unit >() ĊĠĠĠ Ġprivate Ġval Ġlock Ġ= ĠMutex () ĊĠĠĠ Ġprivate Ġval Ġvalue Received Ġ= ĠArray ( 2 ) Ġ{ ĠComp letable Deferred < Unit >() Ġ} ĊĠĠĠ Ġprivate Ġval Ġvalues Ġ= ĠArray < Any ? >( 2 ) Ġ{ ĠNULL Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġon Next ( index : ĠInt , Ġvalue : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠAllow Ġthe Ġfirst Ġvalue Ġto Ġdispatch Ġimmediately , Ġbut Ġfor Ġsubsequent Ġvalues , Ġwe Ġshould ĊĠĠĠĠĠĠĠ Ġ// Ġwait Ġuntil Ġthe Ġother Ġflow Ġemits , Ġso Ġthat Ġwe Ġdon 't Ġoverwrite Ġthe Ġprevious Ġvalue . ĊĠĠĠĠĠĠĠ Ġif Ġ( value Received [ index ]. is Completed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNOTE : ĠWe Ġuse Ġa Ġseparate ĠComp letable Ġhere Ġbecause Ġjust Ġawait ing ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġvalue Received [ 1 Ġ- Ġindex ] Ġcould Ġpotentially Ġallow Ġmultiple Ġcalls Ġto Ġon Next Ġfrom Ġthe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsame ĠFlow Ġto Ġtrigger Ġout Ġof Ġorder . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Dis patched . await () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Received [ index ]. complete ( Unit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġis Initial Ġ= Ġvalues . any Ġ{ Ġit Ġ=== ĠNULL Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalues [ index ] Ġ= Ġvalue ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( values . none Ġ{ Ġit Ġ=== ĠNULL Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġupdate From Ġ= Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis Initial Ġ-> ĠINITIAL ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġindex Ġ== Ġ 0 Ġ-> ĠRE CE IVER ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠOTHER ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend ( values [ 0 ] Ġas ĠT 1 , Ġvalues [ 1 ] Ġas ĠT 2 , Ġupdate From ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Dis patched . complete ( Unit ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠUsed Ġto Ġindicate Ġwhich ĠFlow Ġemission Ġtriggered Ġthe Ġtransform Ġblock Ġin Ġ[ combine Without Batch ing ]. Ċ Ġ* Ċ Ġ* Ġ@ see Ġcombine Without Batch ing Ċ Ġ*/ Ċ internal Ġenum Ġclass ĠCombine Source Ġ{ ĊĠĠĠ ĠINITIAL , ĊĠĠĠ ĠRE CE IVER , ĊĠĠĠ ĠOTHER , Ċ }