package Ġcom . n uts hel lf cm . rx work framework . works Ċ Ċ import Ġcom . n uts hel lf cm . rx work framework . Cancel able Ċ import Ġcom . n uts hel lf cm . rx work framework . Observer Ċ import Ġcom . n uts hel lf cm . rx work framework . sche ud lers . Scheduler Ċ import Ġcom . n uts hel lf cm . rx work framework . sche ud lers . Schedulers Ċ import Ġcom . n uts hel lf cm . rx work framework . Cancelled Observer Ċ import Ġcom . n uts hel lf cm . rx work framework . Subscriber Ċ import Ġcom . n uts hel lf cm . rx work framework . works . run n ables .* Ċ import Ġjava . util . concurrent . atomic . Atomic Reference ĊĊ Ċ internal Ġopen Ġclass ĠWork < T >( ĊĠĠĠ Ġfinal Ġoverride Ġval Ġemitted Runnable : ĠE mitted Runnable < T > Ċ ) Ġ: ĠScheduled Work < T >, ĠParent Work < T >, ĠWork Emitter < T > Ġ{ ĊĊĊĠĠĠ Ġprotected Ġval Ġob servers : ĠMutable Set < Observer < T >> Ġ= Ġmutable Set Of () ĊĊĠĠĠ Ġprotected Ġval Ġsubscribers : ĠMutable Set < Subscriber > Ġ= Ġmutable Set Of () ĊĊĠĠĠ Ġprotected Ġval Ġcancelled Observer s : ĠMutable Set < Cancelled Observer > Ġ= Ġmutable Set Of () ĊĊĠĠĠ Ġfinal Ġoverride Ġvar Ġsubscribe On Scheduler : ĠScheduler Ġ= ĠS chedulers . default Scheduler ĊĊĠĠĠ Ġfinal Ġoverride Ġvar Ġobserve On Scheduler : ĠScheduler Ġ= ĠS chedulers . default Scheduler ĊĊĠĠĠ Ġprotected Ġval Ġis Scheduled Ġ= ĠAtomic Reference < Cancel able >() ĊĊĠĠĠ Ġoverride Ġval Ġcomposite C ancellable : ĠComposite C ancellable Ġ= ĠComposite C ancellable ( this ) ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġemitted Runnable . emitter Ġ= Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Result ( result : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnotify Observer s Runnable Ġ= ĠNotify Observer s Runnable ( ob servers ) ĊĠĠĠĠĠĠĠ Ġnotify Observer s Runnable . result Ġ= Ġresult ĊĠĠĠĠĠĠĠ Ġobserve On Scheduler . schedule ( notify Observer s Runnable ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( error : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnotify Observer s Runnable Ġ= ĠNotify Observer s Runnable ( ob servers ) ĊĠĠĠĠĠĠĠ Ġnotify Observer s Runnable . throwable Ġ= Ġerror ĊĠĠĠĠĠĠĠ Ġobserve On Scheduler . schedule ( notify Observer s Runnable ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġoverride Ġfun Ġon Work Cancelled () Ġ{ ĊĠĠĠĠĠĠĠ Ġobserve On Scheduler . schedule ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠNotify On Cancelled Runnable ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcancelled Observer s , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġob servers Ġas ĠMutable Set < Observer <* >>, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscribers ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsubscribe ( on Result : Ġ( T ) Ġ-> ĠUnit , ĠonError : Ġ( Throwable ) Ġ-> ĠUnit ): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġob servers . add ( Observer ( on Result , ĠonError )) ĊĠĠĠĠĠĠĠ Ġreturn Ġsubscribe Actual ( subscribe On Scheduler ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsubscribe On Observer s ( ob servers : ĠSet < Observer < T >>, Ġscheduler : ĠScheduler Ġ= Ġobserve On Scheduler ): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . ob servers . addAll ( ob servers ) ĊĠĠĠĠĠĠĠ Ġreturn Ġsubscribe Actual ( scheduler ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsubscribe Actual ( scheduler : ĠScheduler ): ĠComposite C ancellable Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscribers . forEach Ġ{ Ġit . on Subscribe () Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( is Scheduled . get () Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcancell able Ġ= Ġscheduler . schedule ( emit ted Runnable ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomposite C ancellable . append ( c ancellable ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis Scheduled . get AndSet ( c ancellable ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġcomposite C ancellable ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġblocking Get (): ĠT ? Ġ= ĠBlock ing Work ( Blocking Runnable ( emit ted Runnable ), Ġthis ). blocking Get () ĊĊĠĠĠ Ġoverride Ġfun Ġsubscribe On ( scheduler : ĠScheduler ): ĠScheduled Work < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . subscribe On Scheduler Ġ= Ġscheduler ĊĠĠĠĠĠĠĠ Ġif Ġ( observe On Scheduler Ġ== ĠS chedulers . default Scheduler ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġobserve On Scheduler Ġ= Ġscheduler ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġobserve On ( scheduler : ĠScheduler ): ĠScheduled Work < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . observe On Scheduler Ġ= Ġscheduler ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġ< S > Ġmap ( transition : Ġ( T ) Ġ-> ĠS ): ĠScheduled Work < S > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMap Work ( Work Map Runnable ( transition ), Ġthis ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġoverride Ġfun Ġ< S > Ġflat Map ( transition : Ġ( T ) Ġ-> ĠScheduled Work < S >): ĠScheduled Work < S > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFlat Map Work ( Work Map Runnable ( transition ), Ġthis ) Ġas ĠScheduled Work < S > ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġfilter ( filter : Ġ( T ) Ġ-> ĠBoolean ): ĠScheduled Work < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFilter Work ( Work Filter Runnable ( filter ), Ġthis ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġdo On Subscribe ( on Subscribe : Ġ() Ġ-> ĠUnit ): ĠScheduled Work < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscribers . add ( Subscriber ( on Subscribe )) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġdo On Cancelled ( on Cancelled : Ġ() Ġ-> ĠUnit ): ĠScheduled Work < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġcancelled Observer s . add ( Cancelled Observer ( on Cancelled )) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} Ċ } Ċ