4.x: Convert from JUnit 4 to JUnit 6

build.gradle

@@ -16,11 +16,13 @@ ext {
     reactiveStreamsVersion = "1.0.4"
     junitVersion = "4.13.2"
     testNgVersion = "7.12.0"
-    mockitoVersion = "4.11.0"
+    mockitoVersion = "5.23.0"
+    mockitoJUnitVersion = "5.23.0"
     jmhLibVersion = "1.37"
     guavaVersion = "33.6.0-jre"
     jupiterVersion = "6.1.1"
     jupiterLauncherVersion = "6.1.1"
+    junitPioneerVersion = "2.3.0"
 }
 
 def releaseTag = System.getenv("BUILD_TAG")
@@ -38,8 +40,8 @@ repositories {
 }
 
 dependencies {
-    testImplementation "junit:junit:$junitVersion"
     testImplementation "org.mockito:mockito-core:$mockitoVersion"
+    testImplementation "org.mockito:mockito-junit-jupiter:$mockitoJUnitVersion"
 
     testImplementation "org.reactivestreams:reactive-streams-tck:$reactiveStreamsVersion"
     testImplementation "org.testng:testng:$testNgVersion"
@@ -52,8 +54,9 @@ dependencies {
     testImplementation "org.junit.platform:junit-platform-commons:$jupiterLauncherVersion"
     testImplementation "org.junit.platform:junit-platform-launcher:$jupiterLauncherVersion"
 
-    testRuntimeOnly "org.junit.vintage:junit-vintage-engine:$jupiterVersion"
     testRuntimeOnly "org.junit.platform:junit-platform-launcher:$jupiterLauncherVersion"  // already have this
+
+    testImplementation "org.junit-pioneer:junit-pioneer:$junitPioneerVersion"
 }
 
 // === Experimental JDK handling for Outreach Program ===

src/main/java/io/reactivex/rxjava4/core/Scheduler.java

@@ -451,15 +451,17 @@ public Disposable schedulePeriodically(@NonNull Runnable run, final long initial
             final long firstNowNanoseconds = now(TimeUnit.NANOSECONDS);
             final long firstStartInNanoseconds = firstNowNanoseconds + unit.toNanos(initialDelay);
 
-            Disposable d = schedule(new PeriodicTask(firstStartInNanoseconds, decoratedRun, firstNowNanoseconds, sd,
-                    periodInNanoseconds), initialDelay, unit);
+            var task = new PeriodicTask(firstStartInNanoseconds, decoratedRun, firstNowNanoseconds, sd,
+                    periodInNanoseconds);
+
+            Disposable d = schedule(task, initialDelay, unit);
 
             if (d == EmptyDisposable.INSTANCE) {
                 return d;
             }
             first.replace(d);
 
-            return sd;
+            return task;
         }
 
         /**
@@ -495,7 +497,7 @@ public Scheduler share() {
          * Holds state and logic to calculate when the next delayed invocation
          * of this task has to happen (accounting for clock drifts).
          */
-        final class PeriodicTask implements Runnable, SchedulerRunnableIntrospection {
+        final class PeriodicTask implements Runnable, SchedulerRunnableIntrospection, Disposable {
             @NonNull
             final Runnable decoratedRun;
             @NonNull
@@ -546,6 +548,16 @@ public void run() {
             public Runnable getWrappedRunnable() {
                 return this.decoratedRun;
             }
+
+            @Override
+            public void dispose() {
+                sd.dispose();
+            }
+
+            @Override
+            public boolean isDisposed() {
+                return sd.isDisposed();
+            }
         }
 
         /**

src/main/java/io/reactivex/rxjava4/internal/schedulers/ParallelScheduler.java

@@ -17,17 +17,19 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.Scheduler;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.Exceptions;
-import io.reactivex.rxjava4.internal.functions.Functions;
+import io.reactivex.rxjava4.internal.functions.*;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
+import io.reactivex.rxjava4.schedulers.SchedulerRunnableIntrospection;
 
 /**
  * Scheduler with a configurable fixed amount of thread-pools.
  * @since 4.0.0
  */
-public final class ParallelScheduler extends Scheduler {
+public final class ParallelScheduler extends Scheduler implements SchedulerMultiWorkerSupport {
 
     static final ScheduledExecutorService[] SHUTDOWN;
 
@@ -41,6 +43,10 @@ public final class ParallelScheduler extends Scheduler {
 
     final AtomicReference<ScheduledExecutorService[]> pool;
 
+    static final TrackingParallelWorker SHUTDOWN_TRACKING_WORKER;
+
+    static final NonTrackingParallelWorker SHUTDOWN_NON_TRACKING_WORKER;
+
     int n;
 
     /**
@@ -53,6 +59,9 @@ public final class ParallelScheduler extends Scheduler {
 
         REJECTING = Executors.newSingleThreadScheduledExecutor();
         REJECTING.shutdownNow();
+
+        SHUTDOWN_TRACKING_WORKER = new TrackingParallelWorker(REJECTING);
+        SHUTDOWN_NON_TRACKING_WORKER = new NonTrackingParallelWorker(REJECTING);
     }
 
     public ParallelScheduler(int parallelism, boolean tracking, ThreadFactory factory) {
@@ -128,14 +137,44 @@ public Worker createWorker() {
         return new NonTrackingParallelWorker(pick());
     }
 
+    @Override
+    public void createWorkers(int number, WorkerCallback callback) {
+        ObjectHelper.verifyPositive(number, "number > 0 required");
+        ScheduledExecutorService[] current = pool.get();
+        int c = current.length;
+        if (c == 0) {
+            for (int i = 0; i < number; i++) {
+                if (tracking) {
+                    callback.onWorker(i, SHUTDOWN_TRACKING_WORKER);
+                } else {
+                    callback.onWorker(i, SHUTDOWN_NON_TRACKING_WORKER);
+                }
+            }
+        } else {
+            int index = n % c;
+            for (int i = 0; i < number; i++) {
+                if (tracking) {
+                    callback.onWorker(i, new TrackingParallelWorker(current[index]));
+                } else {
+                    callback.onWorker(i, new NonTrackingParallelWorker(current[index]));
+                }
+                if (++index == c) {
+                    index = 0;
+                }
+            }
+            n = index;
+        }
+    }
+
     @Override
     public Disposable scheduleDirect(Runnable run) {
         ScheduledExecutorService exec = pick();
         if (exec == REJECTING) {
             return Disposable.disposed();
         }
         try {
-            return Disposable.fromFuture(exec.submit(RxJavaPlugins.onSchedule(run)));
+            var decoratedRun = RxJavaPlugins.onSchedule(run);
+            return createWrapper(exec.submit(decoratedRun), decoratedRun);
         } catch (RejectedExecutionException ex) {
             return Disposable.disposed();
         }
@@ -148,7 +187,8 @@ public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
             return Disposable.disposed();
         }
         try {
-            return Disposable.fromFuture(exec.schedule(RxJavaPlugins.onSchedule(run), delay, unit));
+            var decoratedRun = RxJavaPlugins.onSchedule(run);
+            return createWrapper(exec.schedule(decoratedRun, delay, unit), decoratedRun);
         } catch (RejectedExecutionException ex) {
             return Disposable.disposed();
         }
@@ -160,8 +200,12 @@ public Disposable schedulePeriodicallyDirect(Runnable run, long initialDelay, lo
         if (exec == REJECTING) {
             return Disposable.disposed();
         }
+        if (period <= 0) {
+            return super.schedulePeriodicallyDirect(run, initialDelay, period, unit);
+        }
         try {
-            return Disposable.fromFuture(exec.scheduleAtFixedRate(RxJavaPlugins.onSchedule(run), initialDelay, period, unit));
+            var decoratedRun = RxJavaPlugins.onSchedule(run);
+            return createWrapper(exec.scheduleAtFixedRate(decoratedRun, initialDelay, period, unit), decoratedRun);
         } catch (RejectedExecutionException ex) {
             return Disposable.disposed();
         }
@@ -217,7 +261,7 @@ public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
 
         // Not implementing a custom schedulePeriodically as it would require tracking the Future.
 
-        final class NonTrackingTask implements Callable<Object>, Disposable {
+        final class NonTrackingTask implements Callable<Object>, Disposable, SchedulerRunnableIntrospection {
 
             final Runnable actual;
 
@@ -249,6 +293,41 @@ public void dispose() {
             public boolean isDisposed() {
                 return disposed;
             }
+
+            @Override
+            public @NonNull Runnable getWrappedRunnable() {
+                return actual;
+            }
+        }
+    }
+
+    static DirectTaskWrapper createWrapper(Future<?> future, Runnable run) {
+        return new DirectTaskWrapper(run, future);
+    }
+
+    static final class DirectTaskWrapper implements Disposable, SchedulerRunnableIntrospection {
+        final Runnable actual;
+        final Future<?> future;
+        volatile boolean disposed;
+        DirectTaskWrapper(Runnable actual, Future<?> future) {
+            this.actual = actual;
+            this.future = future;
+        }
+
+        @Override
+        public void dispose() {
+            disposed = true;
+            future.cancel(true);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return disposed;
+        }
+
+        @Override
+        public @NonNull Runnable getWrappedRunnable() {
+            return actual;
         }
     }
 
@@ -309,7 +388,7 @@ public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
 
         static final class TrackedAction
         extends AtomicReference<DisposableContainer>
-        implements Callable<Object>, Disposable {
+        implements Callable<Object>, Disposable, SchedulerRunnableIntrospection {
 
             static final Future<?> FINISHED;
 
@@ -394,6 +473,11 @@ void setFuture(Future<?> d) {
                     }
                 }
             }
+
+            @Override
+            public @NonNull Runnable getWrappedRunnable() {
+                return actual;
+            }
         }
     }
 }

src/main/java/io/reactivex/rxjava4/internal/schedulers/ScheduledRunnable.java

@@ -17,11 +17,13 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
+import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
+import io.reactivex.rxjava4.schedulers.SchedulerRunnableIntrospection;
 
 public final class ScheduledRunnable extends AtomicReferenceArray<Object>
-implements Runnable, Callable<Object>, Disposable {
+implements Runnable, Callable<Object>, Disposable, SchedulerRunnableIntrospection {
 
     @Serial
     private static final long serialVersionUID = -6120223772001106981L;
@@ -176,4 +178,9 @@ public String toString() {
 
         return getClass().getSimpleName() + "[" + state + "]";
     }
+
+    @Override
+    public @NonNull Runnable getWrappedRunnable() {
+        return actual;
+    }
 }

src/main/java/io/reactivex/rxjava4/internal/schedulers/SharedScheduler.java

@@ -17,9 +17,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.Scheduler;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.internal.disposables.DisposableHelper;
+import io.reactivex.rxjava4.schedulers.SchedulerRunnableIntrospection;
 
 /**
  * A Scheduler implementation that uses one of the Workers from another Scheduler
@@ -122,7 +124,7 @@ public long now(TimeUnit unit) {
 
         static final class SharedAction
         extends AtomicReference<DisposableContainer>
-        implements Runnable, Disposable {
+        implements Runnable, Disposable, SchedulerRunnableIntrospection {
             @Serial
             private static final long serialVersionUID = 4949851341419870956L;
 
@@ -186,6 +188,11 @@ void setFuture(Disposable d) {
                     }
                 }
             }
+
+            @Override
+            public @NonNull Runnable getWrappedRunnable() {
+                return actual;
+            }
         }
     }
 }

src/test/java/io/reactivex/rxjava4/completable/CompletableIsolatedTest.java

@@ -13,10 +13,11 @@
 
 package io.reactivex.rxjava4.completable;
 
-import java.util.concurrent.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
-import org.junit.Assert;
 import org.junit.jupiter.api.parallel.Isolated;
 
 import io.reactivex.rxjava4.core.*;
@@ -57,7 +58,7 @@ public void onComplete() {
             }
         });
 
-        Assert.assertEquals("calls count mismatch", 6, calls.get());
-        Assert.assertNull("error present", err.get());
+        assertEquals(6, calls.get(), "calls count mismatch");
+        assertNull(err.get(), "error present");
     }
 }

src/test/java/io/reactivex/rxjava4/completable/CompletableRetryTest.java

@@ -13,11 +13,11 @@
 
 package io.reactivex.rxjava4.completable;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.exceptions.TestException;