From 1b01931f25fce016e5aff3ce4bd0c41528fa6ec5 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Mon, 22 Jun 2026 10:36:34 +0200 Subject: [PATCH] 4.x: Remove Scheduler.when due to maintenance burden --- .../io/reactivex/rxjava4/core/Scheduler.java | 85 ---- .../internal/schedulers/SchedulerWhen.java | 331 --------------- .../schedulers/SchedulerWhenTest.java | 401 ------------------ 3 files changed, 817 deletions(-) delete mode 100644 src/main/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhen.java delete mode 100644 src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhenTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Scheduler.java b/src/main/java/io/reactivex/rxjava4/core/Scheduler.java index baeecdaad5b..7a69bcacbd8 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Scheduler.java +++ b/src/main/java/io/reactivex/rxjava4/core/Scheduler.java @@ -13,13 +13,11 @@ package io.reactivex.rxjava4.core; -import java.util.Objects; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.disposables.*; import io.reactivex.rxjava4.internal.schedulers.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; @@ -299,89 +297,6 @@ public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initial return periodicTask; } - /** - * Allows the use of operators for controlling the timing around when - * actions scheduled on workers are actually done. This makes it possible to - * layer additional behavior on this {@link Scheduler}. The only parameter - * is a function that flattens an {@link Flowable} of {@link Flowable} - * of {@link Completable}s into just one {@link Completable}. There must be - * a chain of operators connecting the returned value to the source - * {@link Flowable} otherwise any work scheduled on the returned - * {@link Scheduler} will not be executed. - *

- * When {@link Scheduler#createWorker()} is invoked a {@link Flowable} of - * {@link Completable}s is onNext'd to the combinator to be flattened. If - * the inner {@link Flowable} is not immediately subscribed to an calls to - * {@link Worker#schedule} are buffered. Once the {@link Flowable} is - * subscribed to actions are then onNext'd as {@link Completable}s. - *

- * Finally the actions scheduled on the parent {@link Scheduler} when the - * inner most {@link Completable}s are subscribed to. - *

- * When the {@link Worker} is unsubscribed the {@link Completable} emits an - * onComplete and triggers any behavior in the flattening operator. The - * {@link Flowable} and all {@link 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(Flowable.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 {@link Worker}s rather than individual - * actions. Generally each {@link Flowable} uses its own {@link Worker}. - * This means that this will essentially limit the number of concurrent - * subscribes. The danger comes from using operators like - * {@link Flowable#zip(java.util.concurrent.Flow.Publisher, java.util.concurrent.Flow.Publisher, io.reactivex.rxjava4.functions.BiFunction)} where - * subscribing to the first {@link Flowable} could deadlock the - * subscription to the second. - * - * 

-     * Scheduler limitScheduler = Schedulers.computation().when(workers -> {
-     *  // use merge max concurrent to limit the number of concurrent
-     *  // Flowables two at a time
-     *  return Completable.merge(Flowable.merge(workers, 2));
-     * });
-     *
- * - * Slowing down the rate to no more than 1 a second. This suffers from - * the same problem as the one above I could find an {@link Flowable} - * 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 - * @param a Scheduler and a Subscription - * @param combine the function that takes a two-level nested Flowable sequence of a Completable and returns - * the Completable that will be subscribed to and should trigger the execution of the scheduled Actions. - * @return the Scheduler with the customized execution behavior - * @throws NullPointerException if {@code combine} is {@code null} - * @since 2.1 - */ - @SuppressWarnings("unchecked") - @NonNull - public S when(@NonNull Function>, Completable> combine) { - Objects.requireNonNull(combine, "combine is null"); - return (S) new SchedulerWhen(combine, this); - } - /** * Turn this {@code Scheduler} into an {@link ExecutorService} implementation * using its various *Direct() methods instead of workers. diff --git a/src/main/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhen.java b/src/main/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhen.java deleted file mode 100644 index c7d03b05f8f..00000000000 --- a/src/main/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhen.java +++ /dev/null @@ -1,331 +0,0 @@ -/* - * Copyright (c) 2016-present, RxJava Contributors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is - * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See - * the License for the specific language governing permissions and limitations under the License. - */ - -package io.reactivex.rxjava4.internal.schedulers; - -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.*; - -import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.functions.Function; -import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.processors.*; - -/** - * Allows the use of operators for controlling the timing around when actions - * scheduled on workers are actually done. This makes it possible to layer - * additional behavior on this {@link Scheduler}. The only parameter is a - * function that flattens an {@link Observable} of {@link Observable} of - * {@link Completable}s into just one {@link Completable}. There must be a chain - * of operators connecting the returned value to the source {@link Observable} - * otherwise any work scheduled on the returned {@link Scheduler} will not be - * executed. - *

- * When {@link Scheduler#createWorker()} is invoked a {@link Observable} of - * {@link Completable}s is onNext'd to the combinator to be flattened. If the - * inner {@link Observable} is not immediately subscribed to an calls to - * {@link Worker#schedule} are buffered. Once the {@link Observable} is - * subscribed to actions are then onNext'd as {@link Completable}s. - *

- * Finally the actions scheduled on the parent {@link Scheduler} when the inner - * most {@link Completable}s are subscribed to. - *

- * When the {@link io.reactivex.rxjava4.core.Scheduler.Worker Worker} is unsubscribed the {@link Completable} emits an - * onComplete and triggers any behavior in the flattening operator. The - * {@link Observable} and all {@link 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: - * - * 

- * {@code
- * 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 {@link io.reactivex.rxjava4.core.Scheduler.Worker Worker}s rather than individual actions. - * Generally each {@link Observable} uses its own {@link io.reactivex.rxjava4.core.Scheduler.Worker Worker}. This means - * that this will essentially limit the number of concurrent subscribes. The - * danger comes from using operators like - * {@link Flowable#zip(java.util.concurrent.Flow.Publisher, java.util.concurrent.Flow.Publisher, io.reactivex.rxjava4.functions.BiFunction)} where - * subscribing to the first {@link Observable} could deadlock the subscription - * to the second. - * - * 

- * {@code
- * 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 1 a second. This suffers from the - * same problem as the one above I could find an {@link Observable} operator - * that limits the rate without dropping the values (aka leaky bucket - * algorithm). - * - * 
- * {@code
- * 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 - * @since 2.1 - */ -public class SchedulerWhen extends Scheduler implements Disposable { - private final Scheduler actualScheduler; - private final FlowableProcessor> workerProcessor; - private Disposable disposable; - - public SchedulerWhen(Function>, Completable> combine, Scheduler actualScheduler) { - this.actualScheduler = actualScheduler; - // workers are converted into completables and put in this queue. - this.workerProcessor = UnicastProcessor.>create().toSerialized(); - // send it to a custom combinator to pick the order and rate at which - // workers are processed. - try { - disposable = combine.apply(workerProcessor).subscribe(); - } catch (Throwable e) { - throw ExceptionHelper.wrapOrThrow(e); - } - } - - @Override - public void dispose() { - disposable.dispose(); - } - - @Override - public boolean isDisposed() { - return disposable.isDisposed(); - } - - @NonNull - @Override - public Worker createWorker() { - final Worker actualWorker = actualScheduler.createWorker(); - // a queue for the actions submitted while worker is waiting to get to - // the subscribe to off the workerQueue. - final FlowableProcessor actionProcessor = UnicastProcessor.create().toSerialized(); - // convert the work of scheduling all the actions into a completable - Flowable actions = actionProcessor.map(new CreateWorkerFunction(actualWorker)); - - // a worker that queues the action to the actionQueue subject. - Worker worker = new QueueWorker(actionProcessor, actualWorker); - - // enqueue the completable that process actions put in reply subject - workerProcessor.onNext(actions); - - // return the worker that adds actions to the reply subject - return worker; - } - - static final Disposable SUBSCRIBED = new SubscribedDisposable(); - - static final Disposable DISPOSED = Disposable.disposed(); - - @SuppressWarnings("serial") - abstract static class ScheduledAction extends AtomicReference implements Disposable { - ScheduledAction() { - super(SUBSCRIBED); - } - - void call(Worker actualWorker, CompletableObserver actionCompletable) { - Disposable oldState = get(); - // either SUBSCRIBED or UNSUBSCRIBED - if (oldState == DISPOSED) { - // no need to schedule return - return; - } - if (oldState != SUBSCRIBED) { - // has already been scheduled return - // should not be able to get here but handle it anyway by not - // rescheduling. - return; - } - - Disposable newState = callActual(actualWorker, actionCompletable); - - if (!compareAndSet(SUBSCRIBED, newState)) { - // set would only fail if the new current state is some other - // subscription from a concurrent call to this method. - // Unsubscribe from the action just scheduled because it lost - // the race. - newState.dispose(); - } - } - - protected abstract Disposable callActual(Worker actualWorker, CompletableObserver actionCompletable); - - @Override - public boolean isDisposed() { - return get().isDisposed(); - } - - @Override - public void dispose() { - getAndSet(DISPOSED).dispose(); - } - } - - @SuppressWarnings("serial") - static class ImmediateAction extends ScheduledAction { - private final Runnable action; - - ImmediateAction(Runnable action) { - this.action = action; - } - - @Override - protected Disposable callActual(Worker actualWorker, CompletableObserver actionCompletable) { - return actualWorker.schedule(new OnCompletedAction(action, actionCompletable)); - } - } - - @SuppressWarnings("serial") - static class DelayedAction extends ScheduledAction { - private final Runnable action; - private final long delayTime; - private final TimeUnit unit; - - DelayedAction(Runnable action, long delayTime, TimeUnit unit) { - this.action = action; - this.delayTime = delayTime; - this.unit = unit; - } - - @Override - protected Disposable callActual(Worker actualWorker, CompletableObserver actionCompletable) { - return actualWorker.schedule(new OnCompletedAction(action, actionCompletable), delayTime, unit); - } - } - - static class OnCompletedAction implements Runnable { - final CompletableObserver actionCompletable; - final Runnable action; - - OnCompletedAction(Runnable action, CompletableObserver actionCompletable) { - this.action = action; - this.actionCompletable = actionCompletable; - } - - @Override - public void run() { - try { - action.run(); - } finally { - actionCompletable.onComplete(); - } - } - } - - static final class CreateWorkerFunction implements Function { - final Worker actualWorker; - - CreateWorkerFunction(Worker actualWorker) { - this.actualWorker = actualWorker; - } - - @Override - public Completable apply(final ScheduledAction action) { - return new WorkerCompletable(action); - } - - final class WorkerCompletable extends Completable { - final ScheduledAction action; - - WorkerCompletable(ScheduledAction action) { - this.action = action; - } - - @Override - protected void subscribeActual(CompletableObserver actionCompletable) { - actionCompletable.onSubscribe(action); - action.call(actualWorker, actionCompletable); - } - } - } - - static final class QueueWorker extends Worker { - private final AtomicBoolean unsubscribed; - private final FlowableProcessor actionProcessor; - private final Worker actualWorker; - - QueueWorker(FlowableProcessor actionProcessor, Worker actualWorker) { - this.actionProcessor = actionProcessor; - this.actualWorker = actualWorker; - unsubscribed = new AtomicBoolean(); - } - - @Override - public void dispose() { - // complete the actionQueue when worker is unsubscribed to make - // room for the next worker in the workerQueue. - if (unsubscribed.compareAndSet(false, true)) { - actionProcessor.onComplete(); - actualWorker.dispose(); - } - } - - @Override - public boolean isDisposed() { - return unsubscribed.get(); - } - - @NonNull - @Override - public Disposable schedule(@NonNull final Runnable action, final long delayTime, @NonNull final TimeUnit unit) { - // send a scheduled action to the actionQueue - DelayedAction delayedAction = new DelayedAction(action, delayTime, unit); - actionProcessor.onNext(delayedAction); - return delayedAction; - } - - @NonNull - @Override - public Disposable schedule(@NonNull final Runnable action) { - // send a scheduled action to the actionQueue - ImmediateAction immediateAction = new ImmediateAction(action); - actionProcessor.onNext(immediateAction); - return immediateAction; - } - } - - static final class SubscribedDisposable implements Disposable { - @Override - public void dispose() { - } - - @Override - public boolean isDisposed() { - return false; - } - } -} diff --git a/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhenTest.java b/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhenTest.java deleted file mode 100644 index 5dee0d502c7..00000000000 --- a/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerWhenTest.java +++ /dev/null @@ -1,401 +0,0 @@ -/* - * Copyright (c) 2016-present, RxJava Contributors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is - * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See - * the License for the specific language governing permissions and limitations under the License. - */ - -package io.reactivex.rxjava4.internal.schedulers; - -import static io.reactivex.rxjava4.core.Flowable.*; -import static java.util.concurrent.TimeUnit.SECONDS; -import static org.junit.Assert.*; - -import java.util.concurrent.atomic.AtomicInteger; - -import org.junit.Test; - -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.core.Scheduler.Worker; -import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.*; -import io.reactivex.rxjava4.internal.schedulers.SchedulerWhen.*; -import io.reactivex.rxjava4.observers.DisposableCompletableObserver; -import io.reactivex.rxjava4.processors.PublishProcessor; -import io.reactivex.rxjava4.schedulers.*; -import io.reactivex.rxjava4.subscribers.TestSubscriber; -import io.reactivex.rxjava4.testsupport.TestHelper; - -public class SchedulerWhenTest extends RxJavaTest { - @Test - public void asyncMaxConcurrent() { - TestScheduler tSched = new TestScheduler(); - SchedulerWhen sched = maxConcurrentScheduler(tSched); - TestSubscriber tSub = TestSubscriber.create(); - - asyncWork(sched).subscribe(tSub); - - tSub.assertValueCount(0); - - tSched.advanceTimeBy(0, SECONDS); - tSub.assertValueCount(0); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(2); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(4); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(5); - tSub.assertComplete(); - - sched.dispose(); - } - - @Test - public void asyncDelaySubscription() { - final TestScheduler tSched = new TestScheduler(); - SchedulerWhen sched = throttleScheduler(tSched); - TestSubscriber tSub = TestSubscriber.create(); - - asyncWork(sched).subscribe(tSub); - - tSub.assertValueCount(0); - - tSched.advanceTimeBy(0, SECONDS); - tSub.assertValueCount(0); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(1); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(1); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(2); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(2); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(3); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(3); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(4); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(4); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(5); - tSub.assertComplete(); - - sched.dispose(); - } - - @Test - public void syncMaxConcurrent() { - TestScheduler tSched = new TestScheduler(); - SchedulerWhen sched = maxConcurrentScheduler(tSched); - TestSubscriber tSub = TestSubscriber.create(); - - syncWork(sched).subscribe(tSub); - - tSub.assertValueCount(0); - tSched.advanceTimeBy(0, SECONDS); - - // since all the work is synchronous nothing is blocked and its all done - tSub.assertValueCount(5); - tSub.assertComplete(); - - sched.dispose(); - } - - @Test - public void syncDelaySubscription() { - final TestScheduler tSched = new TestScheduler(); - SchedulerWhen sched = throttleScheduler(tSched); - TestSubscriber tSub = TestSubscriber.create(); - - syncWork(sched).subscribe(tSub); - - tSub.assertValueCount(0); - - tSched.advanceTimeBy(0, SECONDS); - tSub.assertValueCount(1); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(2); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(3); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(4); - - tSched.advanceTimeBy(1, SECONDS); - tSub.assertValueCount(5); - tSub.assertComplete(); - - sched.dispose(); - } - - private Flowable asyncWork(final Scheduler sched) { - return Flowable.range(1, 5).flatMap(new Function>() { - @Override - public Flowable apply(Integer t) { - return Flowable.timer(1, SECONDS, sched); - } - }); - } - - private Flowable syncWork(final Scheduler sched) { - return Flowable.range(1, 5).flatMap(new Function>() { - @Override - public Flowable apply(Integer t) { - return Flowable.defer(new Supplier>() { - @Override - public Flowable get() { - return Flowable.just(0l); - } - }).subscribeOn(sched); - } - }); - } - - private SchedulerWhen maxConcurrentScheduler(TestScheduler tSched) { - SchedulerWhen sched = new SchedulerWhen(new Function>, Completable>() { - @Override - public Completable apply(Flowable> workerActions) { - Flowable workers = workerActions.map(new Function, Completable>() { - @Override - public Completable apply(Flowable actions) { - return Completable.concat(actions); - } - }); - return Completable.merge(workers, 2); - } - }, tSched); - return sched; - } - - private SchedulerWhen throttleScheduler(final TestScheduler tSched) { - SchedulerWhen sched = new SchedulerWhen(new Function>, Completable>() { - @Override - public Completable apply(Flowable> workerActions) { - Flowable workers = workerActions.map(new Function, Completable>() { - @Override - public Completable apply(Flowable actions) { - return Completable.concat(actions); - } - }); - return Completable.concat(workers.map(new Function() { - @Override - public Completable apply(Completable worker) { - return worker.delay(1, SECONDS, tSched); - } - })); - } - }, tSched); - return sched; - } - - @Test - public void raceConditions() { - Scheduler comp = Schedulers.computation(); - Scheduler limited = comp.when(new Function>, Completable>() { - @Override - public Completable apply(Flowable> t) { - return Completable.merge(Flowable.merge(t, 10)); - } - }); - - merge(just(just(1).subscribeOn(limited).observeOn(comp)).repeat(1000)).blockingSubscribe(); - } - - @Test - public void subscribedDisposable() { - SchedulerWhen.SUBSCRIBED.dispose(); - assertFalse(SchedulerWhen.SUBSCRIBED.isDisposed()); - } - - @SuppressWarnings("resource") - @Test(expected = TestException.class) - public void combineCrashInConstructor() { - new SchedulerWhen(new Function>, Completable>() { - @Override - public Completable apply(Flowable> v) - throws Exception { - throw new TestException(); - } - }, Schedulers.single()); - } - - @SuppressWarnings("resource") - @Test - public void disposed() { - SchedulerWhen sw = new SchedulerWhen(new Function>, Completable>() { - @Override - public Completable apply(Flowable> v) - throws Exception { - return Completable.never(); - } - }, Schedulers.single()); - - assertFalse(sw.isDisposed()); - - sw.dispose(); - - assertTrue(sw.isDisposed()); - } - - @Test - public void scheduledActiondisposedSetRace() { - for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { - try (final var sa = new ScheduledAction() { - - private static final long serialVersionUID = -672980251643733156L; - - @Override - protected Disposable callActual(Worker actualWorker, - CompletableObserver actionCompletable) { - return Disposable.empty(); - } - - }) { - - assertFalse(sa.isDisposed()); - - Runnable r1 = new Runnable() { - @Override - public void run() { - sa.dispose(); - } - }; - - TestHelper.race(r1, r1); - - assertTrue(sa.isDisposed()); - } - } - } - - @SuppressWarnings("resource") - @Test - public void scheduledActionStates() { - final AtomicInteger count = new AtomicInteger(); - ScheduledAction sa = new ScheduledAction() { - - private static final long serialVersionUID = -672980251643733156L; - - @Override - protected Disposable callActual(Worker actualWorker, - CompletableObserver actionCompletable) { - count.incrementAndGet(); - return Disposable.empty(); - } - - }; - - assertFalse(sa.isDisposed()); - - sa.dispose(); - - assertTrue(sa.isDisposed()); - - sa.dispose(); - - assertTrue(sa.isDisposed()); - - // should not run when disposed - sa.call(Schedulers.single().createWorker(), null); - - assertEquals(0, count.get()); - - // should not run when already scheduled - sa.set(Disposable.empty()); - - sa.call(Schedulers.single().createWorker(), null); - - assertEquals(0, count.get()); - - // disposed while in call - sa = new ScheduledAction() { - - private static final long serialVersionUID = -672980251643733156L; - - @Override - protected Disposable callActual(Worker actualWorker, - CompletableObserver actionCompletable) { - count.incrementAndGet(); - dispose(); - return Disposable.empty(); - } - - }; - - sa.call(Schedulers.single().createWorker(), null); - - assertEquals(1, count.get()); - } - - @Test - public void onCompleteActionRunCrash() { - final AtomicInteger count = new AtomicInteger(); - - OnCompletedAction a = new OnCompletedAction(new Runnable() { - @Override - public void run() { - throw new TestException(); - } - }, new DisposableCompletableObserver() { - - @Override - public void onComplete() { - count.incrementAndGet(); - } - - @Override - public void onError(Throwable e) { - count.decrementAndGet(); - e.printStackTrace(); - } - }); - - try { - a.run(); - fail("Should have thrown"); - } catch (TestException expected) { - - } - - assertEquals(1, count.get()); - } - - @Test - public void queueWorkerDispose() { - @SuppressWarnings("resource") - QueueWorker qw = new QueueWorker(PublishProcessor.create(), Schedulers.single().createWorker()); - - assertFalse(qw.isDisposed()); - - qw.dispose(); - - assertTrue(qw.isDisposed()); - - qw.dispose(); - - assertTrue(qw.isDisposed()); - } -}