package Ġcom . jak ew h art on . sdk search . search . presenter Ċ Ċ import Ġcom . jak ew h art on . rx relay 2 . Publish Relay Ċ import Ġcom . jak ew h art on . sdk search . store . Item Ċ import Ġcom . jak ew h art on . sdk search . store . Item Store Ċ import Ġcom . jak ew h art on . sdk search . sync . Item Synchron izer Ċ import Ġio . reactivex . Observable Ċ import Ġkotlinx . coroutines . experimental . Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . experimental . Coroutine Start . UN DISPATCH ED Ċ import Ġkotlinx . coroutines . experimental . Job Ċ import Ġkotlinx . coroutines . experimental . channels . Con fl ated Broadcast Channel Ċ import Ġkotlinx . coroutines . experimental . channels . Receive Channel Ċ import Ġkotlinx . coroutines . experimental . launch Ċ import Ġkotlinx . coroutines . experimental . rx 2 . open Subscription Ċ import Ġkotlinx . coroutines . experimental . select s . select Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġjava . util . function . Consumer Ċ Ċ class ĠSearch Presenter ( ĊĠ Ġprivate Ġval Ġcontext : ĠCor outine Dispatcher , ĊĠ Ġprivate Ġval Ġstore : ĠItem Store , ĊĠ Ġprivate Ġval Ġsynchron izer : ĠItem Synchron izer Ċ ) Ġ{ ĊĠ Ġprivate Ġval Ġ_ models Ġ= ĠCon fl ated Broadcast Channel < Model >() ĊĠ Ġval Ġmodels : ĠReceive Channel < Model > Ġget () Ġ= Ġ_ models . open Subscription () ĊĊĠ Ġprivate Ġval Ġ_ events Ġ= ĠPublish Relay . create < Event >() ĊĠ Ġval Ġevents : ĠConsumer < Event > Ġget () Ġ= ĠConsumer Ġ{ Ġ_ events . accept ( it ) Ġ} ĊĊĠ Ġfun Ġstart (): ĠJob Ġ{ ĊĠĠĠ Ġval Ġitem Count Ġ= Ġstore . count (). open Subscription () ĊĊĠĠĠ Ġval Ġquery Items Ġ= Ġ_ events ĊĠĠĠĠĠĠĠ Ġ. of Type < Event . Query Changed >() ĊĠĠĠĠĠĠĠ Ġ. map ( Event . Query Changed :: query ) ĊĠĠĠĠĠĠĠ Ġ. distinct Until Changed () ĊĠĠĠĠĠĠĠ Ġ. switch Map Ġ{ Ġquery Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġval Ġresults Ġ= Ġif Ġ( query . isBlank ()) ĠObservable . just ( emptyList ()) ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġstore . query Items ( query ). delay Subscription ( 2 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĊĠĠĠĠĠĠĠĠĠ Ġresults . map Ġ{ ĠModel . Query Results ( query , Ġit ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ. open Subscription () ĊĊĠĠĠ Ġval Ġclear Sync Status Ġ= Ġ_ events ĊĠĠĠĠĠĠĠ Ġ. of Type < Event . Clear Sync Status >() ĊĠĠĠĠĠĠĠ Ġ. open Subscription () ĊĊĠĠĠ Ġval Ġjob Ġ= Ġlaunch ( context , ĠUN DISPATCH ED ) Ġ{ ĊĠĠĠĠĠ Ġvar Ġmodel Ġ= ĠModel () ĊĠĠĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠ Ġmodel Ġ= Ġselect Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġitem Count . on Receive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmodel . copy ( count Ġ= Ġit ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġquery Items . on Receive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmodel . copy ( query Results Ġ= Ġit ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġsynchron izer . state . on Receive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfailed Ġ= Ġit . count Ġ{ Ġit . failed Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġin Flight Ġ= Ġit . size Ġ- Ġfailed ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmodel . copy ( sync Status Ġ= ĠModel . Sync Status ( in Flight , Ġfailed )) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġclear Sync Status . on Receive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmodel . copy ( sync Status Ġ= ĠModel . Sync Status ( 0 , Ġ 0 )) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ_ models . offer ( model ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsynchron izer . force Sync () ĊĊĠĠĠ Ġreturn Ġjob ĊĠ Ġ} ĊĊĠ Ġsealed Ġclass ĠEvent Ġ{ ĊĠĠĠ Ġdata Ġclass ĠQuery Changed ( val Ġquery : ĠString ): ĠEvent () ĊĠĠĠ Ġobject ĠClear Sync Status Ġ: ĠEvent () ĊĠ Ġ} ĊĊĠ Ġdata Ġclass ĠModel ( ĊĠĠĠĠĠ Ġval Ġcount : ĠLong Ġ= Ġ 0 , ĊĠĠĠĠĠ Ġval Ġquery Results : ĠQuery Results Ġ= ĠQuery Results (), ĊĠĠĠĠĠ Ġval Ġsync Status : ĠSync Status Ġ= ĠSync Status () ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġdata Ġclass ĠQuery Results ( ĊĠĠĠĠĠĠĠ Ġval Ġquery : ĠString Ġ= Ġ"", ĊĠĠĠĠĠĠĠ Ġval Ġitems : ĠList < Item > Ġ= Ġempty List () ĊĠĠĠ Ġ) ĊĠĠĠ Ġdata Ġclass ĠSync Status ( ĊĠĠĠĠĠĠĠ Ġval Ġin Flight : ĠInt Ġ= Ġ 0 , ĊĠĠĠĠĠĠĠ Ġval Ġfailed : ĠInt Ġ= Ġ 0 ĊĠĠĠ Ġ) ĊĠ Ġ} ĊĊĠ Ġprivate Ġinline Ġfun Ġ< re ified ĠT > ĠObservable <* >. of Type (): ĠObservable < T > Ġ= Ġof Type ( T :: class . java ) Ċ } Ċ