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());
- }
-}