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

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

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

@@ -178,7 +178,7 @@ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, lo
 
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
-        if (tasks.size() == 0) {
+        if (tasks.isEmpty()) {
             throw new IllegalArgumentException("The tasks parameter should contain at least one callable!");
         }
 
@@ -208,7 +208,7 @@ public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws Interrupt
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
             throws InterruptedException, ExecutionException, TimeoutException {
-        if (tasks.size() == 0) {
+        if (tasks.isEmpty()) {
             throw new IllegalArgumentException("The tasks parameter should contain at least one callable!");
         }
 

src/test/java/io/reactivex/rxjava4/flowable/FlowableCollectTest.java

@@ -59,7 +59,7 @@ public void collectToStringFlowable() {
             .collect(
                     StringBuilder::new,
                 (sb, v) -> {
-                if (sb.length() > 0) {
+                if (!sb.isEmpty()) {
                     sb.append("-");
                 }
                 sb.append(v);
@@ -177,7 +177,7 @@ public void collectToString() {
             .collect(
                     StringBuilder::new,
                 (sb, v) -> {
-                if (sb.length() > 0) {
+                if (!sb.isEmpty()) {
                     sb.append("-");
                 }
                 sb.append(v);

src/test/java/io/reactivex/rxjava4/internal/observers/QueueDrainObserverTest.java

@@ -23,7 +23,7 @@
 
 public class QueueDrainObserverTest extends RxJavaTest {
 
-    static final QueueDrainObserver<Integer, Integer, Integer> createUnordered(TestObserver<Integer> to, final Disposable d) {
+    static QueueDrainObserver<Integer, Integer, Integer> createUnordered(TestObserver<Integer> to, final Disposable d) {
         return new QueueDrainObserver<Integer, Integer, Integer>(to, new SpscArrayQueue<>(4)) {
             @Override
             public void onNext(Integer t) {
@@ -50,7 +50,7 @@ public void accept(Observer<? super Integer> a, Integer v) {
         };
     }
 
-    static final QueueDrainObserver<Integer, Integer, Integer> createOrdered(TestObserver<Integer> to, final Disposable d) {
+    static QueueDrainObserver<Integer, Integer, Integer> createOrdered(TestObserver<Integer> to, final Disposable d) {
         return new QueueDrainObserver<Integer, Integer, Integer>(to, new SpscArrayQueue<>(4)) {
             @Override
             public void onNext(Integer t) {

src/test/java/io/reactivex/rxjava4/internal/operators/completable/CompletableDetachTest.java

@@ -65,7 +65,7 @@ public void cancelDetaches() throws Exception {
             @Override
             protected void subscribeActual(CompletableObserver observer) {
                 observer.onSubscribe(wr.get());
-            };
+            }
         }
         .onTerminateDetach()
         .test();
@@ -93,7 +93,7 @@ protected void subscribeActual(CompletableObserver observer) {
                 observer.onSubscribe(wr.get());
                 observer.onComplete();
                 observer.onComplete();
-            };
+            }
         }
         .onTerminateDetach()
         .test();
@@ -119,7 +119,7 @@ protected void subscribeActual(CompletableObserver observer) {
                 observer.onSubscribe(wr.get());
                 observer.onError(new TestException());
                 observer.onError(new IOException());
-            };
+            }
         }
         .onTerminateDetach()
         .test();

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCombineLatestTest.java

@@ -895,9 +895,7 @@ public void combineLatestNArguments() throws Exception {
             params[i] = ArgsToString.INSTANCE;
 
             StringBuilder b = new StringBuilder();
-            for (int j = 0; j < i; j++) {
-                b.append('1');
-            }
+            b.repeat("1", i);
 
             ((Flowable)m.invoke(null, params)).test().assertResult(b.toString());
 
@@ -1050,7 +1048,7 @@ public void onErrorRace() {
 
                 TestHelper.race(r1, r2);
 
-                if (ts.errors().size() != 0) {
+                if (!ts.errors().isEmpty()) {
                     if (ts.errors().getFirst() instanceof CompositeException) {
                         ts.assertSubscribed()
                         .assertNotComplete()

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableConcatMapSchedulerTest.java

@@ -301,7 +301,7 @@ public void onNext(Integer t) {
                 if (counter.getAndIncrement() % 100 == 0) {
                     System.out.print("testIssue2890NoStackoverflow -> ");
                     System.out.println(counter.get());
-                };
+                }
             }
 
             @Override

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableConcatTest.java

@@ -720,7 +720,7 @@ public void onNext(Integer t) {
                 if (counter.getAndIncrement() % 100 == 0) {
                     System.out.print("testIssue2890NoStackoverflow -> ");
                     System.out.println(counter.get());
-                };
+                }
             }
 
             @Override

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableFromSourceTest.java

@@ -694,7 +694,7 @@ public void onNext(Integer v) {
 
             processor.subscribe(as);
 
-            t.setCancellable(as::dispose);;
+            t.setCancellable(as::dispose);
         }
 
         @Override

src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java

@@ -602,7 +602,7 @@ public String toString() {
 
     Flowable<Event> ASYNC_INFINITE_OBSERVABLE_OF_EVENT(final int numGroups, final AtomicInteger subscribeCounter, final AtomicInteger sentEventCounter) {
         return SYNC_INFINITE_OBSERVABLE_OF_EVENT(numGroups, subscribeCounter, sentEventCounter).subscribeOn(Schedulers.newThread());
-    };
+    }
 
     Flowable<Event> SYNC_INFINITE_OBSERVABLE_OF_EVENT(final int numGroups, final AtomicInteger subscribeCounter, final AtomicInteger sentEventCounter) {
         return Flowable.unsafeCreate(op -> {
@@ -620,7 +620,7 @@ Flowable<Event> SYNC_INFINITE_OBSERVABLE_OF_EVENT(final int numGroups, final Ato
             }
             op.onComplete();
         });
-    };
+    }
 
     @Test
     public void groupByOnAsynchronousSourceAcceptsMultipleSubscriptions() throws InterruptedException {