4.x: API reductions: sequenceEqual, switchOnNext, zip (#8190)

* 4.x: API reductions: sequenceEqual, switchOnNext, zip

* Fix copy-paste error prefetch -> bufferSize param naming

Author: akarnokd

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

@@ -13,7 +13,7 @@
 
 package io.reactivex.rxjava4.core.config;
 
-import io.reactivex.rxjava4.core.Flowable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.internal.functions.ObjectHelper;
 
 /**
@@ -39,7 +39,7 @@ public record CompletableMergeConfig(boolean delayErrors, int maxConcurrency) {
      * @param delayErrors should the error propagation be delayed?
      */
     public CompletableMergeConfig(boolean delayErrors) {
-        this(delayErrors, 2);
+        this(delayErrors, Flowable.bufferSize());
     }
 
     /**

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

@@ -13,7 +13,7 @@
 
 package io.reactivex.rxjava4.core.config;
 
-import io.reactivex.rxjava4.core.Flowable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.internal.functions.ObjectHelper;
 
 /**
@@ -40,7 +40,7 @@ public record MaybeConcatEagerConfig(boolean delayError, int maxConcurrency, int
      * @param delayError should the error propagation be delayed?
      */
     public MaybeConcatEagerConfig(boolean delayError) {
-        this(delayError, 2, 2);
+        this(delayError, Flowable.bufferSize(), Flowable.bufferSize());
     }
 
     /**

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

@@ -25,12 +25,12 @@
 public record ObservableCombineLatestConfig(boolean delayError, int bufferSize) {
 
     /**
-     * The default configuration with no error delays and prefetch of 2.
+     * The default configuration with no error delays and bufferSize of Observable.bufferSize().
      */
     public static final ObservableCombineLatestConfig DEFAULT = new ObservableCombineLatestConfig(false, Observable.bufferSize());
 
     /**
-     * The default configuration with error delays and prefetch of 2.
+     * The default configuration with error delays and  bufferSize of Observable.bufferSize().
      */
     public static final ObservableCombineLatestConfig DELAY_ERROR = new ObservableCombineLatestConfig(true, Observable.bufferSize());
 
@@ -56,6 +56,6 @@ public ObservableCombineLatestConfig(int bufferSize) {
      * @param bufferSize the expected number of row combination items to be buffered internally
      */
     public ObservableCombineLatestConfig {
-        ObjectHelper.verifyPositive(bufferSize, "prefetch");
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
     }
 }

src/main/java/io/reactivex/rxjava4/core/config/ObservableCombineLatestConfig.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 java.util.Objects;
+
+import io.reactivex.rxjava4.core.Observable;
+import io.reactivex.rxjava4.functions.BiPredicate;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+/**
+ * Configuration record for Observable.sequenceEqual() operators.
+ * @param <T> the element type of the sequences being compared
+ * @param bufferSize the expected number of items to cache from the inner {@code ObservableSource}s
+ * @param isEqual the custom lambda to compare two elements
+ * @since 4.0.0
+ */
+public record ObservableSequenceEqualConfig<T>(int bufferSize, BiPredicate<? super T, ? super T> isEqual) {
+
+    /**
+     * The default configuration with bufferSize of Observable.bufferSize() and a default Objects.equals predicate.
+     */
+    public static final ObservableSequenceEqualConfig<Object> DEFAULT =
+            new ObservableSequenceEqualConfig<>(Observable.bufferSize(), ObjectHelper.equalsPredicate());
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of row combination items to be buffered internally
+     */
+    public ObservableSequenceEqualConfig(int bufferSize) {
+        this(bufferSize, ObjectHelper.equalsPredicate());
+    }
+
+    /**
+     * Constructs a configuration record.
+ * @param isEqual the custom lambda to compare two elements
+     */
+    public ObservableSequenceEqualConfig(BiPredicate<? super T, ? super T> isEqual) {
+        this(Observable.bufferSize(), isEqual);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of items to cache from the inner {@code ObservableSource}s
+     * @param isEqual the custom lambda to compare two elements
+     */
+    public ObservableSequenceEqualConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+        Objects.requireNonNull(isEqual, "isEqual is null");
+    }
+}

src/main/java/io/reactivex/rxjava4/core/config/ObservableSequenceEqualConfig.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.switchOnNext() operators.
+ * @param delayError should the error propagation be delayed?
+ * @param bufferSize the expected number of items to cache from the inner {@code ObservableSource}s
+ * @since 4.0.0
+ */
+public record ObservableSwitchConfig(boolean delayError, int bufferSize) {
+
+    /**
+     * The default configuration with no error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableSwitchConfig DEFAULT = new ObservableSwitchConfig(false, Observable.bufferSize());
+
+    /**
+     * The default configuration with error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableSwitchConfig DELAY_ERROR = new ObservableSwitchConfig(true, Observable.bufferSize());
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     */
+    public ObservableSwitchConfig(boolean delayError) {
+        this(delayError, Observable.bufferSize());
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of row combination items to be buffered internally
+     */
+    public ObservableSwitchConfig(int bufferSize) {
+        this(false, bufferSize);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     * @param bufferSize the expected number of row combination items to be buffered internally
+     */
+    public ObservableSwitchConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+    }
+}

src/main/java/io/reactivex/rxjava4/core/config/ObservableSwitchConfig.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.zip() operators.
+ * @param delayError should the error propagation be delayed?
+ * @param bufferSize the expected number of items to cache from the inner {@code ObservableSource}s
+ * @since 4.0.0
+ */
+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());
+
+    /**
+     * The default configuration with error delays and bufferSize of Observable.bufferSize().
+     */
+    public static final ObservableZipConfig DELAY_ERROR = new ObservableZipConfig(true, Observable.bufferSize());
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     */
+    public ObservableZipConfig(boolean delayError) {
+        this(delayError, Observable.bufferSize());
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param bufferSize the expected number of row combination items to be buffered internally
+     */
+    public ObservableZipConfig(int bufferSize) {
+        this(false, bufferSize);
+    }
+
+    /**
+     * Constructs a configuration record.
+     * @param delayError should the error propagation be delayed?
+     * @param bufferSize the expected number of row combination items to be buffered internally
+     */
+    public ObservableZipConfig {
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+    }
+}

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

@@ -13,7 +13,7 @@
 
 package io.reactivex.rxjava4.core.config;
 
-import io.reactivex.rxjava4.core.Flowable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.internal.functions.ObjectHelper;
 
 /**
@@ -40,7 +40,7 @@ public record SingleConcatEagerConfig(boolean delayError, int maxConcurrency, in
      * @param delayError should the error propagation be delayed?
      */
     public SingleConcatEagerConfig(boolean delayError) {
-        this(delayError, 2, 2);
+        this(delayError, Flowable.bufferSize(), Flowable.bufferSize());
     }
 
     /**

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

@@ -22,12 +22,12 @@ public class ObservableCombineLatestConfigTest extends RxJavaTest {
 
     @Test
     public void validation() {
-        assertTrue(new ObservableCombineLatestConfig(true).delayError(), "delayError - true");
-        assertFalse(new ObservableCombineLatestConfig(false).delayError(), "delayError - false");
-        assertEquals(5, new ObservableCombineLatestConfig(5).bufferSize(), "bufferSize - 5");
-        assertEquals(5, new ObservableCombineLatestConfig(true, 5).bufferSize(), "bufferSize both - true, 5");
-        assertEquals(5, new ObservableCombineLatestConfig(false, 5).bufferSize(), "bufferSize both - false, 5");
-        assertTrue(new ObservableCombineLatestConfig(true, 5).delayError(), "delayError both - true, 5");
-        assertFalse(new ObservableCombineLatestConfig(false, 5).delayError(), "delayError both - false, 5");
+        assertTrue(new ObservableSwitchConfig(true).delayError(), "delayError - true");
+        assertFalse(new ObservableSwitchConfig(false).delayError(), "delayError - false");
+        assertEquals(5, new ObservableSwitchConfig(5).bufferSize(), "bufferSize - 5");
+        assertEquals(5, new ObservableSwitchConfig(true, 5).bufferSize(), "bufferSize both - true, 5");
+        assertEquals(5, new ObservableSwitchConfig(false, 5).bufferSize(), "bufferSize both - false, 5");
+        assertTrue(new ObservableSwitchConfig(true, 5).delayError(), "delayError both - true, 5");
+        assertFalse(new ObservableSwitchConfig(false, 5).delayError(), "delayError both - false, 5");
     }
 }

src/test/java/io/reactivex/rxjava4/core/config/ObservableCombineLatestConfigTest.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 static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Objects;
+
+import org.junit.jupiter.api.Test;
+
+import io.reactivex.rxjava4.core.RxJavaTest;
+import io.reactivex.rxjava4.functions.BiPredicate;
+import io.reactivex.rxjava4.internal.functions.ObjectHelper;
+
+public class ObservableSequenceEqualConfigTest extends RxJavaTest {
+
+    @Test
+    public void validation() {
+        BiPredicate<Object, Object> predicate = Objects::equals;
+
+        assertEquals(5, new ObservableSequenceEqualConfig<>(5).bufferSize(), "bufferSize - 5");
+        assertEquals(ObjectHelper.equalsPredicate(), new ObservableSequenceEqualConfig<>(5).isEqual(), "isEqual - std");
+        assertEquals(5, new ObservableSequenceEqualConfig<>(5, predicate).bufferSize(), "bufferSize - 5");
+        assertEquals(predicate, new ObservableSequenceEqualConfig<>(5, predicate).isEqual(), "isEqual - custom");
+    }
+}