4.x: Streamable.timer to dispose the executor/scheduler upon finish

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java

@@ -96,7 +96,7 @@ void interrupedSleep(InterruptedException ex) {
         @Override
         public @NonNull CompletionStage<Void> finish() {
             waiter.complete(false);
-            lazySet(DisposableHelper.DISPOSED);
+            DisposableHelper.dispose(this);
             return FINISHED;
         }
 

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<TimedRunnable> {
+    record TimedRunnable(TestWorker worker, long time, Runnable run,
+                         long id) implements Comparable<TimedRunnable> {
 
         @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

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