4.x: Flowable API reduction: remaining operators & other related changes (#8200)

* 4.x: Flowable API reduction: remaining operators & other related changes

* Small cleanup and coverage

Author: akarnokd

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.0-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip
 networkTimeout=10000
 retries=0
 retryBackOffMs=500

src/jmh/java/io/reactivex/rxjava4/core/OperatorMergePerf.java

@@ -83,7 +83,7 @@ public void mergeNAsyncStreamsOfN(final InputThousand input) throws InterruptedE
     public void mergeTwoAsyncStreamsOfN(final InputThousand input) throws InterruptedException {
         PerfSubscriber o = input.newLatchedObserver();
         Flowable<Integer> ob = Flowable.range(0, input.size).subscribeOn(Schedulers.computation());
-        Flowable.merge(ob, ob).subscribe(o);
+        Flowable.mergeArray(ob, ob).subscribe(o);
         if (input.size == 1) {
             while (o.latch.getCount() != 0) { }
         } else {

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

@@ -220,7 +220,8 @@ public static Completable concatArray(@NonNull StandardBufferedConfig config, @N
             return wrap(sources[0]);
         }
         if (config.delayErrors()) {
-            return Flowable.fromArray(sources).concatMapCompletableDelayError(Functions.identity(), true, config.bufferSize());
+            return Flowable.fromArray(sources)
+                    .concatMapCompletable(Functions.identity(), config);
         }
         return RxJavaPlugins.onAssembly(new CompletableConcatArray(sources));
 
@@ -283,7 +284,6 @@ public static Completable concat(@NonNull Publisher<@NonNull ? extends Completab
      * @param config the operatar configuration record
      * @return the new {@code Completable} instance
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
-     * @throws IllegalArgumentException if {@code prefetch} is non-positive
      * @since 4.0.0
      */
     @CheckReturnValue
@@ -295,7 +295,8 @@ public static Completable concat(@NonNull Publisher<@NonNull ? extends Completab
         Objects.requireNonNull(sources, "sources is null");
         Objects.requireNonNull(config, "config is null");
         if (config.delayErrors()) {
-            return Flowable.fromPublisher(sources).concatMapCompletableDelayError(Functions.identity(), true, config.bufferSize());
+            return Flowable.fromPublisher(sources)
+                    .concatMapCompletable(Functions.identity(), config);
         }
         return RxJavaPlugins.onAssembly(new CompletableConcat(sources, config.bufferSize()));
     }
@@ -321,7 +322,8 @@ public static Completable concat(@NonNull Iterable<@NonNull ? extends Completabl
         Objects.requireNonNull(sources, "sources is null");
         Objects.requireNonNull(config, "config is null");
         if (config.delayErrors()) {
-            return Flowable.fromIterable(sources).concatMapCompletableDelayError(Functions.identity(), true, config.bufferSize());
+            return Flowable.fromIterable(sources)
+                    .concatMapCompletable(Functions.identity(), config);
         }
         return RxJavaPlugins.onAssembly(new CompletableConcatIterable(sources));
     }
@@ -870,7 +872,6 @@ public static Completable merge(@NonNull Publisher<@NonNull ? extends Completabl
      * @param config the configuration record for this operator
      * @return the new {@code Completable} instance
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
-     * @throws IllegalArgumentException if {@code maxConcurrency} is less than 1
      * @since 4.0.0
      */
     @CheckReturnValue
@@ -898,6 +899,7 @@ public static Completable merge(@NonNull Publisher<@NonNull ? extends Completabl
      * @param config the configuration record for this operator
      * @return the new {@code Completable} instance
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
+     * @since 4.0.0
      */
     @CheckReturnValue
     @NonNull

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

@@ -234,7 +234,6 @@ public abstract class Maybe<@NonNull T> implements MaybeSource<T> {
      * @param sources the {@code Publisher} of {@code MaybeSource} instances
      * @param config the configuration record for this operator
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
-     * @throws IllegalArgumentException if {@code prefetch} is non-positive
      * @return the new {@code Flowable} instance
      * @since 4.0.0
      */
@@ -521,7 +520,6 @@ public abstract class Maybe<@NonNull T> implements MaybeSource<T> {
      * @param config the configuration record for this operator
      * @return the new {@code Flowable} instance with the specified concatenation behavior
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
-     * @throws IllegalArgumentException if {@code maxConcurrency} is non-positive
      * @since 4.0.0
      */
     @SuppressWarnings({ "rawtypes", "unchecked" })
@@ -1135,7 +1133,6 @@ public abstract class Maybe<@NonNull T> implements MaybeSource<T> {
      * @param config the configuration record for this operator
      * @return the new {@code Flowable} instance
      * @throws NullPointerException if {@code sources} or {@code config} is {@code null}
-     * @throws IllegalArgumentException if {@code maxConcurrency} is non-positive
      * @since 4.0.0
      */
     @BackpressureSupport(BackpressureKind.FULL)
@@ -2686,32 +2683,6 @@ public final Maybe<T> delay(long time, @NonNull TimeUnit unit) {
         return delay(time, unit, Schedulers.computation(), false);
     }
 
-    /**
-     * Returns a {@code Maybe} that signals the events emitted by the current {@code Maybe} shifted forward in time by a
-     * specified delay.
-     * <p>
-     * <img width="640" height="340" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.delay.tb.png" alt="">
-     * <dl>
-     *   <dt><b>Scheduler:</b></dt>
-     *   <dd>This version of {@code delay} operates by default on the {@code computation} {@link Scheduler}.</dd>
-     * </dl>
-     *
-     * @param time the delay to shift the source by
-     * @param unit the {@link TimeUnit} in which {@code time} is defined
-     * @param delayError if {@code true}, both success and error signals are delayed. if {@code false}, only success signals are delayed.
-     * @return the new {@code Maybe} instance
-     * @throws NullPointerException if {@code unit} is {@code null}
-     * @see <a href="http://reactivex.io/documentation/operators/delay.html">ReactiveX operators documentation: Delay</a>
-     * @see #delay(long, TimeUnit, Scheduler, boolean)
-     * @since 3.0.0
-     */
-    @CheckReturnValue
-    @SchedulerSupport(SchedulerSupport.COMPUTATION)
-    @NonNull
-    public final Maybe<T> delay(long time, @NonNull TimeUnit unit, boolean delayError) {
-        return delay(time, unit, Schedulers.computation(), delayError);
-    }
-
     /**
      * Returns a {@code Maybe} that signals the events emitted by the current {@code Maybe} shifted forward in time by a
      * specified delay.