4.x: Cleanu API: Observable flatMap*, groupBy, observeOn (#8192)

Author: akarnokd

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

@@ -28,14 +28,14 @@
 public record FlatMapConfig(boolean delayErrors, int maxConcurrency, int bufferSize) {
 
     /**
-     * The default config with no error delay and Flowable#bufferSize() as the maximum concurrency setting.
+     * The default configuration with no error delay and Flowable#bufferSize() as the maximum concurrency setting.
      */
-    public static final FlatMapConfig DEFAULT = new FlatMapConfig(false, Flowable.bufferSize());
+    public static final FlatMapConfig DEFAULT = new FlatMapConfig(false);
 
     /**
-     * The default config with error delay and Flowable#bufferSize() as the maximum concurrency setting.
+     * The default configuration with error delay and Flowable#bufferSize() as the maximum concurrency setting.
      */
-    public static final FlatMapConfig DELAY_ERRORS = new FlatMapConfig(true, Flowable.bufferSize());
+    public static final FlatMapConfig DELAY_ERRORS = new FlatMapConfig(true);
 
     /**
      * Optionally delay error, {@link Flowable#bufferSize()} sizes

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

@@ -0,0 +1,61 @@
+/*
+ * 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.Observable;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+/**
+ * Configuration record for Observable.groupBy() operators.
+ * @param delayError should the error propagation be delayed?
+ * @param bufferSize the expected number of items to buffer until it can be processed
+ * @since 4.0.0
+ */
+public record ObservableGroupByConfig(boolean delayError, int bufferSize) {
+
+    /**
+     * The default configuration with no error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableGroupByConfig DEFAULT = new ObservableGroupByConfig(false);
+
+    /**
+     * The default configuration with error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableGroupByConfig DELAY_ERROR = new ObservableGroupByConfig(true);
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     */
+    public ObservableGroupByConfig(boolean delayError) {
+        this(delayError, Observable.bufferSize());
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of items to buffer until it can be processed
+     */
+    public ObservableGroupByConfig(int bufferSize) {
+        this(false, bufferSize);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     * @param bufferSize the expected number of items to buffer until it can be processed
+     */
+    public ObservableGroupByConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+    }
+}

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

@@ -13,12 +13,11 @@
 
 package io.reactivex.rxjava4.core.config;
 
-import io.reactivex.rxjava4.core.Flowable;
-import io.reactivex.rxjava4.core.Observable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.internal.functions.ObjectHelper;
 
 /**
- * Configuration record for Observable.merge() operators.
+ * Configuration record for Observable.merge() and Observable.flatMap operators.
  * @param delayErrors should the error be delayed?
  * @param maxConcurrency the maximum number of concurrent flows?
  * @param bufferSize what would be the buffer size?
@@ -27,25 +26,25 @@
 public record ObservableMergeConfig(boolean delayErrors, int maxConcurrency, int bufferSize) {
 
     /**
-     * The default configuration with no error delays and bufferSize of {@link Observable#bufferSize()}.
+     * The default configuration with no error delays, MAX_VALUE concurrency and bufferSize of {@link Observable#bufferSize()}.
      */
-    public static final ObservableMergeConfig DEFAULT = new ObservableMergeConfig(false, Observable.bufferSize());
+    public static final ObservableMergeConfig DEFAULT = new ObservableMergeConfig(false);
 
     /**
-     * The default configuration with error delays and bufferSize of {@link Observable#bufferSize()}.
+     * The default configuration with error delays, MAX_VALUE concurrency and bufferSize of {@link Observable#bufferSize()}.
      */
-    public static final ObservableMergeConfig DELAY_ERROR = new ObservableMergeConfig(true, Observable.bufferSize());
+    public static final ObservableMergeConfig DELAY_ERROR = new ObservableMergeConfig(true);
 
     /**
-     * Optionally delay error, {@link Flowable#bufferSize()} sizes
+     * Optionally delay error, {@link Observable#bufferSize()} sizes
      * @param delayErrors should the error be delayed?
      */
     public ObservableMergeConfig(boolean delayErrors) {
-        this(delayErrors, Flowable.bufferSize(), Flowable.bufferSize());
+        this(delayErrors, Integer.MAX_VALUE, Observable.bufferSize());
     }
 
     /**
-     * Optionally set the buffer size, no delay errors.
+     * Optionally set the maximum concurrency levels, no errors and a buffer size of {@link Observable#bufferSize()}.
      * @param maxConcurrency the maximum number of concurrent flows
      */
     public ObservableMergeConfig(int maxConcurrency) {

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

@@ -0,0 +1,61 @@
+/*
+ * 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.Observable;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+/**
+ * Configuration record for Observable.observeOn() operators.
+ * @param delayError should the error propagation be delayed?
+ * @param bufferSize the expected number of items to cache from the upstream
+ * @since 4.0.0
+ */
+public record ObservableObserveOnConfig(boolean delayError, int bufferSize) {
+
+    /**
+     * The default configuration with no error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableObserveOnConfig DEFAULT = new ObservableObserveOnConfig(false);
+
+    /**
+     * The default configuration with error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableObserveOnConfig DELAY_ERROR = new ObservableObserveOnConfig(true);
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     */
+    public ObservableObserveOnConfig(boolean delayError) {
+        this(delayError, Observable.bufferSize());
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of items to cache from the upstream
+     */
+    public ObservableObserveOnConfig(int bufferSize) {
+        this(false, bufferSize);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     * @param bufferSize the expected number of items to cache from the upstream
+     */
+    public ObservableObserveOnConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+    }
+}

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

@@ -27,12 +27,12 @@ public record ObservableZipConfig(boolean delayError, int bufferSize) {
     /**
      * The default configuration with no error delays and bufferSize of Observable.bufferSize().
      */
-    public static final ObservableZipConfig DEFAULT = new ObservableZipConfig(false, Observable.bufferSize());
+    public static final ObservableZipConfig DEFAULT = new ObservableZipConfig(false);
 
     /**
      * The default configuration with error delays and bufferSize of Observable.bufferSize().
      */
-    public static final ObservableZipConfig DELAY_ERROR = new ObservableZipConfig(true, Observable.bufferSize());
+    public static final ObservableZipConfig DELAY_ERROR = new ObservableZipConfig(true);
 
     /**
      * Constructs a configuration record.

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

@@ -38,16 +38,19 @@ public final class ObservableFlatMapCompletableCompletable<T> extends Completabl
 
     final boolean delayErrors;
 
+    final int bufferSize;
+
     public ObservableFlatMapCompletableCompletable(ObservableSource<T> source,
-            Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
+            Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors, int bufferSize) {
         this.source = source;
         this.mapper = mapper;
         this.delayErrors = delayErrors;
+        this.bufferSize = bufferSize;
     }
 
     @Override
     protected void subscribeActual(CompletableObserver observer) {
-        source.subscribe(new FlatMapCompletableMainObserver<>(observer, mapper, delayErrors));
+        source.subscribe(new FlatMapCompletableMainObserver<>(observer, mapper, delayErrors, bufferSize));
     }
 
     @Override
@@ -67,16 +70,20 @@ static final class FlatMapCompletableMainObserver<T> extends AtomicInteger imple
 
         final boolean delayErrors;
 
+        final int bufferSize;
+
         final CompositeDisposable set;
 
         Disposable upstream;
 
         volatile boolean disposed;
 
-        FlatMapCompletableMainObserver(CompletableObserver observer, Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
+        FlatMapCompletableMainObserver(CompletableObserver observer, Function<? super T, ? extends CompletableSource> mapper,
+                boolean delayErrors, int bufferSize) {
             this.downstream = observer;
             this.mapper = mapper;
             this.delayErrors = delayErrors;
+            this.bufferSize = bufferSize;
             this.errors = new AtomicThrowable();
             this.set = new CompositeDisposable();
             this.lazySet(1);

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

@@ -36,16 +36,20 @@ public final class ObservableFlatMapMaybe<T, R> extends AbstractObservableWithUp
 
     final boolean delayErrors;
 
+    final int bufferSize;
+
     public ObservableFlatMapMaybe(ObservableSource<T> source, Function<? super T, ? extends MaybeSource<? extends R>> mapper,
-            boolean delayError) {
+            boolean delayError,
+            int bufferSize) {
         super(source);
         this.mapper = mapper;
         this.delayErrors = delayError;
+        this.bufferSize = bufferSize;
     }
 
     @Override
     protected void subscribeActual(Observer<? super R> observer) {
-        source.subscribe(new FlatMapMaybeObserver<>(observer, mapper, delayErrors));
+        source.subscribe(new FlatMapMaybeObserver<>(observer, mapper, delayErrors, bufferSize));
     }
 
     static final class FlatMapMaybeObserver<T, R>
@@ -59,6 +63,8 @@ static final class FlatMapMaybeObserver<T, R>
 
         final boolean delayErrors;
 
+        final int bufferSize;
+
         final CompositeDisposable set;
 
         final AtomicInteger active;
@@ -74,10 +80,11 @@ static final class FlatMapMaybeObserver<T, R>
         volatile boolean cancelled;
 
         FlatMapMaybeObserver(Observer<? super R> actual,
-                Function<? super T, ? extends MaybeSource<? extends R>> mapper, boolean delayErrors) {
+                Function<? super T, ? extends MaybeSource<? extends R>> mapper, boolean delayErrors, int bufferSize) {
             this.downstream = actual;
             this.mapper = mapper;
             this.delayErrors = delayErrors;
+            this.bufferSize = bufferSize;
             this.set = new CompositeDisposable();
             this.errors = new AtomicThrowable();
             this.active = new AtomicInteger(1);
@@ -178,7 +185,7 @@ SpscLinkedArrayQueue<R> getOrCreateQueue() {
             if (current != null) {
                 return current;
             }
-            current = new SpscLinkedArrayQueue<>(Observable.bufferSize());
+            current = new SpscLinkedArrayQueue<>(bufferSize);
             if (queue.compareAndSet(null, current)) {
                 return current;
             }

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

@@ -36,16 +36,21 @@ public final class ObservableFlatMapSingle<T, R> extends AbstractObservableWithU
 
     final boolean delayErrors;
 
-    public ObservableFlatMapSingle(ObservableSource<T> source, Function<? super T, ? extends SingleSource<? extends R>> mapper,
-            boolean delayError) {
+    final int bufferSize;
+
+    public ObservableFlatMapSingle(ObservableSource<T> source,
+            Function<? super T, ? extends SingleSource<? extends R>> mapper,
+            boolean delayError,
+            int bufferSize) {
         super(source);
         this.mapper = mapper;
         this.delayErrors = delayError;
+        this.bufferSize = bufferSize;
     }
 
     @Override
     protected void subscribeActual(Observer<? super R> observer) {
-        source.subscribe(new FlatMapSingleObserver<>(observer, mapper, delayErrors));
+        source.subscribe(new FlatMapSingleObserver<>(observer, mapper, delayErrors, bufferSize));
     }
 
     static final class FlatMapSingleObserver<T, R>
@@ -59,6 +64,8 @@ static final class FlatMapSingleObserver<T, R>
 
         final boolean delayErrors;
 
+        final int bufferSize;
+
         final CompositeDisposable set;
 
         final AtomicInteger active;
@@ -74,10 +81,12 @@ static final class FlatMapSingleObserver<T, R>
         volatile boolean cancelled;
 
         FlatMapSingleObserver(Observer<? super R> actual,
-                Function<? super T, ? extends SingleSource<? extends R>> mapper, boolean delayErrors) {
+                Function<? super T, ? extends SingleSource<? extends R>> mapper,
+                boolean delayErrors, int bufferSize) {
             this.downstream = actual;
             this.mapper = mapper;
             this.delayErrors = delayErrors;
+            this.bufferSize = bufferSize;
             this.set = new CompositeDisposable();
             this.errors = new AtomicThrowable();
             this.active = new AtomicInteger(1);
@@ -178,7 +187,7 @@ SpscLinkedArrayQueue<R> getOrCreateQueue() {
             if (current != null) {
                 return current;
             }
-            current = new SpscLinkedArrayQueue<>(Observable.bufferSize());
+            current = new SpscLinkedArrayQueue<>(bufferSize);
             if (queue.compareAndSet(null, current)) {
                 return current;
             }

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

@@ -0,0 +1,33 @@
+/*
+ * 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.RxJavaTest;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class ObservableGroupByConfigTest extends RxJavaTest {
+
+    @Test
+    public void validation() {
+        assertTrue(new ObservableGroupByConfig(true).delayError(), "delayError - true");
+        assertFalse(new ObservableGroupByConfig(false).delayError(), "delayError - false");
+        assertEquals(5, new ObservableGroupByConfig(5).bufferSize(), "bufferSize - 5");
+        assertEquals(5, new ObservableGroupByConfig(true, 5).bufferSize(), "bufferSize both - true, 5");
+        assertEquals(5, new ObservableGroupByConfig(false, 5).bufferSize(), "bufferSize both - false, 5");
+        assertTrue(new ObservableGroupByConfig(true, 5).delayError(), "delayError both - true, 5");
+        assertFalse(new ObservableGroupByConfig(false, 5).delayError(), "delayError both - false, 5");
+    }
+}