public class SchedulerWhen extends Scheduler implements Disposable
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
Worker#schedule
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 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 limitScheduler = 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 Worker
s rather than individual actions.
Generally each Observable
uses its own Worker
. This means
that this will essentially limit the number of concurrent subscribes. The
danger comes from using operators like
Flowable.zip(org.reactivestreams.Publisher, org.reactivestreams.Publisher, io.reactivex.functions.BiFunction)
where
subscribing to the first Observable
could deadlock the subscription
to the second.
Scheduler limitScheduler = 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 slowScheduler = 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)); })); });
History 2.0.1 - experimental
Scheduler.Worker
Constructor and Description |
---|
SchedulerWhen(Function<Flowable<Flowable<Completable>>,Completable> combine,
Scheduler actualScheduler) |
Modifier and Type | Method and Description |
---|---|
Scheduler.Worker |
createWorker()
Retrieves or creates a new
Scheduler.Worker that represents sequential execution of actions. |
void |
dispose()
Dispose the resource, the operation should be idempotent.
|
boolean |
isDisposed()
Returns true if this resource has been disposed.
|
clockDriftTolerance, now, scheduleDirect, scheduleDirect, schedulePeriodicallyDirect, shutdown, start, when
public SchedulerWhen(Function<Flowable<Flowable<Completable>>,Completable> combine, Scheduler actualScheduler)
public void dispose()
Disposable
dispose
in interface Disposable
public boolean isDisposed()
Disposable
isDisposed
in interface Disposable
@NonNull public Scheduler.Worker createWorker()
Scheduler
Scheduler.Worker
that represents sequential execution of actions.
When work is completed, the Worker
instance should be released
by calling Disposable.dispose()
to avoid potential resource leaks in the
underlying task-execution scheme.
Work on a Scheduler.Worker
is guaranteed to be sequential and non-overlapping.
createWorker
in class Scheduler
Copyright © 2019. All rights reserved.