4.x: API turn Completable concatX and mergeX into config-based (#8166)

Author: akarnokd

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava4.core.config;
+
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+/**
+ * Configuration record for Completable.concat() operators.
+ * @param delayError should the error propagation be delayed?
+ * @param prefetch the number of source sequences to request from a backpressured source
+ * @since 4.0.0
+ */
+public record CompletableConcatConfig(boolean delayError, int prefetch) {
+
+    /**
+     * The default configuration with no error delays and prefetch of 2.
+     */
+    public static final CompletableConcatConfig DEFAULT = new CompletableConcatConfig(false, 2);
+
+    /**
+     * The default configuration with error delays and prefetch of 2.
+     */
+    public static final CompletableConcatConfig DELAY_ERROR = new CompletableConcatConfig(true, 2);
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     */
+    public CompletableConcatConfig(boolean delayError) {
+        this(delayError, 2);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param prefetch the number of source sequences to request from a backpressured source
+     */
+    public CompletableConcatConfig(int prefetch) {
+        this(false, prefetch);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     * @param prefetch the number of source sequences to request from a backpressured source
+     */
+    public CompletableConcatConfig(boolean delayError, int prefetch) {
+        ObjectHelper.verifyPositive(prefetch, "prefetch");
+        this.delayError = delayError;
+        this.prefetch = prefetch;
+    }
+}

src/main/java/io/reactivex/rxjava4/core/config/CompletableConcatConfig.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava4.core.config;
+
+import io.reactivex.rxjava4.core.Flowable;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+/**
+ * Configuration record for Completable.merge() operators.
+ * @param delayErrors should the error propagation be delayed?
+ * @param maxConcurrency the number of source sequences run concurrently
+ * @since 4.0.0
+ */
+public record CompletableMergeConfig(boolean delayErrors, int maxConcurrency) {
+
+    /**
+     * The default config with no error delay and Flowable#bufferSize() as the maximum concurrency setting.
+     */
+    public static final CompletableMergeConfig DEFAULT = new CompletableMergeConfig(false, Flowable.bufferSize());
+
+    /**
+     * The default config with error delay and Flowable#bufferSize() as the maximum concurrency setting.
+     */
+    public static final CompletableMergeConfig DELAY_ERRORS = new CompletableMergeConfig(true, Flowable.bufferSize());
+
+    /**
+     * Constructs a configuration record.
+     * @param delayErrors should the error propagation be delayed?
+     */
+    public CompletableMergeConfig(boolean delayErrors) {
+        this(delayErrors, 2);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param maxConcurrency the number of source sequences run concurrently
+     */
+    public CompletableMergeConfig(int maxConcurrency) {
+        this(false, maxConcurrency);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayErrors should the error propagation be delayed?
+     * @param maxConcurrency the number of source sequences run concurrently
+     */
+    public CompletableMergeConfig(boolean delayErrors, int maxConcurrency) {
+        ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
+        this.delayErrors = delayErrors;
+        this.maxConcurrency = maxConcurrency;
+    }
+}

src/main/java/io/reactivex/rxjava4/core/config/CompletableMergeConfig.java

@@ -28,6 +28,7 @@
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.core.Observable;
+import io.reactivex.rxjava4.core.config.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.exceptions.*;
 import io.reactivex.rxjava4.functions.*;
@@ -321,7 +322,7 @@ public void concatObservablePrefetch() {
                 .repeat(10)
                 .doOnRequest(requested::add);
 
-        Completable c = Completable.concat(cs, 5);
+        Completable c = Completable.concat(cs, new CompletableConcatConfig(5));
 
         c.blockingAwait();
 
@@ -683,7 +684,7 @@ public void mergeObservableMaxConcurrent() {
                 .repeat(10)
                 .doOnRequest(requested::add);
 
-        Completable c = Completable.merge(cs, 5);
+        Completable c = Completable.merge(cs, new CompletableMergeConfig(5));
 
         c.blockingAwait();
 
@@ -693,14 +694,14 @@ public void mergeObservableMaxConcurrent() {
 
     @Test
     public void mergeDelayErrorEmpty() {
-        Completable c = Completable.mergeArrayDelayError();
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorSingleSource() {
-        Completable c = Completable.mergeArrayDelayError(normal.completable);
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS, normal.completable);
 
         c.blockingAwait();
 
@@ -709,14 +710,14 @@ public void mergeDelayErrorSingleSource() {
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorSingleSourceThrows() {
-        Completable c = Completable.mergeArrayDelayError(error.completable);
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS, error.completable);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorMultipleSources() {
-        Completable c = Completable.mergeArrayDelayError(normal.completable, normal.completable, normal.completable);
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS, normal.completable, normal.completable, normal.completable);
 
         c.blockingAwait();
 
@@ -725,7 +726,7 @@ public void mergeDelayErrorMultipleSources() {
 
     @Test
     public void mergeDelayErrorMultipleOneThrows() {
-        Completable c = Completable.mergeArrayDelayError(normal.completable, error.completable, normal.completable);
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS, normal.completable, error.completable, normal.completable);
 
         try {
             c.blockingAwait();
@@ -736,28 +737,28 @@ public void mergeDelayErrorMultipleOneThrows() {
 
     @Test(expected = NullPointerException.class)
     public void mergeDelayErrorMultipleOneIsNull() {
-        Completable c = Completable.mergeArrayDelayError(normal.completable, null);
+        Completable c = Completable.mergeArray(CompletableMergeConfig.DELAY_ERRORS, normal.completable, null);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorIterableEmpty() {
-        Completable c = Completable.mergeDelayError(Collections.<Completable>emptyList());
+        Completable c = Completable.merge(Collections.<Completable>emptyList(), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test(expected = NullPointerException.class)
     public void mergeDelayErrorIterableIteratorNull() {
-        Completable c = Completable.mergeDelayError((Iterable<Completable>) () -> null);
+        Completable c = Completable.merge((Iterable<Completable>) () -> null, CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorIterableSingle() {
-        Completable c = Completable.mergeDelayError(Collections.singleton(normal.completable));
+        Completable c = Completable.merge(Collections.singleton(normal.completable), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
 
@@ -766,7 +767,8 @@ public void mergeDelayErrorIterableSingle() {
 
     @Test
     public void mergeDelayErrorIterableMany() {
-        Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, normal.completable, normal.completable));
+        Completable c = Completable.merge(
+                Arrays.asList(normal.completable, normal.completable, normal.completable), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
 
@@ -775,14 +777,15 @@ public void mergeDelayErrorIterableMany() {
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorIterableOneThrows() {
-        Completable c = Completable.mergeDelayError(Collections.singleton(error.completable));
+        Completable c = Completable.merge(Collections.singleton(error.completable), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorIterableManyOneThrows() {
-        Completable c = Completable.mergeDelayError(Arrays.asList(normal.completable, error.completable, normal.completable));
+        Completable c = Completable.merge(
+                Arrays.asList(normal.completable, error.completable, normal.completable), CompletableMergeConfig.DELAY_ERRORS);
 
         try {
             c.blockingAwait();
@@ -793,44 +796,44 @@ public void mergeDelayErrorIterableManyOneThrows() {
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorIterableIterableThrows() {
-        Completable c = Completable.mergeDelayError((Iterable<Completable>) () -> {
+        Completable c = Completable.merge((Iterable<Completable>) () -> {
             throw new TestException();
-        });
+        }, CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorIterableIteratorHasNextThrows() {
-        Completable c = Completable.mergeDelayError(new IterableIteratorHasNextThrows());
+        Completable c = Completable.merge(new IterableIteratorHasNextThrows(), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorIterableIteratorNextThrows() {
-        Completable c = Completable.mergeDelayError(new IterableIteratorNextThrows());
+        Completable c = Completable.merge(new IterableIteratorNextThrows(), CompletableMergeConfig.DELAY_ERRORS);
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorObservableEmpty() {
-        Completable c = Completable.mergeDelayError(Flowable.<Completable>empty());
+        Completable c = Completable.merge(Flowable.<Completable>empty(), new CompletableMergeConfig(true));
 
         c.blockingAwait();
     }
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorObservableError() {
-        Completable c = Completable.mergeDelayError(Flowable.<Completable>error(TestException::new));
+        Completable c = Completable.merge(Flowable.<Completable>error(TestException::new), new CompletableMergeConfig(true));
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorObservableSingle() {
-        Completable c = Completable.mergeDelayError(Flowable.just(normal.completable));
+        Completable c = Completable.merge(Flowable.just(normal.completable), new CompletableMergeConfig(true));
 
         c.blockingAwait();
 
@@ -839,14 +842,14 @@ public void mergeDelayErrorObservableSingle() {
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorObservableSingleThrows() {
-        Completable c = Completable.mergeDelayError(Flowable.just(error.completable));
+        Completable c = Completable.merge(Flowable.just(error.completable), new CompletableMergeConfig(true));
 
         c.blockingAwait();
     }
 
     @Test
     public void mergeDelayErrorObservableMany() {
-        Completable c = Completable.mergeDelayError(Flowable.just(normal.completable).repeat(3));
+        Completable c = Completable.merge(Flowable.just(normal.completable).repeat(3), new CompletableMergeConfig(true));
 
         c.blockingAwait();
 
@@ -855,7 +858,7 @@ public void mergeDelayErrorObservableMany() {
 
     @Test(expected = TestException.class)
     public void mergeDelayErrorObservableManyOneThrows() {
-        Completable c = Completable.mergeDelayError(Flowable.just(normal.completable, error.completable));
+        Completable c = Completable.merge(Flowable.just(normal.completable, error.completable), new CompletableMergeConfig(true));
 
         c.blockingAwait();
     }
@@ -868,7 +871,7 @@ public void mergeDelayErrorObservableMaxConcurrent() {
                 .repeat(10)
                 .doOnRequest(requested::add);
 
-        Completable c = Completable.mergeDelayError(cs, 5);
+        Completable c = Completable.merge(cs, new CompletableMergeConfig(true, 5));
 
         c.blockingAwait();
 

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava4.core.config;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.Test;
+
+import io.reactivex.rxjava4.core.RxJavaTest;
+
+public class CompletableConcatConfigTest extends RxJavaTest {
+
+    @Test
+    public void validation() {
+        assertTrue(new CompletableConcatConfig(true).delayError(), "delayError - true");
+        assertFalse(new CompletableConcatConfig(false).delayError(), "delayError - false");
+        assertEquals(5, new CompletableConcatConfig(5).prefetch(), "prefetch - 5");
+        assertEquals(5, new CompletableConcatConfig(true, 5).prefetch(), "prefetch both - true, 5");
+        assertEquals(5, new CompletableConcatConfig(false, 5).prefetch(), "prefetch both - false, 5");
+        assertTrue(new CompletableConcatConfig(true, 5).delayError(), "delayError both - true, 5");
+        assertFalse(new CompletableConcatConfig(false, 5).delayError(), "delayError both - false, 5");
+    }
+}