package Ġse map . rx . view model Ċ Ċ import Ġandroidx . lifecycle . ViewModel Ċ import Ġio . reactivex . rx java 3 . core . Notification Ċ import Ġio . reactivex . rx java 3 . core . Observable Ċ import Ġio . reactivex . rx java 3 . core . Scheduler Ċ import Ġio . reactivex . rx java 3 . functions . Bi Function Ċ import Ġio . reactivex . rx java 3 . schedul ers . Schedulers Ċ import Ġio . reactivex . rx java 3 . subjects . Behavior Subject Ċ import Ġio . reactivex . rx java 3 . subjects . Publish Subject Ċ import Ġio . reactivex . rx java 3 . subjects . Replay Subject Ċ import Ġio . reactivex . rx java 3 . subjects . Un icast Subject Ċ import Ġse map . rx . view model . Action Execution Mode .* Ċ Ċ abstract Ġclass ĠRx ViewModel < A , ĠS >: ĠViewModel () Ġ{ ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġoutputs Ġ( Observable s ) Ġof Ġthe ĠRx ViewModel ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġcurrent State : ĠS ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġstate Behavior Subject . value Ġ?: Ġcreate InitialState () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof ĠState , Ġit Ġwill Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġstate Observable : ĠObservable < S > Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . wrapper Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . is Complete ) Ġaction On Complete Subject . onNext ( it . action And State ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction On Next Subject . onNext ( it . action And State ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ! it . is Complete Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . action And State . state Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. start With ( Observable . from Callable Ġ{ Ġcreate InitialState () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. replay ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. auto Connect () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof Ġloading , Ġit Ġwill Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġloading Observable : ĠObservable < Boolean > Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġloading Count ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . second Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Boolean Observable () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof ĠAction And State , Ġit Ġdoes ĠNOT Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġaction On Next Observable : ĠObservable < Action And State < A , ĠS >> Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . action On Next Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Latest From ( state Observable , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBi Function < Action And State < A , ĠS >, ĠS , ĠAction And State < A , ĠS >> Ġ{ Ġans , Ġ_ Ġ-> Ġans Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof ĠAction And State , Ġit Ġdoes ĠNOT Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġaction On Complete Observable : ĠObservable < Action And State < A , ĠS >> Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . action On Complete Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Latest From ( state Observable , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBi Function < Action And State < A , ĠS >, ĠS , ĠAction And State < A , ĠS >> Ġ{ Ġans , Ġ_ Ġ-> Ġans Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof ĠAction And State , Ġit Ġdoes ĠNOT Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġaction Error Observable : ĠObservable < Action And Error < A >> Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġerror Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Latest From ( state Observable , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBi Function < Action And Error < A >, ĠS , ĠAction And Error < A >> Ġ{ Ġans , Ġ_ Ġ-> Ġans Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠObservable Ġof ĠError , Ġit Ġdoes ĠNOT Ġreplay ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġval Ġerror Observable : ĠObservable < Throwable > Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . action Error Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġ(_, Ġerror ) Ġ-> Ġerror Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġinputs Ġ( the Ġpublish Event Ġsubjects Ġthat Ġaccepts Ġactions ) Ġof Ġthe ĠRx ViewModel ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġconcat E ager Action Subject Ġ= ĠUn icast Subject . create < Def err able Action >(). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġconcurrent Action Subject Ġ= ĠUn icast Subject . create < A >(). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġsequential Action Subject Ġ= ĠUn icast Subject . create < List < A >> (). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġswitch Map Latest Action Subject Ġ= ĠUn icast Subject . create < A >(). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġdeferred Action Subject Ġ= ĠPublish Subject . create < A >(). to Serialized () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠInternal Ġdata ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġstate Behavior Subject Ġ= ĠBehavior Subject . create < S >() ĊĠĠĠ Ġprivate Ġval Ġerror Subject Ġ= ĠPublish Subject . create < Action And Error < A >> (). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġloading Count Ġ= ĠRe play Subject . create < Pair < A , ĠInt >> (). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġaction On Next Subject Ġ= ĠPublish Subject . create < Action And State < A , ĠS >> (). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġaction On Complete Subject Ġ= ĠPublish Subject . create < Action And State < A , ĠS >> (). to Serialized () ĊĠĠĠ Ġprivate Ġval Ġwrapper Observable : ĠObservable < An s Wrapper > Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġscheduler Ġ= Ġdefault Scheduler () ĊĠĠĠĠĠĠĠ Ġval Ġsequential Ġ= Ġsequential Action Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat Map Ġ{ Ġlist Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . from Iterable ( list ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat Map Ġ{ Ġexecute And Combine ( it , Ġtrue ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Resume Next Ġ{ ĠObservable . empty < An s Wrapper >() Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġconcurrent Ġ= Ġconcurrent Action Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġexecute And Combine ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġconcat E ager Ġ= Ġconcat E ager Action Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat Map E ager Ġ{ Ġexecute And Combine ( it . action , Ġthrow Error Ġ= Ġfalse , Ġdeferred Action Ġ= Ġit . is Deferred ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġflat Map Latest Ġ= Ġswitch Map Latest Action Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. switch Map Ġ{ Ġexecute And Combine ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġdeferred Ġ= Ġdeferred Action Subject ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( scheduler ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat Map Ġ{ Ġexecute And Combine ( it ) Ġ} ĊĊĠĠĠĠĠĠĠ ĠObservable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. merge ( listOf ( sequential , Ġconcat E ager , Ġconcurrent , Ġflat Map Latest , Ġdeferred )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. publish () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. auto Connect () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ return Ġthe Ġinitial Ġstate Ġof Ġthe ĠViewModel ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġabstract Ġfun Ġcreate InitialState (): ĠS ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThis Ġis Ġthe Ġmethod Ġto Ġtell Ġthe ĠRx ViewModel Ġ" how Ġto Ġchange Ġthe Ġcurrent Ġstate " Ġfor Ġa Ġgiven Ġaction . ĊĠĠĠĠ Ġ* Ġ@ param Ġaction Ġthe Ġaction Ġneeded Ġto Ġbe Ġexecuted ĊĠĠĠĠ Ġ* Ġ@ return Ġthe ĠObservable Ġof ĠRed uc ers Ġ( how Ġto Ġchange Ġthe Ġcurrent Ġstate ). ĠIf Ġit Ġreturns Ġnull , Ġor Ġan Ġempty Ġobservable , ĊĠĠĠĠ Ġ* Ġit Ġmeans Ġthe Ġgiven Ġaction Ġdoes Ġnot Ġchange Ġthe Ġstate . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġabstract Ġfun Ġcreate Reducer Observable ( action : ĠA ): ĠObservable < Reducer < S >> ? ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* Ġ@ param Ġaction ĊĠĠĠĠ Ġ* Ġ@ return Ġtrue Ġif Ġthe Ġapp Ġneeds Ġto Ġshow Ġthe Ġspinner Ġfor Ġthe Ġaction . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġopen Ġfun Ġshow Spinner ( action : ĠA ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDefault Ġscheduler Ġfor Ġobservable s ĊĠĠĠĠ Ġ* Ġ@ return ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġopen Ġfun Ġdefault Scheduler (): ĠScheduler Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠS chedulers . computation () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġif Ġthe Ġreturn Ġvalue Ġis Ġtrue , Ġit Ġmeans Ġthe Ġerror Ġhas Ġbeen Ġhandled , Ġand Ġit Ġwill Ġnot Ġbe Ġin Ġthe Ġerror Subject ĊĠĠĠ Ġopen Ġfun Ġhandle Error ( action : ĠA ?, Ġerror : ĠThrowable ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġyou Ġcan Ġput Ġa Ġbreakpoint Ġhere Ġso Ġyou Ġwill Ġget Ġnotified Ġwhenever Ġthere Ġis Ġan Ġerror . ĊĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T > Ġto LiveData ( observable : ĠObservable < T >): ĠRx LiveData < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġobservable Without Error Ġ= Ġobservable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Resume Next Ġ{ Ġerror : ĠThrowable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! handleError ( null , Ġerror )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Subject . onNext ( Action And Error ( null , Ġerror )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . empty < T >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠRx LiveData ( observable Without Error ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠExecute Ġan Ġaction . ĊĠĠĠĠ Ġ* Ġ@ param Ġaction ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinline Ġfun Ġexecute ( action : ĠA ) Ġ{ ĊĠĠĠĠĠĠĠ Ġexecute ( action , ĠParallel Default ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠExecute Ġan Ġaction Ġwith Ġa Ġspecific Ġexecution Mode . ĊĠĠĠĠ Ġ* Ġ@ param Ġaction ĊĠĠĠĠ Ġ* Ġ@ param Ġmode ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġexecute ( action : ĠA , Ġmode : ĠAction Execution Mode ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen ( mode ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠParallel Ġ-> Ġ Ġenqueue In Parallel ( action , Ġstate Map In Order Ġ= Ġfalse , Ġdeferred Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠParallel De fer Ġ-> Ġenqueue In Parallel ( action , Ġstate Map In Order Ġ= Ġtrue , Ġdeferred Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSequence Ġ-> Ġenqueue In Sequence ( action ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSwitch Map Ġ-> Ġenqueue With Switch Map ( action ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġenqueue In Parallel ( action , Ġstate Map In Order Ġ= Ġtrue , Ġdeferred Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠExecute Ġthe Ġactions Ġin Ġsequence . ĠThe ĠView Ġclasses Ġcreate ĠActions Ġand Ġcall Ġthis Ġmethod Ġto Ġrun Ġthem Ġin Ġsequence . ĊĠĠĠĠ Ġ* ĠIf Ġan Ġaction Ġfails , Ġthe Ġrest Ġof Ġthe Ġactions Ġwill Ġnot Ġbe Ġexecuted . ĊĠĠĠĠ Ġ* Ġ@ param Ġactions ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġexecute In Sequence ( var arg Ġactions : ĠA ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsequential Action Subject . onNext ( actions . asList ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T > ĠObservable < T >. as LiveData (): ĠRx LiveData < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . as LiveData ( this @ Rx ViewModel ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA > Ġaction On Next Observable ( clazz : ĠClass < T >): ĠObservable < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction On Next Observable Ġ{ Ġclazz . is Instance ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. cast ( clazz ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġaction On Next Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean ): ĠObservable < A > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . action On Next Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . action Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġpredicate ( it ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA , ĠR > Ġaction On Next Observable ( clazz : ĠClass < T >, Ġmapper : Ġ( S ) Ġ-> ĠR ?): ĠObservable < R > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction On Next Observable ({ Ġclazz . is Instance ( it ) Ġ} Ġ, Ġmapper ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< R > Ġaction On Next Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean , Ġmapper : Ġ( S ) Ġ-> ĠR ?): ĠObservable < R > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . action On Next Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġpredicate ( it . action ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. skip Null Ġ{ Ġmapper . invoke ( it . state ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA > Ġaction On Complete Observable ( clazz : ĠClass < T >): ĠObservable < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction On Complete Observable Ġ{ Ġclazz . is Instance ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. cast ( clazz ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġaction On Complete Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean ): ĠObservable < A > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . action On Complete Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . action Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġpredicate ( it ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA , ĠR > Ġaction On Complete Observable ( clazz : ĠClass < T >, Ġmapper : Ġ( S ) Ġ-> ĠR ?): ĠObservable < R > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction On Complete Observable ({ Ġclazz . is Instance ( it ) Ġ}, Ġmapper ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< R > Ġaction On Complete Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean , Ġmapper : Ġ( S ) Ġ-> ĠR ?): ĠObservable < R > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . action On Complete Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġpredicate ( it . action ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. skip Null Ġ{ Ġmapper . invoke ( it . state ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA > Ġaction Error Observable ( clazz : ĠClass < T >): ĠObservable < Throwable > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġaction Error Observable Ġ{ Ġclazz . is Instance ( it ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġaction Error Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean ): ĠObservable < Throwable > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . action Error Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . action ?. let Ġ{ Ġa Ġ-> Ġpredicate ( a ) Ġ} Ġ?: Ġfalse Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . error Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ< T : ĠA > Ġloading Observable ( clazz : ĠClass < T >): ĠObservable < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġloading Observable Ġ{ Ġclazz . is Instance ( it ) Ġ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġloading Observable ( predicate : Ġ( A ) Ġ-> ĠBoolean ): ĠObservable < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġloading Count ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġpredicate ( it . first ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . second Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Boolean Observable () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠObservable < Int >. to Boolean Observable () Ġ: ĠObservable < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . scan Ġ{ Ġx , Ġy Ġ-> Ġx Ġ+ Ġy Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġcount Ġ-> Ġcount Ġ> Ġ 0 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. start With Item ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. distinct Until Changed () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Latest From ( state Observable , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBi Function < Boolean , ĠS , ĠBoolean > Ġ{ Ġb , Ġ_ Ġ-> Ġb Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. replay ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. auto Connect () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġexecute And Combine ( action : ĠA , Ġthrow Error : ĠBoolean Ġ= Ġfalse , Ġdeferred Action : ĠBoolean Ġ= Ġfalse ): ĠObservable < An s Wrapper > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠObservable . combine Latest ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . just ( action ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( deferred Action ) ĠObservable . empty () Ġelse Ġto Reducer Observable ( action ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBi Function < A , ĠRed ucer < S >, ĠPair < A , ĠRed ucer < S >>> Ġ{ Ġa , Ġm Ġ-> ĠPair ( a , Ġm ) Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . single ()) ĠĠĠĠ Ġ// Ġuse ĠS chedulers . single () Ġto Ġavoid Ġrace Ġcondition ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠAction And State ( it . first , Ġit . second . invoke ( currentState )) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġstate Behavior Subject . onNext ( it . state ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( default Scheduler ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠAn s Wrapper ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. material ize () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġif Ġ( deferred Action Ġ&& Ġit . is On Complete ) Ġdeferred Action Subject . onNext ( action ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ! deferred Action Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . error ?. let Ġ{ throwable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! handleError ( action , Ġthrowable )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Subject . onNext ( Action And Error ( action , Ġthrowable )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . is On Complete ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . just ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠNotification . create On Next ( An s Wrapper ( Action And State ( action , Ġcurrent State ), Ġtrue )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠNotification . create On Complete < An s Wrapper >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . just ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. de material ize Ġ{ Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Resume Next Ġ{ Ġthrowable : ĠThrowable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( throw Error ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġthrowable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . empty < An s Wrapper >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! deferred Action Ġ&& Ġshow Spinner ( action )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġloading Count . onNext ( Pair ( action , Ġ 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do Finally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! deferred Action Ġ&& Ġshow Spinner ( action )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġloading Count . onNext ( Pair ( action , Ġ- 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe On ( default Scheduler ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġenqueue In Parallel ( action : ĠA , Ġstate Map In Order : ĠBoolean , Ġdeferred : ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( state Map In Order ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconcat E ager Action Subject . onNext ( Def err able Action ( action , Ġdeferred )) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconcurrent Action Subject . onNext ( action ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġenqueue In Sequence ( action : ĠA ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsequential Action Subject . onNext ( listOf ( action )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġenqueue With Switch Map ( action : ĠA ) Ġ{ ĊĠĠĠĠĠĠĠ Ġswitch Map Latest Action Subject . onNext ( action ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġto Reducer Observable ( action : ĠA ): ĠObservable < Reducer < S >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠObservable . just ( action ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġa Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstate Mapper Observable : ĠObservable < Reducer < S >> Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Reducer Observable ( a ) Ġ?: ĠObservable . just Ġ{ Ġs Ġ-> Ġs Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate Mapper Observable ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. default If Empty Ġ{ Ġs Ġ-> Ġs Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinner Ġclass ĠAn s Wrapper ( val Ġaction And State : ĠAction And State < A , ĠS >, Ġval Ġis Complete : ĠBoolean Ġ= Ġfalse ) ĊĠĠĠ Ġprivate Ġinner Ġclass ĠDef err able Action ( val Ġaction : ĠA , Ġval Ġis Deferred : ĠBoolean Ġ= Ġfalse ) Ċ }