3.x: Fix Observable amb, combineLatest & zip ArrayStoreException (#6756)

Author: akarnokd

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmb.java

@@ -36,7 +36,7 @@ public void subscribeActual(Observer<? super T> observer) {
         ObservableSource<? extends T>[] sources = this.sources;
         int count = 0;
         if (sources == null) {
-            sources = new Observable[8];
+            sources = new ObservableSource[8];
             try {
                 for (ObservableSource<? extends T> p : sourcesIterable) {
                     if (p == null) {

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCombineLatest.java

@@ -48,7 +48,7 @@ public void subscribeActual(Observer<? super R> observer) {
         ObservableSource<? extends T>[] sources = this.sources;
         int count = 0;
         if (sources == null) {
-            sources = new Observable[8];
+            sources = new ObservableSource[8];
             for (ObservableSource<? extends T> p : sourcesIterable) {
                 if (count == sources.length) {
                     ObservableSource<? extends T>[] b = new ObservableSource[count + (count >> 2)];

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableZip.java

@@ -50,7 +50,7 @@ public void subscribeActual(Observer<? super R> observer) {
         ObservableSource<? extends T>[] sources = this.sources;
         int count = 0;
         if (sources == null) {
-            sources = new Observable[8];
+            sources = new ObservableSource[8];
             for (ObservableSource<? extends T> p : sourcesIterable) {
                 if (count == sources.length) {
                     ObservableSource<? extends T>[] b = new ObservableSource[count + (count >> 2)];

src/test/java/io/reactivex/rxjava3/internal/operators/completable/CompletableAmbTest.java

@@ -315,4 +315,18 @@ public void run() throws Exception {
             assertFalse("Interrupted!", interrupted.get());
         }
     }
+
+    @Test
+    public void completableSourcesInIterable() {
+        CompletableSource source = new CompletableSource() {
+            @Override
+            public void subscribe(CompletableObserver observer) {
+                Completable.complete().subscribe(observer);
+            }
+        };
+
+        Completable.amb(Arrays.asList(source, source))
+        .test()
+        .assertResult();
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableAmbTest.java

@@ -667,4 +667,19 @@ public void run() throws Exception {
             assertFalse("Interrupted!", interrupted.get());
         }
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void publishersInIterable() {
+        Publisher<Integer> source = new Publisher<Integer>() {
+            @Override
+            public void subscribe(Subscriber<? super Integer> subscriber) {
+                Flowable.just(1).subscribe(subscriber);
+            }
+        };
+
+        Flowable.amb(Arrays.asList(source, source))
+        .test()
+        .assertResult(1);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableCombineLatestTest.java

@@ -1551,4 +1551,24 @@ public Object apply(Object[] a) throws Exception {
         .awaitDone(5, TimeUnit.SECONDS)
         .assertFailure(TestException.class, 42);
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void publishersInIterable() {
+        Publisher<Integer> source = new Publisher<Integer>() {
+            @Override
+            public void subscribe(Subscriber<? super Integer> subscriber) {
+                Flowable.just(1).subscribe(subscriber);
+            }
+        };
+
+        Flowable.combineLatest(Arrays.asList(source, source), new Function<Object[], Integer>() {
+            @Override
+            public Integer apply(Object[] t) throws Throwable {
+                return 2;
+            }
+        })
+        .test()
+        .assertResult(2);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableZipTest.java

@@ -1896,4 +1896,24 @@ public Object apply(Object[] a) throws Exception {
 
         assertEquals(0, counter.get());
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void publishersInIterable() {
+        Publisher<Integer> source = new Publisher<Integer>() {
+            @Override
+            public void subscribe(Subscriber<? super Integer> subscriber) {
+                Flowable.just(1).subscribe(subscriber);
+            }
+        };
+
+        Flowable.zip(Arrays.asList(source, source), new Function<Object[], Integer>() {
+            @Override
+            public Integer apply(Object[] t) throws Throwable {
+                return 2;
+            }
+        })
+        .test()
+        .assertResult(2);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeAmbTest.java

@@ -254,4 +254,19 @@ public void run() {
             }
         }
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void maybeSourcesInIterable() {
+        MaybeSource<Integer> source = new MaybeSource<Integer>() {
+            @Override
+            public void subscribe(MaybeObserver<? super Integer> observer) {
+                Maybe.just(1).subscribe(observer);
+            }
+        };
+
+        Maybe.amb(Arrays.asList(source, source))
+        .test()
+        .assertResult(1);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/maybe/MaybeZipIterableTest.java

@@ -221,4 +221,24 @@ public void singleSourceZipperReturnsNull() {
         .to(TestHelper.<Object>testConsumer())
         .assertFailureAndMessage(NullPointerException.class, "The zipper returned a null value");
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void maybeSourcesInIterable() {
+        MaybeSource<Integer> source = new MaybeSource<Integer>() {
+            @Override
+            public void subscribe(MaybeObserver<? super Integer> observer) {
+                Maybe.just(1).subscribe(observer);
+            }
+        };
+
+        Maybe.zip(Arrays.asList(source, source), new Function<Object[], Integer>() {
+            @Override
+            public Integer apply(Object[] t) throws Throwable {
+                return 2;
+            }
+        })
+        .test()
+        .assertResult(2);
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmbTest.java

@@ -467,4 +467,18 @@ public void run() throws Exception {
         }
     }
 
+    @SuppressWarnings("unchecked")
+    @Test
+    public void observableSourcesInIterable() {
+        ObservableSource<Integer> source = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe(Observer<? super Integer> observer) {
+                Observable.just(1).subscribe(observer);
+            }
+        };
+
+        Observable.amb(Arrays.asList(source, source))
+        .test()
+        .assertResult(1);
+    }
 }