package Ġcom . co op src . x android . downloader . helper Ċ Ċ import Ġcom . co op src . x android . downloader . utils . Constants Ċ import Ġcom . co op src . x android . downloader . utils . Downloader Utils Ċ import Ġcom . co op src . x android . downloader . utils . Logger Ċ import Ġio . reactivex . Back pressure Strategy Ċ import Ġio . reactivex . Flow able Ċ import Ġio . reactivex . Maybe Ċ import Ġio . reactivex . dis pos ables . Disposable Ċ import Ġio . reactivex . processors . Behavior Processor Ċ import Ġio . reactivex . processors . Flow able Processor Ċ import Ġio . reactivex . schedul ers . Schedulers Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ /** Ċ Ġ* ĠCreated Ġby Ġt ing ku o . Ċ Ġ* ĠDate : Ġ 2 0 1 8 - 0 8 - 2 0 Ċ Ġ* ĠTime : Ġ 1 7 : 3 8 Ċ Ġ*/ Ċ class ĠWork Queue < T >( ĊĠĠĠ Ġprivate Ġval Ġperiod : ĠLong , ĊĠĠĠ Ġprivate Ġval Ġunit : ĠTimeUnit Ġ= ĠTimeUnit . SECONDS , ĊĠĠĠ Ġprivate Ġval Ġaction : Ġ( data : ĠT ) Ġ-> ĠUnit Ċ ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġtag Ġ= Ġ" Work Queue " ĊĊĠĠĠ Ġprivate Ġval Ġqueue : ĠQueue < T > Ġ= ĠLinkedList < T >() ĊĊĠĠĠ Ġprivate Ġval Ġaction Processor : ĠFlow able Processor < T > Ġ= ĠBehavior Processor . create < T >(). to Serialized () ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġaction Flow able : ĠFlow able < T > ĊĠĠĠ Ġprivate Ġvar Ġaction Disposable : ĠDis posable ? Ġ= Ġnull ĊĊĠĠĠ Ġprivate Ġvar Ġis Complete : ĠBoolean Ġ= Ġfalse ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ ĠMaybe . create < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Flow able () ĊĠĠĠĠĠĠĠ Ġ}. subscribe On ( Schedulers . new Thread ()). do OnError Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLogger . e ( tag , Ġ" init Ġerror : Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ}. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Flow able () Ġ{ ĊĠĠĠĠĠĠĠ Ġaction Flow able Ġ= ĠFlow able . create < Any >({ Ġemitter Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ(! is Complete Ġ|| Ġ! queue . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onNext ( System . currentTimeMillis ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onComplete () ĊĠĠĠĠĠĠĠ Ġ}, ĠBack pressure Strategy . BUFFER ). sample ( period , Ġunit ). do On Subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLogger . i ( tag , Ġ" do On Subscribe : ĠWork Queue ĠCreated ") ĊĠĠĠĠĠĠĠ Ġ}. subscribe On ( Schedulers . io ()). do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLogger . i ( tag , Ġ" do On Next : Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ}. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdo Action ( poll ()) ĊĠĠĠĠĠĠĠ Ġ}. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Processor . onNext ( it ) ĊĠĠĠĠĠĠĠ Ġ}. do OnError Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Processor . onError ( it ) ĊĠĠĠĠĠĠĠ Ġ}. do On Complete Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Processor . onComplete () ĊĠĠĠĠĠĠĠ Ġ}. do Finally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Disposable Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget Flow able (): ĠFlow able < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction Processor ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġoffer ( element : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġqueue . offer ( element ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġpoll (): ĠT Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġqueue . poll () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdo Action ( element : ĠT ): ĠFlow able < out ĠT > Ġ{ ĊĠĠĠĠĠĠĠ ĠLogger . i ( tag , Ġ" do Action : Ġ$ element ") ĊĊĠĠĠĠĠĠĠ Ġaction ( element ) ĊĊĠĠĠĠĠĠĠ Ġreturn ĠFlow able . just ( element ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstart (): ĠMaybe < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMaybe . create < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . on Success ( Constants . any ) ĊĠĠĠĠĠĠĠ Ġ}. subscribe On ( Schedulers . new Thread ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsubscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( action Disposable Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Disposable Ġ= Ġaction Flow able . subscribe () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstop (): ĠMaybe < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMaybe . create < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġun Subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . on Success ( Constants . any ) ĊĠĠĠĠĠĠĠ Ġ}. subscribe On ( Schedulers . new Thread ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġun Subscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( action Disposable Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠDown loader Utils . dispose ( action Disposable ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Disposable Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġset Complete () Ġ{ ĊĠĠĠĠĠĠĠ Ġis Complete Ġ= Ġtrue ĊĠĠĠ Ġ} Ċ }