/* Ċ Ġ* ĠCopyright Ġ 2 0 1 7 ĠAll an ĠY osh io ĠH ase g awa Ċ Ġ* Ċ Ġ* Ġ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 Ġio . temp o . internal Ċ Ċ import Ġandroid . annotation . Suppress Lint Ċ import Ġio . reactivex . Flow able Ċ import Ġio . reactivex . Single Ċ import Ġio . reactivex . functions . Bi Function Ċ import Ġio . reactivex . processors . Replay Processor Ċ import Ġio . reactivex . schedul ers . Schedulers Ċ import Ġio . temp o . Device Clock s Ċ import Ġio . temp o . Scheduler Ċ import Ġio . temp o . Storage Ċ import Ġio . temp o . Sync Retry Strategy Ċ import Ġio . temp o . Temp o Config Ċ import Ġio . temp o . Temp o Event Ċ import Ġio . temp o . Time Source Ċ import Ġio . temp o . Time Source Cache Ċ import Ġio . temp o . Time Source Wrapper Ċ import Ġio . temp o . schedul ers . No Op Scheduler Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ internal Ġclass ĠTemp o Instance ( ĊĠĠĠ Ġval Ġtime Sources : ĠList < Time Source >, ĊĠĠĠ Ġval Ġconfig : ĠTemp o Config , ĊĠĠĠ Ġprivate Ġval Ġstorage : ĠStorage , ĊĠĠĠ Ġprivate Ġval Ġdevice Clock s : ĠDevice Clock s , ĊĠĠĠ Ġprivate Ġval Ġscheduler : ĠScheduler ) Ġ{ ĊĊĠĠĠ Ġval Ġinitialized Ġget () Ġ= Ġactive Time Wrapper () Ġ!= Ġnull ĊĊĠĠĠ Ġprivate Ġvar Ġactive Time Source Name : ĠString ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġval Ġtime Wrappers Ġ= Ġmutable Map Of < String , ĠTime Source Wrapper >() ĊĊĠĠĠ Ġprivate Ġval Ġevents Subject Ġ= ĠRe play Processor . create With Time < Temp o Event >( ĊĠĠĠĠĠĠĠĠ 1 0 0 0 , ĠTimeUnit . MILLISECONDS , ĠS chedulers . io ()) ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire ( time Sources . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"' time Sources ' Ġmust Ġnot Ġbe Ġempty ." ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġrequire ( time Sources . map Ġ{ Ġit . config (). id Ġ}. distinct (). size Ġ== Ġtime Sources . size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" Duplicate Ġids Ġin Ġ' time Sources ' Ġaren 't Ġallowed ." ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfire Up First Sync Flow () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġobserve Events (): ĠFlow able < Temp o Event > Ġ= Ġevents Subject . on Back pressure Latest () ĊĊĠĠĠ Ġfun Ġnow (): ĠLong ? Ġ= Ġactive Time Wrapper ()?. now From Cache ( device Clock s . uptime ()) ĊĊĠĠĠ Ġfun Ġactive Time Wrapper (): ĠTime Source Wrapper ? Ġ= Ġactive Time Source Name ?. let Ġ{ Ġtime Wrappers [ it ] Ġ} ĊĊĠĠĠ Ġfun Ġsync Flow (): ĠFlow able < Temp o Event > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFlow able . from Iterable ( time Sources ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġsource Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Time ( source ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. timeout ( config . sync Timeout Ms , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Flow able () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġwrapper Ġ-> ĠTemp o Event . T SS ync Success ( wrapper ) Ġas ĠTemp o Event Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. start With ( Temp o Event . T SS ync Request ( source ) Ġas ĠTemp o Event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Return Ġ{ Ġerror Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdefault Msg Ġ= Ġ" Error Ġrequesting Ġtime Ġto Ġ'${ source . config (). id } '" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTemp o Event . T SS ync Failure ( source , Ġerror , Ġerror . message Ġ?: Ġdefault Msg ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. publish Ġ{ Ġflow Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġend Flow Ġ= Ġflow ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. buffer ( time Sources . size Ġ* Ġ 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. take ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . any Ġ{ Ġit Ġis ĠTemp o Event . T SS ync Success Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġhas Success Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġactive Tw Ġ= Ġactive Time Wrapper () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( has Success Ġ&& Ġactive Tw Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTemp o Event . Sync Success ( active Tw ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTemp o Event . Sync Fail () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinitialized Flow Ġ= ĠFlow able . from Callable Ġ{ Ġinitialized Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( it ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> ĠFlow able . just ( Temp o Event . Initialized ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠFlow able . empty () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . merge With ( end Flow ). concat With ( initialized Flow ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. start With ( Temp o Event . Sync Start ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( event ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event Ġis ĠTemp o Event . T SS ync Success ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġname Ġ= Ġevent . wrapper . time Source . config (). id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcache Ġ= Ġevent . wrapper . cache ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( time Wrappers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstorage . put Cache ( cache ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime Wrappers [ name ] Ġ= Ġevent . wrapper ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Active Time Wrapper () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( Temp o Event . Cache Saved ( cache )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. repeat When Ġ{ Ġcompleted Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġretry Flow Ġ= Ġsync Retry Str at Flow ( config . sync Retry Strategy ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompleted . zip With ( retry Flow , ĠBi Function Ġ{ Ġ_: ĠAny , Ġ_: ĠAny Ġ-> Ġinitialized Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. take While Ġ{ Ġ! it Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġrequest Time ( time Source : ĠTime Source ): ĠSingle < Time Source Wrapper > Ġ= ĊĠĠĠĠĠĠĠ Ġtime Source . request Time () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġreq Time Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcache Ġ= ĠTime Source Cache ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime Source Id Ġ= Ġtime Source . config (). id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġestimated Boot Time Ġ= Ġdevice Clock s . estimated Boot Time (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Device U ptime Ġ= Ġdevice Clock s . uptime (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Time Ġ= Ġreq Time ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTime Source Wrapper ( time Source , Ġcache ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġupdate Active Time Wrapper () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtop Priority Ġ= Ġtime Wrappers . values . max By Ġ{ Ġit . time Source . config (). priority Ġ} ĊĠĠĠĠĠĠĠ Ġactive Time Source Name Ġ= Ġtop Priority ?. time Source ?. config ()?. id ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġrestore Cache () Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġis Cache Valid ( cache : ĠTime Source Cache ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġestimated Boot Time Ġ= Ġdevice Clock s . estimated Boot Time () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcache Est imated Boot Time Ġ= Ġcache . estimated Boot Time ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠMath . abs ( cache Est imated Boot Time Ġ- Ġestimated Boot Time ) Ġ<= Ġ 5 0 0 0 L ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġtime Sources ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġsource Ġ-> Ġsource Ġto Ġstorage . getCache ( source . config (). id ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . second ?. let (:: is Cache Valid ) Ġ?: Ġfalse Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Each Ġ{ Ġit . second ?. let Ġ{ Ġevents Subject . onNext ( Temp o Event . Cache Restore d ( it )) Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . first . config (). id Ġto ĠTime Source Wrapper ( it . first , Ġit . second !! ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( time Wrappers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime Wrappers . put All ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Active Time Wrapper () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsetup Scheduler () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( scheduler Ġ! is ĠNo Op Scheduler ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( Temp o Event . Scheduler Setup Start ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler . setup () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( Temp o Event . Scheduler Setup Complete ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( Temp o Event . Scheduler Setup Failure ( e , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Error Ġwhile Ġsetting Ġup Ġscheduler .")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subject . onNext ( Temp o Event . Scheduler Setup Skip ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsync Retry Str at Flow ( strat : ĠSync Retry Strategy ): ĠFlow able < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( strat ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSync Retry Strategy . None Ġ-> ĠFlow able . empty () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSync Retry Strategy . Constant Interval Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinterval Ġ= ĠFlow able . interval ( strat . timer Ms , Ġstrat . interval Ms , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtries Ġ= ĠFlow able . range ( 1 , Ġstrat . retries ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlow able . zip < Long , ĠInt , ĠAny >( interval , Ġtries , ĠBi Function Ġ{ Ġ_, Ġ_ Ġ-> Ġ 0 Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSync Retry Strategy . Exp Backoff Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtries Ġ= ĠFlow able . range ( 1 , Ġstrat . retries ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinterval Ġ= Ġ{ Ġidx : ĠInt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbackoff Ġ= ĠMath . pow ( 2 . 0 , Ġidx . toDouble ()) Ġ* Ġstrat . multiplier ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtimer Ġ= Ġ( strat . timer Ms Ġ+ Ġbackoff ). to Long () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. coerce At Most ( strat . max Interval Ms ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlow able . timer ( timer , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtries . concat Map Ġ{ Ġidx Ġ-> Ġinterval ( idx ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress Lint (" Check Result ") ĊĠĠĠ Ġprivate Ġfun Ġfire Up First Sync Flow () Ġ{ ĊĠĠĠĠĠĠĠ ĠFlow able . just ( time Sources ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġevents Subject . onNext ( Temp o Event . Initializing ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġrestore Cache () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġsetup Scheduler () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġsync Flow () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ({}, Ġ{}, Ġ{}) ĊĠĠĠ Ġ} Ċ }