/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 Ġ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 . Ċ Ġ*/ Ċ Ċ package Ġandroidx . paging Ċ Ċ import Ġandroidx . annotation . Visible ForTesting Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . channels . Closed Send Channel Exception Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Mutable State Flow Ċ import Ġkotlinx . coroutines . flow . Sharing Started Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . flow . consume As Flow Ċ import Ġkotlinx . coroutines . flow . flow Of Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . on Completion Ċ import Ġkotlinx . coroutines . flow . on Each Ċ import Ġkotlinx . coroutines . flow . share In Ċ import Ġkotlinx . coroutines . flow . take While Ċ import Ġkotlinx . coroutines . flow . with Index Ċ import Ġkotlinx . coroutines . join All Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . sync . Mutex Ċ import Ġkotlinx . coroutines . sync . with Lock Ċ import Ġjava . util . Array Deque Ċ Ċ /** Ċ Ġ* ĠAn Ġintermediate Ġflow Ġproducer Ġthat Ġfl att ens Ġprevious Ġpage Ġevents Ġand Ġgives Ġany Ġnew Ġdownstream Ċ Ġ* Ġjust Ġthose Ġevents Ġinstead Ġof Ġthe Ġfull Ġhistory . Ċ Ġ*/ Ċ internal Ġclass ĠCached Page Event Flow < T Ġ: ĠAny >( ĊĠĠĠ Ġsrc : ĠFlow < Page Event < T >>, ĊĠĠĠ Ġscope : ĠCor outine Scope Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġpage Controller Ġ= ĠFlat ten ed Page Controller < T >() ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFlow < Page Event > Ġcan Ġbe Ġcollected Ġonly Ġonce Ġso Ġthis Ġgoes Ġfrom Ġactive Ġto Ġpassive Ġonly . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġactive Ġ= ĠMutable State Flow ( true ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠShared Ġflow Ġfor Ġupstream . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠNote Ġthat Ġshared Ġflows Ġcannot Ġbe Ġcompleted , Ġhence Ġwe Ġuse Ġa Ġcustom Ġwrapper Ġtype ĊĠĠĠĠ Ġ* Ġ[ Flatten ed Page Event Storage . Upstream Message ] Ġto Ġspecify Ġend Ġof Ġstream . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġshared Src Ġ= Ġactive . simple Flat Map Latest Ġ{ Ġactive Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( active ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsrc . with Index (). on Each ( page Controller :: record ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlat ten ed Page Event Storage . Upstream Message . Value ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Completion Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġclose Ġit Ġso Ġthat Ġwe Ġcan Ġdispatch Ġthe Ġcompleted Ġmessage . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis @ Cached Page Event Flow . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsend Ġa Ġcomplete Ġmessage Ġto Ġnotify Ġend Ġof Ġstream ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Of ( Flatten ed Page Event Storage . Upstream Message . Completed ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ}. share In ( ĊĠĠĠĠĠĠĠ Ġscope Ġ= Ġscope , ĊĠĠĠĠĠĠĠ Ġstarted Ġ= ĠSh aring Started . L az ily , Ġ// Ġnever Ġstop , Ġwe 'll Ġbe Ġclosed Ġmanually ĊĠĠĠĠĠĠĠ Ġ// ĠDon 't Ġreplay Ġfor Ġevents , Ġwe Ġhave Ġour Ġown Ġcond ensed Ġreplay Ġmechanism Ġthat Ġmerge s Ġevents . ĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġonly Ġreplay Ġonce Ġto Ġaccount Ġfor Ġthe ĠCompleted Ġterminal Ġevent . ĠIf Ġwe Ġdon 't , Ġshared Ġflow ĊĠĠĠĠĠĠĠ Ġ// Ġwould Ġnever Ġemit Ġany Ġvalue Ġhence Ġthe Ġupstream Ġwould Ġlook Ġas Ġif Ġit Ġis Ġopen Ġeven Ġthough ĊĠĠĠĠĠĠĠ Ġ// Ġit Ġis Ġnot . ĊĠĠĠĠĠĠĠ Ġ// Ġnote Ġthat Ġnot Ġreplay ing Ġdoes Ġnot Ġmean Ġvalues Ġare Ġnot Ġbuffered . ĠBuffer ing Ġstill ĊĠĠĠĠĠĠĠ Ġ// Ġhappens Ġas Ġlong Ġas Ġthe Ġsubscriber Ġhas Ġar rived Ġin Ġtime Ġbefore Ġthe Ġvalue Ġis Ġemitted . ĊĠĠĠĠĠĠĠ Ġ// Ġ[ down stream Flow ] Ġtakes Ġcare Ġof Ġordering Ġthem Ġbehind Ġa Ġmutex . ĊĠĠĠĠĠĠĠ Ġ// Ġhttps :// kotlin . github . io / kotlin x . coroutines / kotlin x - coroutines - core / kotlin x . coroutines . flow / share - in . html ĊĠĠĠĠĠĠĠ Ġreplay Ġ= Ġ 1 ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġactive . compare AndSet ( expect Ġ= Ġtrue , Ġupdate Ġ= Ġfalse ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġdownstream Flow Ġ= Ġsimple Channel Flow < Page Event < T >> Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġget Ġa Ġnew Ġsnapshot . Ġthis Ġwill Ġimmediately Ġhook Ġus Ġto Ġthe Ġupstream Ġchannel ĊĠĠĠĠĠĠĠ Ġval Ġsnapshot Ġ= Ġpage Controller . create Temporary Down stream () ĊĠĠĠĠĠĠĠ Ġvar Ġlast Received History Index Ġ= ĠInt . MIN _ VALUE ĊĠĠĠĠĠĠĠ Ġ// Ġfirst , Ġdispatch Ġeverything Ġin Ġthe Ġsnapshot ĊĠĠĠĠĠĠĠ Ġ// Ġthese Ġare Ġevents Ġthat Ġsync Ġus Ġto Ġthe Ġcurrent Ġstate Ġ+ Ġother Ġevents Ġthat Ġmight Ġbe ĊĠĠĠĠĠĠĠ Ġ// Ġemitted Ġby Ġthe Ġupstream Ġbetween Ġthe Ġtime Ġof Ġsnapshot Ġand Ġus Ġconnecting Ġto Ġthe Ġupstream ĊĠĠĠĠĠĠĠ Ġval Ġhistory Collection Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsnapshot . consume History (). collect Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Received History Index Ġ= Ġit . index ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend ( it . value ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġactive Stream Collection Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshared Src . take While Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit Ġ!== ĠFlat ten ed Page Event Storage . Upstream Message . Completed ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. on Completion Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġif Ġmain Ġupstream Ġfinishes , Ġmake Ġsure Ġto Ġclose Ġthe Ġhistory Ġstream ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġotherwise Ġit Ġmight Ġstay Ġopen Ġforever ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsnapshot . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. collect Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġgot Ġa Ġvalue Ġfrom Ġreal Ġstream , Ġclose Ġhistory ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsnapshot . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwait Ġfor Ġit Ġto Ġbe Ġdone Ġso Ġthat Ġall Ġevents Ġthere Ġare Ġdis patched ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhistory Collection . join () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnow Ġif Ġit Ġis Ġnew Ġenough , Ġemit Ġit , Ġotherwise , Ġskip Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it Ġis ĠFlat ten ed Page Event Storage . Upstream Message . Value <* >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . event . index Ġ> Ġlast Received History Index ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend ( it . event . value Ġas ĠPage Event < T >) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġjoin All ( active Stream Collection , Ġhistory Collection ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠThis Ġintermediate Ġclass Ġis Ġused Ġto Ġensure Ġnew Ġdownstream Ġdoes Ġnot Ġmiss Ġany Ġevents . Ċ Ġ* Ċ Ġ* ĠThere Ġis Ġa Ġrace Ġcondition Ġwhere , Ġwhile Ġit Ġis Ġcollecting Ġthe Ġinitial Ġstate , Ġthere Ġmight Ġbe Ġmore Ċ Ġ* Ġevents Ġdis patched Ġby Ġthe Ġreal Ġsource Ġwhich Ġwill Ġbe Ġmissed Ġbecause Ġnew Ġdownstream Ġis Ġnot Ċ Ġ* Ġsubscribed Ġto Ġthe Ġreal Ġupstream Ġyet . Ċ Ġ* Ċ Ġ* ĠFlat ten ed Page Controller Ġtracks Ġthese Ġtemporary Ġchannels Ġand Ġsends Ġeach Ġevent Ġinto Ġthem Ġas Ġwell Ċ Ġ* Ġso Ġthat Ġthey Ġare Ġnot Ġlost Ġuntil Ġthe Ġnew Ġdownstream Ġconnects Ġto Ġthe Ġreal Ġupstream . Ċ Ġ* ĠAfter Ġthat Ġconnection Ġis Ġestablished , Ġthis Ġchannel Ġis Ġclosed . Ċ Ġ* Ċ Ġ* ĠEach Ġvalue Ġis Ġindexed Ġto Ġavoid Ġsending Ġthe Ġsame Ġevent Ġmultiple Ġtimes . Ċ Ġ*/ Ċ private Ġclass ĠTemporary Down stream < T Ġ: ĠAny > Ġ{ ĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠList Ġof Ġevents Ġthat Ġsync Ġus Ġto Ġcurrent Ġstate Ġ+ Ġany Ġother Ġevents Ġthat Ġmight Ġar rive Ġwhile ĊĠĠĠĠ Ġ* Ġwe Ġare Ġdoing Ġthe Ġinitial Ġsync ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġhistory Channel : ĠChannel < Indexed Value < Page Event < T >>> Ġ= ĠChannel ( Channel . UN LIMIT ED ) ĊĊĠĠĠ Ġfun Ġconsume History () Ġ= Ġhistory Channel . consume As Flow () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠT ries Ġto Ġsend Ġthe Ġvalue Ġto Ġthe Ġhistory Ġchannel . ĊĠĠĠĠ Ġ* ĠReturns Ġfalse Ġif Ġsend Ġfails Ġ( e . g . Ġchannel Ġis Ġalready Ġclosed ) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġsend ( event : ĠIndexed Value < Page Event < T >> ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhistory Channel . send ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( closed : ĠClosed Send Channel Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġhistory Channel . close () ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġclass ĠFlat ten ed Page Controller < T Ġ: ĠAny > Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlist Ġ= ĠFlat ten ed Page Event Storage < T >() ĊĠĠĠ Ġprivate Ġvar Ġsnapshots Ġ= ĠlistOf < Temporary Down stream < T >>() ĊĠĠĠ Ġprivate Ġval Ġlock Ġ= ĠMutex () ĊĠĠĠ Ġprivate Ġvar Ġmax Event Index Ġ= Ġ- 1 ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠRecord Ġthe Ġevent . ĊĠĠĠĠ Ġ* ĠThis Ġsends Ġthe Ġevent Ġinto Ġstorage Ġbut Ġalso Ġinto Ġany Ġother Ġactive ĠTemporary Down stream . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġrecord ( event : ĠIndexed Value < Page Event < T >>) Ġ{ ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmax Event Index Ġ= Ġevent . index ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlist . add ( event . value ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsnapshots Ġ= Ġsnapshots . filter Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . send ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCreates Ġa Ġtemporary Ġdownstream Ġwhich Ġwill Ġhave Ġevents Ġthat Ġare Ġnecessary Ġto Ġsync Ġthe ĊĠĠĠĠ Ġ* Ġcurrent Ġstate Ġ+ Ġany Ġother Ġevents Ġthat Ġmight Ġar rive Ġwhile Ġthat Ġsync Ġis Ġin Ġprogress . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsuspend Ġfun Ġcreate Temporary Down stream (): ĠTemporary Down stream < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsend Ġthe Ġcurrent Ġstate Ġinto Ġthe Ġtemporary Ġdownstream Ġwhile Ġwe Ġare Ġinside Ġthe Ġmutex ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġensure Ġthat Ġwe Ġdon 't Ġprocess Ġany Ġnew Ġevents Ġuntil Ġwe Ġcopy Ġall Ġinitial Ġevents ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġinto Ġthe Ġchannel Ġof Ġtemporary Ġdownstream ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsnap Ġ= ĠTemporary Down stream < T >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcond ensed Ġevents Ġto Ġbring Ġdownstream Ġup Ġto Ġthe Ġcurrent Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcatch up Events Ġ= Ġlist . getAs Events () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġmake Ġsure Ġthe Ġindices Ġhere Ġmatch Ġthe Ġindices Ġcoming Ġfrom Ġupstream Ġso Ġthat Ġif Ġthe Ġsame ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġevent Ġshows Ġup Ġin Ġboth Ġplaces , Ġwe Ġwon 't Ġdispatch Ġit Ġtwice . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwe Ġwant Ġthe Ġlast Ġevent 's Ġindex Ġto Ġmatch Ġthe Ġlatest Ġevent Ġindex Ġwe 've Ġreceived Ġso ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthat Ġdownstream Ġcan Ġignore Ġthat Ġevent Ġif Ġthey Ġreceive Ġit Ġalso Ġfrom Ġthe Ġoriginal ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġupstream . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstart Event Index Ġ= Ġmax Event Index Ġ- Ġcatch up Events . size Ġ+ Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcatch up Events . forEach Indexed Ġ{ Ġindex , Ġpage Event Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsnap . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndexed Value ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġindex Ġ= Ġstart Event Index Ġ+ Ġindex , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Ġ= Ġpage Event ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsnap ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠKeep s Ġa Ġlist Ġof Ġpage Ġevents Ġand Ġcan Ġdispatch Ġthem Ġat Ġonce Ġas ĠPage Event Ġinstead Ġof Ġmultiple Ċ Ġ* Ġevents . Ċ Ġ* Ċ Ġ* ĠThere Ġis Ġno Ġsynchronization Ġin Ġthis Ġcode Ġso Ġit Ġshould Ġbe Ġused Ġwith Ġlocks Ġaround Ġif Ġnecessary . Ċ Ġ*/ Ċ @ Visible ForTesting ( otherwise Ġ= ĠVis ible ForTesting . PRIVATE ) Ċ internal Ġclass ĠFlat ten ed Page Event Storage < T Ġ: ĠAny > Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġplaceholders Before : ĠInt Ġ= Ġ 0 ĊĠĠĠ Ġprivate Ġvar Ġplaceholders After : ĠInt Ġ= Ġ 0 ĊĠĠĠ Ġprivate Ġval Ġpages Ġ= ĠArray Deque < Transform able Page < T >>() ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠNote Ġ- Ġthis Ġis Ġinitialized Ġwithout Ġremote Ġstate , Ġsince Ġwe Ġdon 't Ġknow Ġif Ġwe Ġhave Ġremote ĊĠĠĠĠ Ġ* Ġdata Ġonce Ġwe Ġstart Ġgetting Ġevents . ĠThis Ġis Ġfine , Ġsince Ġdownstream Ġneeds Ġto Ġhandle Ġthis ĊĠĠĠĠ Ġ* Ġanyway Ġ- Ġremote Ġstate Ġbeing Ġadded Ġafter Ġinitial , Ġempty , ĠP aging Data . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġload States Ġ= ĠMutable Load State Collection () ĊĠĠĠ Ġfun Ġadd ( event : ĠPage Event < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( event ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPage Event . Insert < T > Ġ-> Ġhandle Insert ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPage Event . Drop < T > Ġ-> Ġhandle Page Drop ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠPage Event . Load State Update < T > Ġ-> Ġhandle Load State Update ( event ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Page Drop ( event : ĠPage Event . Drop < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTODO : Ġinclude Ġstate Ġin Ġdrop Ġevent Ġfor Ġsimplicity , Ġinstead Ġof Ġreconstruct ing Ġbehavior . ĊĠĠĠĠĠĠĠ Ġ// Ġ ĠThis Ġallows Ġupstream Ġto Ġcontrol Ġhow Ġdrop Ġaffects Ġstates Ġ( e . g . Ġletting Ġdrop Ġaffect Ġboth ĊĠĠĠĠĠĠĠ Ġ// Ġ Ġremote Ġand Ġlocal ) ĊĠĠĠĠĠĠĠ Ġload States . set ( event . load Type , Ġfalse , ĠLoad State . Not Loading . Incomplete ) ĊĊĠĠĠĠĠĠĠ Ġwhen Ġ( event . load Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLoad Type . PRE PEND Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders Before Ġ= Ġevent . placeholders Remaining ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( event . page Count ) Ġ{ Ġpages . remove First () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLoad Type . APPEND Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders After Ġ= Ġevent . placeholders Remaining ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( event . page Count ) Ġ{ Ġpages . remove Last () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalArgumentException (" Page Ġdrop Ġtype Ġmust Ġbe Ġprepend Ġor Ġappend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Insert ( event : ĠPage Event . Insert < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġload States . set ( event . combined Load States ) ĊĠĠĠĠĠĠĠ Ġwhen Ġ( event . load Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLoad Type . REFRESH Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpages . clear () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders After Ġ= Ġevent . placeholders After ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders Before Ġ= Ġevent . placeholders Before ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpages . addAll ( event . pages ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLoad Type . PRE PEND Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders Before Ġ= Ġevent . placeholders Before ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( event . pages . size Ġ- Ġ 1 Ġdown To Ġ 0 ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpages . add First ( event . pages [ it ]) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLoad Type . APPEND Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders After Ġ= Ġevent . placeholders After ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpages . addAll ( event . pages ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Load State Update ( event : ĠPage Event . Load State Update < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġload States . set ( event . load Type , Ġevent . from Mediator , Ġevent . load State ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget As Events (): ĠList < Page Event < T >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġevents Ġ= Ġmutable ListOf < Page Event < T >>() ĊĠĠĠĠĠĠĠ Ġif Ġ( pages . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevents . add ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPage Event . Insert . Refresh ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpages Ġ= Ġpages . toList (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders Before Ġ= Ġplaceholders Before , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders After Ġ= Ġplaceholders After , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcombined Load States Ġ= Ġload States . snapshot () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġload States . forEach Ġ{ Ġtype , Ġfrom Mediator , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Page Event . Load State Update . can Dispatch Without Insert ( state , Ġfrom Mediator )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents . add ( Page Event . Load State Update ( type , Ġfrom Mediator , Ġstate )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġevents ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠMessage Ġtype Ġfor Ġthe Ġshared Ġstream ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġsealed Ġclass ĠUp stream Message Ġ{ ĊĠĠĠĠĠĠĠ Ġobject ĠCompleted Ġ: ĠUp stream Message () ĊĠĠĠĠĠĠĠ Ġclass ĠValue < T Ġ: ĠAny >( val Ġevent : ĠIndexed Value < Page Event < T >>) Ġ: ĠUp stream Message () ĊĠĠĠ Ġ} Ċ } Ċ