4.x: Cleanup; build, safevarargs, small typos (#8183)

* 4.x: Cleanup; build, safevarargs, small typos

* Rename validators to CheckXXXTest & some try() cleanup

* isEmpty cleanup

* append fixes, typo fixes, unused removals

* DisposableContainer +tests

Author: akarnokd

build.gradle

@@ -179,17 +179,18 @@ jmh {
 
 def isCI = System.getenv("CI") != null
 def testLoggingConfig = ["skipped", "failed"]
+def parallelism = Runtime.runtime.availableProcessors()
+
 if (!isCI) {
     testLoggingConfig = ["failed"]
+    parallelism = parallelism.intdiv(2) ?: 1
 }
 
 test {
     maxHeapSize = "1200m"
-    if (System.getenv("CI") != null) {
-        maxParallelForks = Runtime.runtime.availableProcessors()
-    } else {
-        maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
-    }
+    maxParallelForks = parallelism
+    timeout = Duration.ofMinutes(30)
+
     useJUnitPlatform()
 }
 
@@ -206,19 +207,8 @@ 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 = 1
-
-    // Ensure JUnit-compatible XML output in the standard location
-    reports {
-        html.required = true
-        junitXml.required = true
-        junitXml.outputLocation = file("${buildDir}/test-results/test")   // ← important
-    }
+    maxParallelForks = parallelism
+    timeout = Duration.ofMinutes(30)
 
     // Ensure JUnit-compatible XML output in the standard location
     reports {
@@ -307,7 +297,7 @@ if (project.hasProperty("releaseMode")) {
   }
   mavenPublishing {
       // or when publishing to https://central.sonatype.com/
-      publishToMavenCentral(true)
+      publishToMavenCentral(true, DeploymentValidation.PUBLISHED)
 
       // signAllPublications()
   }

docs/Creating-Observables.md

@@ -99,8 +99,8 @@ When a consumer subscribes, the given `java.util.concurrent.Callable` is invoked
 ```java
 Callable<String> callable = () -> {
     System.out.println("Hello World!");
-    return "Hello World!");
-}
+    return "Hello World!";
+};
 
 Observable<String> observable = Observable.fromCallable(callable);
 

docs/What's-different-in-2.0.md

@@ -60,7 +60,7 @@ Observable.just(1).map(v -> null)
 This means that `Observable<Void>` can no longer emit any values but only terminate normally or with an exception. API designers may instead choose to define `Observable<Object>` with no guarantee on what `Object` will be (which should be irrelevant anyway). For example, if one needs a signaller-like source, a shared enum can be defined and its solo instance `onNext`'d:
 
 ```java
-enum Irrelevant { INSTANCE; }
+enum Irrelevant { INSTANCE}
 
 Observable<Object> source = Observable.create((ObservableEmitter<Object> emitter) -> {
    System.out.println("Side-effect 1");

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");
@@ -3262,8 +3257,8 @@ public final TestObserver<Void> test() {
     }
 
     /**
-     * Creates a {@link TestObserver} optionally in cancelled state, then subscribes it to this {@code Completable}.
-     * @param dispose if {@code true}, the {@code TestObserver} will be cancelled before subscribing to this
+     * Creates a {@link TestObserver} optionally in canceled state, then subscribes it to this {@code Completable}.
+     * @param dispose if {@code true}, the {@code TestObserver} will be canceled before subscribing to this
      * {@code Completable}.
      * <p>
      * <img width="640" height="499" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Completable.test.b.png" alt="">
@@ -3332,7 +3327,7 @@ public static Completable fromCompletionStage(@NonNull CompletionStage<?> stage)
      * The upstream can be canceled by converting the resulting {@code CompletionStage} into
      * {@link CompletableFuture} via {@link CompletionStage#toCompletableFuture()} and
      * calling {@link CompletableFuture#cancel(boolean)} on it.
-     * The upstream will be also cancelled if the resulting {@code CompletionStage} is converted to and
+     * The upstream will be also canceled if the resulting {@code CompletionStage} is converted to and
      * completed manually by {@link CompletableFuture#complete(Object)} or {@link CompletableFuture#completeExceptionally(Throwable)}.
      * <p>
      * {@code CompletionStage}s don't have a notion of emptiness and allow {@code null}s, therefore, one can either use

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/Streamable.java

@@ -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));
             }
@@ -208,7 +208,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
                 }
             }, emitter.canceller(), exec)) {
                 mainSource.await(emitter.canceller());
-            };
+            }
         }, exec);
     }
 

src/main/java/io/reactivex/rxjava4/exceptions/CompositeException.java

@@ -114,28 +114,22 @@ public synchronized Throwable getCause() { // NOPMD
                 for (Throwable inner : exceptions) {
                     int depth = 0;
                     while (inner != null) {
-                        for (int i = 0; i < depth; i++) {
-                            aggregateMessage.append("  ");
-                        }
+                        aggregateMessage.repeat("  ", Math.max(0, depth));
                         aggregateMessage.append("|-- ");
                         aggregateMessage.append(inner.getClass().getCanonicalName()).append(": ");
                         String innerMessage = inner.getMessage();
                         if (innerMessage != null && innerMessage.contains(separator)) {
                             aggregateMessage.append(separator);
                             for (String line : innerMessage.split(separator)) {
-                                for (int i = 0; i < depth + 2; i++) {
-                                    aggregateMessage.append("  ");
-                                }
+                                aggregateMessage.repeat("  ", Math.max(0, depth + 2));
                                 aggregateMessage.append(line).append(separator);
                             }
                         } else {
                             aggregateMessage.append(innerMessage);
                             aggregateMessage.append(separator);
                         }
 
-                        for (int i = 0; i < depth + 2; i++) {
-                            aggregateMessage.append("  ");
-                        }
+                        aggregateMessage.repeat("  ", Math.max(0, depth + 2));
                         StackTraceElement[] st = inner.getStackTrace();
                         if (st.length > 0) {
                             aggregateMessage.append("at ").append(st[0]).append(separator);
@@ -149,9 +143,7 @@ public synchronized Throwable getCause() { // NOPMD
                         } else {
                             inner = inner.getCause();
                             if (inner != null) {
-                                for (int i = 0; i < depth + 2; i++) {
-                                    aggregateMessage.append("  ");
-                                }
+                                aggregateMessage.repeat("  ", Math.max(0, depth + 2));
                                 aggregateMessage.append("|-- ");
                                 aggregateMessage.append("(cause not expanded again) ");
                                 aggregateMessage.append(inner.getClass().getCanonicalName()).append(": ");

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

@@ -297,7 +297,7 @@ else if (o instanceof WindowEndSubscriberIntercept) {
 
                         continue;
                     }
-                    else if (openDone && windows.size() == 0) {
+                    else if (openDone && windows.isEmpty()) {
                         upstream.cancel();
                         startSubscriber.cancel();
                         resources.dispose();

src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableWindowBoundarySelector.java

@@ -282,7 +282,7 @@ else if (o instanceof WindowEndObserverIntercept) {
 
                         continue;
                     }
-                    else if (openDone && windows.size() == 0) {
+                    else if (openDone && windows.isEmpty()) {
                         upstream.dispose();
                         startObserver.dispose();
                         resources.dispose();

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

@@ -23,7 +23,7 @@
 public record StreamableJust<T>(@NonNull T item) implements Streamable<T> {
 
     public StreamableJust(T item) {
-        this.item = Objects.requireNonNull(item, "item is null");;
+        this.item = Objects.requireNonNull(item, "item is null");
     }
 
     @Override