4.x: Observable.concat API reduction

src/main/java/io/reactivex/rxjava4/internal/util/ErrorMode.java → src/main/java/io/reactivex/rxjava4/core/ErrorMode.java

@@ -11,10 +11,11 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.rxjava4.internal.util;
+package io.reactivex.rxjava4.core;
 
 /**
  * Indicates when an error from the main source should be reported.
+ * @since 4.0.0
  */
 public enum ErrorMode {
     /** Report the error immediately, cancelling the active inner source. */

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

@@ -31,7 +31,6 @@
 import io.reactivex.rxjava4.internal.operators.maybe.*;
 import io.reactivex.rxjava4.internal.operators.mixed.*;
 import io.reactivex.rxjava4.internal.operators.observable.ObservableElementAtMaybe;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
 import io.reactivex.rxjava4.observers.TestObserver;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.schedulers.*;

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

@@ -33,7 +33,6 @@
 import io.reactivex.rxjava4.internal.operators.mixed.*;
 import io.reactivex.rxjava4.internal.operators.observable.ObservableSingleSingle;
 import io.reactivex.rxjava4.internal.operators.single.*;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
 import io.reactivex.rxjava4.observers.TestObserver;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.schedulers.*;

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

@@ -54,9 +54,7 @@ public MaybeConcatConfig(int prefetch) {
      * @param delayError should the error propagation be delayed?
      * @param prefetch the number of source sequences to request from a backpressured source
      */
-    public MaybeConcatConfig(boolean delayError, int prefetch) {
+    public MaybeConcatConfig {
         ObjectHelper.verifyPositive(prefetch, "prefetch");
-        this.delayError = delayError;
-        this.prefetch = prefetch;
     }
 }

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

@@ -0,0 +1,68 @@
+/*
+ * 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.annotations.NonNull;
+import io.reactivex.rxjava4.core.Observable;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+import io.reactivex.rxjava4.core.ErrorMode;
+
+/**
+ * Configuration record for Observable.concat() operators.
+ * @param errorMode how to handle when errors appear from the inner or outer sources
+ * @param bufferSize the expected number of outer sources to buffer while processing an inner source
+ * @since 4.0.0
+ */
+public record ObservableConcatConfig(@NonNull ErrorMode errorMode, int bufferSize) {
+
+    /**
+     * The default configuration with no error delays and bufferSize of {@link Observable#bufferSize()}.
+     */
+    public static final ObservableConcatConfig DEFAULT = new ObservableConcatConfig(ErrorMode.IMMEDIATE, Observable.bufferSize());
+
+    /**
+     * The default configuration with error delays and bufferSize of {@link Observable#bufferSize()}.
+     */
+    public static final ObservableConcatConfig DELAY_ERROR = new ObservableConcatConfig(ErrorMode.END, Observable.bufferSize());
+
+    /**
+     * The default configuration with error delays and bufferSize of {@link Observable#bufferSize()}.
+     */
+    public static final ObservableConcatConfig DELAY_ERROR_BOUNDARY = new ObservableConcatConfig(ErrorMode.BOUNDARY, Observable.bufferSize());
+
+    /**
+     * Constructs a configuration record.
+     * @param errorMode how to handle when errors appear from the inner or outer sources
+     */
+    public ObservableConcatConfig(@NonNull ErrorMode errorMode) {
+        this(errorMode, Observable.bufferSize());
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of outer sources to buffer while processing an inner source
+     */
+    public ObservableConcatConfig(int bufferSize) {
+        this(ErrorMode.IMMEDIATE, bufferSize);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param errorMode how to handle when errors appear from the inner or outer sources
+     * @param bufferSize the expected number of outer sources to buffer while processing an inner source
+     */
+    public ObservableConcatConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+    }
+}

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

@@ -18,7 +18,7 @@
 import io.reactivex.rxjava4.core.Flowable;
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.operators.flowable.FlowableConcatMapEager.ConcatMapEagerDelayErrorSubscriber;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 
 /**
  * ConcatMapEager which works with an arbitrary Publisher source.

src/main/java/io/reactivex/rxjava4/internal/operators/mixed/ConcatMapXMainObserver.java

@@ -15,6 +15,7 @@
 
 import java.util.concurrent.atomic.AtomicInteger;
 
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.core.Observer;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.internal.disposables.DisposableHelper;

src/main/java/io/reactivex/rxjava4/internal/operators/mixed/ConcatMapXMainSubscriber.java

@@ -17,6 +17,7 @@
 
 import static java.util.concurrent.Flow.*;
 
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.core.FlowableSubscriber;
 import io.reactivex.rxjava4.exceptions.*;
 import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper;

src/main/java/io/reactivex/rxjava4/internal/operators/mixed/FlowableConcatMapMaybePublisher.java

@@ -18,7 +18,7 @@
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.operators.mixed.FlowableConcatMapMaybe.ConcatMapMaybeSubscriber;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 
 /**
  * Maps each upstream item into a {@link MaybeSource}, subscribes to them one after the other terminates

src/main/java/io/reactivex/rxjava4/internal/operators/mixed/FlowableConcatMapSinglePublisher.java

@@ -18,7 +18,7 @@
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.operators.mixed.FlowableConcatMapSingle.ConcatMapSingleSubscriber;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 
 /**
  * Maps each upstream item into a {@link SingleSource}, subscribes to them one after the other terminates

src/main/java/io/reactivex/rxjava4/internal/operators/parallel/ParallelConcatMap.java

@@ -17,7 +17,7 @@
 
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.operators.flowable.FlowableConcatMap;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.parallel.ParallelFlowable;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 

src/test/java/io/reactivex/rxjava4/core/config/ObservableConcatConfigTest.java

@@ -0,0 +1,37 @@
+/*
+ * 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.ErrorMode;
+import io.reactivex.rxjava4.core.RxJavaTest;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class ObservableConcatConfigTest extends RxJavaTest {
+
+    @Test
+    public void validation() {
+        assertEquals(ErrorMode.IMMEDIATE, new ObservableConcatConfig(ErrorMode.IMMEDIATE).errorMode(), "errorMode - true");
+        assertEquals(ErrorMode.BOUNDARY, new ObservableConcatConfig(ErrorMode.BOUNDARY).errorMode(), "errorMode - false");
+        assertEquals(ErrorMode.END, new ObservableConcatConfig(ErrorMode.END).errorMode(), "errorMode - false");
+        assertEquals(5, new ObservableConcatConfig(5).bufferSize(), "bufferSize - 5");
+        assertEquals(5, new ObservableConcatConfig(ErrorMode.IMMEDIATE, 5).bufferSize(), "bufferSize both - true, 5");
+        assertEquals(5, new ObservableConcatConfig(ErrorMode.BOUNDARY, 5).bufferSize(), "bufferSize both - false, 5");
+        assertEquals(5, new ObservableConcatConfig(ErrorMode.END, 5).bufferSize(), "bufferSize both - false, 5");
+        assertEquals(ErrorMode.IMMEDIATE, new ObservableConcatConfig(ErrorMode.IMMEDIATE, 5).errorMode(), "errorMode both - true, 5");
+        assertEquals(ErrorMode.BOUNDARY, new ObservableConcatConfig(ErrorMode.BOUNDARY, 5).errorMode(), "errorMode both - false, 5");
+        assertEquals(ErrorMode.END, new ObservableConcatConfig(ErrorMode.END, 5).errorMode(), "errorMode both - false, 5");
+    }
+}

src/test/java/io/reactivex/rxjava4/internal/operators/mixed/FlowableConcatMapMaybeTest.java

@@ -29,7 +29,7 @@
 import io.reactivex.rxjava4.internal.functions.Functions;
 import io.reactivex.rxjava4.internal.operators.mixed.FlowableConcatMapMaybe.ConcatMapMaybeSubscriber;
 import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.processors.*;
 import io.reactivex.rxjava4.schedulers.Schedulers;

src/test/java/io/reactivex/rxjava4/internal/operators/mixed/FlowableConcatMapSingleTest.java

@@ -28,7 +28,7 @@
 import io.reactivex.rxjava4.internal.functions.Functions;
 import io.reactivex.rxjava4.internal.operators.mixed.FlowableConcatMapSingle.ConcatMapSingleSubscriber;
 import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.processors.*;
 import io.reactivex.rxjava4.subjects.SingleSubject;

src/test/java/io/reactivex/rxjava4/internal/operators/mixed/ObservableConcatMapMaybeTest.java

@@ -27,7 +27,7 @@
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.functions.Functions;
 import io.reactivex.rxjava4.internal.operators.mixed.ObservableConcatMapMaybe.ConcatMapMaybeMainObserver;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.observers.TestObserver;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.schedulers.Schedulers;

src/test/java/io/reactivex/rxjava4/internal/operators/mixed/ObservableConcatMapSingleTest.java

@@ -26,7 +26,7 @@
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.functions.Functions;
 import io.reactivex.rxjava4.internal.operators.mixed.ObservableConcatMapSingle.ConcatMapSingleMainObserver;
-import io.reactivex.rxjava4.internal.util.ErrorMode;
+import io.reactivex.rxjava4.core.ErrorMode;
 import io.reactivex.rxjava4.observers.TestObserver;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.subjects.*;