4.x Cleanup: isEmpty, general intellij cleanups

build.gradle

@@ -120,7 +120,7 @@ javadoc {
 moduleConfig {
     moduleInfoPath = 'src/main/module/module-info.java'
     multiReleaseVersion = 9
-	version = project.version
+    version = project.version
 }
 
 import aQute.bnd.gradle.Bundle   // ← important import
@@ -178,18 +178,19 @@ jmh {
 }
 
 def isCI = System.getenv("CI") != null
-def testLoggingConfig = ["skipped", "failed"]
+def parallelForks = Runtime.getRuntime().availableProcessors();
+def testLoggingConfig = ["skipped", "failed" /*, "started", "passed" */]
 if (!isCI) {
     testLoggingConfig = ["failed"]
+    parallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1;
+} else {
+    parallelForks = 1 // for now
 }
 
 test {
     maxHeapSize = "1200m"
-    if (System.getenv("CI") != null) {
-        maxParallelForks = Runtime.runtime.availableProcessors()
-    } else {
-        maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
-    }
+    timeout = Duration.ofMinutes(15)   // or 30, whatever is reasonable for your suite
+    maxParallelForks = parallelForks
     useJUnitPlatform()
 }
 
@@ -206,11 +207,7 @@ tasks.register('testNG', Test) {
     useTestNG()
 
     maxHeapSize = "1200m"
-    if (System.getenv("CI") != null) {
-        maxParallelForks = Runtime.runtime.availableProcessors()
-    } else {
-        maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
-    }
+    maxParallelForks = parallelForks
     // maxParallelForks = 1
 
     // Ensure JUnit-compatible XML output in the standard location
@@ -220,13 +217,6 @@ tasks.register('testNG', Test) {
         junitXml.outputLocation = file("${buildDir}/test-results/test")   // ← important
     }
 
-    // Ensure JUnit-compatible XML output in the standard location
-    reports {
-        html.required = true
-        junitXml.required = true
-        junitXml.outputLocation = file("${buildDir}/test-results/test")   // ← important
-    }
-
     // Optional: fine-tune includes/excludes
     // include '**/*NgTest.class', '**/*TestNG.class'
     // exclude '**/*JUnitTest.class'

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

@@ -122,7 +122,6 @@ public abstract class Completable implements CompletableSource {
     @CheckReturnValue
     @NonNull
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SafeVarargs
     public static Completable ambArray(@NonNull CompletableSource... sources) {
         Objects.requireNonNull(sources, "sources is null");
         if (sources.length == 0) {
@@ -190,7 +189,6 @@ public static Completable complete() {
     @CheckReturnValue
     @NonNull
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SafeVarargs
     public static Completable concatArray(@NonNull CompletableSource... sources) {
         return concatArray(CompletableConcatConfig.DEFAULT, sources);
     }
@@ -212,7 +210,6 @@ public static Completable concatArray(@NonNull CompletableSource... sources) {
     @CheckReturnValue
     @NonNull
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SafeVarargs
     public static Completable concatArray(@NonNull CompletableConcatConfig config, @NonNull CompletableSource... sources) {
         Objects.requireNonNull(sources, "sources is null");
         Objects.requireNonNull(config, "config is null");
@@ -775,7 +772,6 @@ public static Completable fromSupplier(@NonNull Supplier<?> supplier) {
     @CheckReturnValue
     @NonNull
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SafeVarargs
     public static Completable mergeArray(@NonNull CompletableSource... sources) {
         return mergeArray(CompletableMergeConfig.DEFAULT, sources);
     }
@@ -906,7 +902,6 @@ public static Completable merge(@NonNull Publisher<@NonNull ? extends Completabl
     @CheckReturnValue
     @NonNull
     @SchedulerSupport(SchedulerSupport.NONE)
-    @SafeVarargs
     public static Completable mergeArray(@NonNull CompletableMergeConfig config, @NonNull CompletableSource... sources) {
         Objects.requireNonNull(sources, "sources is null");
         Objects.requireNonNull(config, "config is null");

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

@@ -86,7 +86,7 @@ public CompletionStageDisposable(@NonNull CompletionStage<T> stage, @NonNull Dis
      * Await the completion of the current stage.
      */
     public void await() {
-        state.lazySet(true);;
+        state.lazySet(true);
         AwaitCoordinatorStatic.await(stage);
     }
 
@@ -95,15 +95,15 @@ public void await() {
      * @param canceller the canceller link
      */
     public void await(DisposableContainer canceller) {
-        state.lazySet(true);;
+        state.lazySet(true);
         AwaitCoordinatorStatic.await(stage, canceller);
     }
 
     /**
      * Indicate this instance is deliberately not awaiting its stage.
      */
     public void ignore() {
-        state.lazySet(true);;
+        state.lazySet(true);
     }
 
     @Override

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

@@ -2467,18 +2467,13 @@ public static int bufferSize() {
         Objects.requireNonNull(source, "source is null");
         Objects.requireNonNull(strategy, "strategy is null");
         Flowable<T> f = new FlowableFromObservable<>(source);
-        switch (strategy) {
-            case DROP:
-                return f.onBackpressureDrop();
-            case LATEST:
-                return f.onBackpressureLatest();
-            case MISSING:
-                return f;
-            case ERROR:
-                return RxJavaPlugins.onAssembly(new FlowableOnBackpressureError<>(f));
-            default:
-                return f.onBackpressureBuffer();
-        }
+        return switch (strategy) {
+            case DROP -> f.onBackpressureDrop();
+            case LATEST -> f.onBackpressureLatest();
+            case MISSING -> f;
+            case ERROR -> RxJavaPlugins.onAssembly(new FlowableOnBackpressureError<>(f));
+            default -> f.onBackpressureBuffer();
+        };
     }
 
     /**
@@ -9412,7 +9407,7 @@ public final Flowable<T> delaySubscription(long time, @NonNull TimeUnit unit, @N
     @SchedulerSupport(SchedulerSupport.NONE)
     @NonNull
     public final Flowable<T> distinct() {
-        return distinct(Functions.identity(), Functions.<T>createHashSet());
+        return distinct(Functions.identity(), Functions.createHashSet());
     }
 
     /**
@@ -14526,7 +14521,7 @@ public final Flowable<T> retryUntil(@NonNull BooleanSupplier stop) {
      * <img width="640" height="430" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/retryWhen.f.v3.png" alt="">
      * <p>
      * Example:
-     *
+     * <br>
      * This retries 3 times, each time incrementing the number of seconds it waits.
      *
      * <pre><code>
@@ -15632,7 +15627,7 @@ public final Flowable<T> startWithIterable(@NonNull Iterable<? extends T> items)
     @BackpressureSupport(BackpressureKind.FULL)
     public final Flowable<T> startWith(@NonNull CompletableSource other) {
         Objects.requireNonNull(other, "other is null");
-        return Flowable.concat(Completable.wrap(other).<T>toFlowable(), this);
+        return Flowable.concat(Completable.wrap(other).toFlowable(), this);
     }
 
     /**

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

@@ -35,7 +35,6 @@ public interface FlowableSubscriber<@NonNull T> extends Subscriber<T> {
      * calling {@link Subscription#request(long)}. In practice this means
      * no initialization should happen after the {@code request()} call and
      * additional behavior is thread safe in respect to {@code onNext}.
-     *
      * {@inheritDoc}
      */
     @Override

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

@@ -4358,7 +4358,7 @@ public final Maybe<T> retryUntil(@NonNull BooleanSupplier stop) {
      * <img width="640" height="405" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.retryWhen.png" alt="">
      * <p>
      * Example:
-     *
+     * <br>
      * This retries 3 times, each time incrementing the number of seconds it waits.
      *
      * <pre><code>
@@ -4476,7 +4476,7 @@ public final void safeSubscribe(@NonNull MaybeObserver<? super T> observer) {
     @BackpressureSupport(BackpressureKind.FULL)
     public final Flowable<T> startWith(@NonNull CompletableSource other) {
         Objects.requireNonNull(other, "other is null");
-        return Flowable.concat(Completable.wrap(other).<T>toFlowable(), toFlowable());
+        return Flowable.concat(Completable.wrap(other).toFlowable(), toFlowable());
     }
 
     /**
@@ -5391,7 +5391,7 @@ public final Maybe<T> unsubscribeOn(@NonNull Scheduler scheduler) {
     /**
      * Waits until this and the other {@link MaybeSource} signal a success value then applies the given {@link BiFunction}
      * to those values and emits the {@code BiFunction}'s resulting value to downstream.
-     *
+     * <br>
      * <img width="640" height="451" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.zipWith.png" alt="">
      *
      * <p>If either this or the other {@code MaybeSource} is empty or signals an error, the resulting {@code Maybe} will

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

@@ -11899,7 +11899,7 @@ public final Observable<T> startWithIterable(@NonNull Iterable<? extends T> item
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> startWith(@NonNull CompletableSource other) {
         Objects.requireNonNull(other, "other is null");
-        return Observable.concatArray(Completable.wrap(other).<T>toObservable(), this);
+        return Observable.concatArray(Completable.wrap(other).toObservable(), this);
     }
 
     /**

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

@@ -60,7 +60,7 @@
  * </ul>
  * From the {@code Observable}'s perspective, an {@code Observer} is the end consumer thus it is the {@code Observer}'s
  * responsibility to handle the error case and signal it "further down". This means unreliable code in the {@code onXXX}
- * methods should be wrapped into `try-catch`es, specifically in {@link #onError(Throwable)} or {@link #onComplete()}, and handled there
+ * methods should be wrapped into `try-catches, specifically in {@link #onError(Throwable)} or {@link #onComplete()}, and handled there
  * (for example, by logging it or presenting the user with an error dialog). However, if the error would be thrown from
  * {@link #onNext(Object)}, <a href="https://github.com/reactive-streams/reactive-streams-jvm#2.13">Rule 2.13</a> mandates
  * the implementation calls {@link Disposable#dispose()} and signals the exception in a way that is adequate to the target context,

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

@@ -4135,7 +4135,7 @@ public final void safeSubscribe(@NonNull SingleObserver<? super T> observer) {
     @BackpressureSupport(BackpressureKind.FULL)
     public final Flowable<T> startWith(@NonNull CompletableSource other) {
         Objects.requireNonNull(other, "other is null");
-        return Flowable.concat(Completable.wrap(other).<T>toFlowable(), toFlowable());
+        return Flowable.concat(Completable.wrap(other).toFlowable(), toFlowable());
     }
 
     /**
@@ -4488,7 +4488,7 @@ public final Single<T> subscribeOn(@NonNull Scheduler scheduler) {
      * <p>
      * If the current {@code Single} fails, the resulting {@code Single} will
      * pass along the signal to the downstream. To measure the time to error,
-     * use {@link #materialize()} and apply {@link #timeInterval()}.
+     * use {@link #materialize()} and apply {@code timeInterval()}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timeInterval} uses the {@code computation} {@link Scheduler}
@@ -4514,7 +4514,7 @@ public final Single<Timed<T>> timeInterval() {
      * <p>
      * If the current {@code Single} fails, the resulting {@code Single} will
      * pass along the signal to the downstream. To measure the time to error,
-     * use {@link #materialize()} and apply {@link #timeInterval(Scheduler)}.
+     * use {@link #materialize()} and apply {@code timeInterval(Scheduler)}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timeInterval} uses the provided {@link Scheduler}
@@ -4570,7 +4570,7 @@ public final Single<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
      * <p>
      * If the current {@code Single} is empty or fails, the resulting {@code Single} will
      * pass along the signals to the downstream. To measure the time to termination,
-     * use {@link #materialize()} and apply {@link #timeInterval(TimeUnit, Scheduler)}.
+     * use {@link #materialize()} and apply {@code #timeInterval(TimeUnit, Scheduler)}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timeInterval} uses the provided {@link Scheduler}
@@ -4601,7 +4601,7 @@ public final Single<Timed<T>> timeInterval(@NonNull TimeUnit unit, @NonNull Sche
      * <p>
      * If the current {@code Single} is empty or fails, the resulting {@code Single} will
      * pass along the signals to the downstream. To get the timestamp of the error,
-     * use {@link #materialize()} and apply {@link #timestamp()}.
+     * use {@link #materialize()} and apply {@code timestamp()}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timestamp} uses the {@code computation} {@code Scheduler}
@@ -4627,7 +4627,7 @@ public final Single<Timed<T>> timestamp() {
      * <p>
      * If the current {@code Single} is empty or fails, the resulting {@code Single} will
      * pass along the signals to the downstream. To get the timestamp of the error,
-     * use {@link #materialize()} and apply {@link #timestamp(Scheduler)}.
+     * use {@link #materialize()} and apply {@code #timestamp(Scheduler)}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timestamp} uses the provided {@code Scheduler}
@@ -4655,7 +4655,7 @@ public final Single<Timed<T>> timestamp(@NonNull Scheduler scheduler) {
      * <p>
      * If the current {@code Single} is empty or fails, the resulting {@code Single} will
      * pass along the signals to the downstream. To get the timestamp of the error,
-     * use {@link #materialize()} and apply {@link #timestamp(TimeUnit)}.
+     * use {@link #materialize()} and apply {@code timestamp(TimeUnit)}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timestamp} uses the {@code computation} {@code Scheduler},
@@ -4683,7 +4683,7 @@ public final Single<Timed<T>> timestamp(@NonNull TimeUnit unit) {
      * <p>
      * If the current {@code Single} is empty or fails, the resulting {@code Single} will
      * pass along the signals to the downstream. To get the timestamp of the error,
-     * use {@link #materialize()} and apply {@link #timestamp(TimeUnit, Scheduler)}.
+     * use {@link #materialize()} and apply {@code timestamp(TimeUnit, Scheduler)}.
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code timestamp} uses the provided {@code Scheduler},

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

@@ -115,7 +115,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
     @NonNull
     static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNull ExecutorService executor) {
         Objects.requireNonNull(source, "source is null");
-        return new StreamableFromPublisher<T>(source, executor);
+        return new StreamableFromPublisher<>(source, executor);
     }
 
     /**
@@ -186,7 +186,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
             for(var stage : stages) {
                 list.add(stage);
             }
-            while (list.size() != 0) {
+            while (!list.isEmpty()) {
                 var winner = AwaitCoordinatorStatic.awaitFirstIndex(list, emitter.canceller());
                 emitter.emit((CompletionStage<T>)list.remove(winner));
             }
@@ -361,7 +361,7 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
             return null;
         });
         canceller.add(Disposable.fromFuture(future));
-        return new CompletionStageDisposable<Void>(StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
+        return new CompletionStageDisposable<>(StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
     }
 
     /**

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

@@ -22,7 +22,7 @@
 
 /**
  * A realized stream which can then be consumed asynchronously in steps.
- * Think of it as the {@IAsyncEnumerator} of the Java world. Runs best on Virtual Threads.
+ * Think of it as the {@code IAsyncEnumerator} of the Java world. Runs best on Virtual Threads.
  * <p>
  * To make sure you can run finish, use {@link DisposableContainer#clear()} or {@link DisposableContainer#reset()}
  * to get rid of all previous registered disposables. finish() will create its own, and if that
@@ -93,8 +93,10 @@ default void close() {
      */
     default Streamer<T> finishVia(@NonNull DisposableContainer canceller) {
         Objects.requireNonNull(canceller, "canceller is null");
-        if (this instanceof StreamerFinishViaDisposableContainerCanceller<T> augment) {
-            if (augment.streamer == this && augment.canceller == canceller) {
+        if (this instanceof StreamerFinishViaDisposableContainerCanceller<T>(
+                Streamer<T> streamer, DisposableContainer canceller1
+        )) {
+            if (streamer == this && canceller1 == canceller) {
                 // DO not rewrap!
                 return this;
             }
@@ -107,7 +109,7 @@ default Streamer<T> finishVia(@NonNull DisposableContainer canceller) {
      * Augments the base streamer with a canceller so that it can be injected at the various await calls.
      * @param <T> the element type of the stream
      */
-    static record StreamerFinishViaDisposableContainerCanceller<T>(
+    record StreamerFinishViaDisposableContainerCanceller<T>(
             @NonNull Streamer<T> streamer, @NonNull DisposableContainer canceller)
     implements Streamer<T> {
 
@@ -141,7 +143,7 @@ default Streamer<T> hide() {
      * Hides the identity of the Streamer for debug or deoptimization purposes.
      * @param <T> the element type of the streamer
      */
-    static record HiddenStreamer<T>(@NonNull Streamer<T> streamer) implements Streamer<T> {
+    record HiddenStreamer<T>(@NonNull Streamer<T> streamer) implements Streamer<T> {
 
         @Override
         public @NonNull CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation) {

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableConcatMap.java

@@ -47,14 +47,11 @@ public FlowableConcatMap(Flowable<T> source,
 
     public static <T, R> Subscriber<T> subscribe(Subscriber<? super R> s, Function<? super T, ? extends Publisher<? extends R>> mapper,
             int prefetch, ErrorMode errorMode) {
-        switch (errorMode) {
-        case BOUNDARY:
-            return new ConcatMapDelayed<>(s, mapper, prefetch, false);
-        case END:
-            return new ConcatMapDelayed<>(s, mapper, prefetch, true);
-        default:
-            return new ConcatMapImmediate<>(s, mapper, prefetch);
-        }
+        return switch (errorMode) {
+            case BOUNDARY -> new ConcatMapDelayed<>(s, mapper, prefetch, false);
+            case END -> new ConcatMapDelayed<>(s, mapper, prefetch, true);
+            default -> new ConcatMapImmediate<>(s, mapper, prefetch);
+        };
     }
 
     @Override