Class SchedulerWhen
- Record Components:
combine
-
- All Implemented Interfaces:
Subscription
Scheduler
. The only parameter is a
function that flattens an Observable
of Observable
of
Completable
s into just one Completable
. There must be a chain
of operators connecting the returned value to the source Observable
otherwise any work scheduled on the returned Scheduler
will not be
executed.
When Scheduler.createWorker()
is invoked a Observable
of
Completable
s is onNext'd to the combinator to be flattened. If the
inner Observable
is not immediately subscribed to an calls to
Scheduler.Worker.schedule(rx.functions.Action0)
are buffered. Once the Observable
is
subscribed to actions are then onNext'd as Completable
s.
Finally the actions scheduled on the parent Scheduler
when the inner
most Completable
s are subscribed to.
When the Scheduler.Worker
is unsubscribed the Completable
emits an
onComplete and triggers any behavior in the flattening operator. The
Observable
and all Completable
s give to the flattening
function never onError.
Limit the amount concurrency two at a time without creating a new fix size thread pool:
Scheduler limitSched = Schedulers.computation().when(workers -> { // use merge max concurrent to limit the number of concurrent // callbacks two at a time return Completable.merge(Observable.merge(workers), 2); });
This is a slightly different way to limit the concurrency but it has some
interesting benefits and drawbacks to the method above. It works by limited
the number of concurrent Scheduler.Worker
s rather than individual actions.
Generally each Observable
uses its own Scheduler.Worker
. This means
that this will essentially limit the number of concurrent subscribes. The
danger comes from using operators like
Observable.zip(Observable, Observable, rx.functions.Func2)
where
subscribing to the first Observable
could deadlock the subscription
to the second.
Scheduler limitSched = Schedulers.computation().when(workers -> { // use merge max concurrent to limit the number of concurrent // Observables two at a time return Completable.merge(Observable.merge(workers, 2)); });Slowing down the rate to no more than than 1 a second. This suffers from the same problem as the one above I could find an
Observable
operator
that limits the rate without dropping the values (aka leaky bucket
algorithm).
Scheduler slowSched = Schedulers.computation().when(workers -> { // use concatenate to make each worker happen one at a time. return Completable.concat(workers.map(actions -> { // delay the starting of the next worker by 1 second. return Completable.merge(actions.delaySubscription(1, TimeUnit.SECONDS)); })); });
-
Nested Class Summary
Nested ClassesModifier and TypeClassDescriptionprivate static class
private static class
private static class
Nested classes/interfaces inherited from class rx.Scheduler
Scheduler.Worker
-
Field Summary
FieldsModifier and TypeFieldDescriptionprivate final Scheduler
private static final Subscription
private final Subscription
private static final Subscription
private final Observer
<Observable<Completable>> -
Constructor Summary
ConstructorsConstructorDescriptionSchedulerWhen
(Func1<Observable<Observable<Completable>>, Completable> combine, Scheduler actualScheduler) -
Method Summary
Modifier and TypeMethodDescriptionRetrieves or creates a newScheduler.Worker
that represents serial execution of actions.boolean
Indicates whether thisSubscription
is currently unsubscribed.void
Stops the receipt of notifications on theSubscriber
that was registered when this Subscription was received.
-
Field Details
-
actualScheduler
-
workerObserver
-
subscription
-
SUBSCRIBED
-
UNSUBSCRIBED
-
-
Constructor Details
-
SchedulerWhen
public SchedulerWhen(Func1<Observable<Observable<Completable>>, Completable> combine, Scheduler actualScheduler)
-
-
Method Details
-
unsubscribe
public void unsubscribe()Description copied from interface:Subscription
Stops the receipt of notifications on theSubscriber
that was registered when this Subscription was received.This allows unregistering an
Subscriber
before it has finished receiving all events (i.e. before onCompleted is called).- Specified by:
unsubscribe
in interfaceSubscription
-
isUnsubscribed
public boolean isUnsubscribed()Description copied from interface:Subscription
Indicates whether thisSubscription
is currently unsubscribed.- Specified by:
isUnsubscribed
in interfaceSubscription
- Returns:
true
if thisSubscription
is currently unsubscribed,false
otherwise
-
createWorker
Description copied from class:Scheduler
Retrieves or creates a newScheduler.Worker
that represents serial execution of actions.When work is completed it should be unsubscribed using
Subscription.unsubscribe()
.Work on a
Scheduler.Worker
is guaranteed to be sequential.- Specified by:
createWorker
in classScheduler
- Returns:
- a Worker representing a serial queue of actions to be executed
-