diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java index 4996e16789..202d4a7d5a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java @@ -96,7 +96,7 @@ void interrupedSleep(InterruptedException ex) { @Override public @NonNull CompletionStage finish() { waiter.complete(false); - lazySet(DisposableHelper.DISPOSED); + DisposableHelper.dispose(this); return FINISHED; } diff --git a/src/main/java/io/reactivex/rxjava4/schedulers/TestScheduler.java b/src/main/java/io/reactivex/rxjava4/schedulers/TestScheduler.java index ac817c29aa..959debd434 100644 --- a/src/main/java/io/reactivex/rxjava4/schedulers/TestScheduler.java +++ b/src/main/java/io/reactivex/rxjava4/schedulers/TestScheduler.java @@ -95,24 +95,28 @@ public TestScheduler(long delayTime, TimeUnit unit, boolean useOnScheduleHook) { } /** - * @param count for differentiating tasks at same time + * Represents an unit of work scheduled for a time and an unique total-order index. + * @param worker the parent {@link TestWorker} instance + * @param time the time in nanoseconds this runnable was scheduled to execute + * @param run the {@link Runnable} to execute + * @param id the unique, monotonic total-order index of this runnable */ - record TimedRunnable(TestWorker scheduler, long time, Runnable run, - long count) implements Comparable { + record TimedRunnable(TestWorker worker, long time, Runnable run, + long id) implements Comparable { @Override - public String toString() { - return String.format("TimedRunnable(time = %d, run = %s)", time, run.toString()); - } + public String toString() { + return String.format("TimedRunnable(time = %d, run = %s)", time, run.toString()); + } - @Override - public int compareTo(TimedRunnable o) { - if (time == o.time) { - return Long.compare(count, o.count); - } - return Long.compare(time, o.time); + @Override + public int compareTo(TimedRunnable o) { + if (time == o.time) { + return Long.compare(id, o.id); } + return Long.compare(time, o.time); } + } @Override public long now(@NonNull TimeUnit unit) { @@ -163,13 +167,22 @@ private void triggerActions(long targetTimeInNanoseconds) { queue.remove(current); // Only execute if not unsubscribed - if (!current.scheduler.disposed) { + if (!current.worker.disposed) { current.run.run(); } } time = targetTimeInNanoseconds; } + /** + * Returns the currently scheduled tasks waiting for execution in the internal queue. + * @return the number of {@link Runnable}s waiting to be executed via {@link #advanceTimeBy(long, TimeUnit)} + * or {@link #advanceTimeTo(long, TimeUnit)}. + */ + public int runnableCount() { + return queue.size(); + } + @NonNull @Override public Worker createWorker() { @@ -183,6 +196,8 @@ final class TestWorker extends Worker { @Override public void dispose() { disposed = true; + // make sure disposing the worker removes all tasks queued up, via ownership predicate + queue.removeIf(tr -> tr.worker() == this); } @Override diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimerTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimerTest.java index deee508260..51c10a9870 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimerTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimerTest.java @@ -13,14 +13,15 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.*; import java.util.concurrent.*; import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.Streamable; -import io.reactivex.rxjava4.schedulers.Schedulers; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.schedulers.*; import io.reactivex.rxjava4.testsupport.TestHelper; public class StreamableTimerTest extends StreamableBaseTest { @@ -159,4 +160,41 @@ public void basicSchedulederoDelay() throws Throwable { } } + @Test + public void finishStopsTimerActionScheduler() throws Throwable { + var scheduler = new TestScheduler(); + + var streamer = Streamable.timer(1, TimeUnit.MINUTES, scheduler).stream(new CompositeDisposable()); + + assertEquals(1, scheduler.runnableCount()); + + streamer.awaitFinish(); + + assertEquals(0, scheduler.runnableCount()); + } + + @Test + public void finishStopsTimerActionScheduledExecutor() throws Throwable { + var exec = (ScheduledThreadPoolExecutor)Executors.newScheduledThreadPool(1); + exec.setRemoveOnCancelPolicy(true); + try { + var streamer = Streamable.timer(1, TimeUnit.MINUTES, exec).stream(new CompositeDisposable()); + + streamer.awaitFinish(); + } finally { + assertEquals(0, exec.shutdownNow().size()); + } + } + + @Test + public void finishStopsTimerActionExecutor() throws Throwable { + var exec = Executors.newFixedThreadPool(1); + try { + var streamer = Streamable.timer(1, TimeUnit.MINUTES, exec).stream(new CompositeDisposable()); + + streamer.awaitFinish(); + } finally { + assertEquals(0, exec.shutdownNow().size()); + } + } }