2.x: Zip, CombineLatest, and Amb operators throw when supplied with O…

…bservableSource implementation that doesn't subclass Observable (#6754) * 2.x: Zip, CombineLatest, and Amb operators throw when supplied with ObservableSource implementation that doesn't subclass Observable #6753 * 2.x: add tests for allowing arbitrary ObservableSource implementations
ReactiveX · Dec 11, 2019 · 52346a1 · 52346a1
1 parent 31b407f
commit 52346a1
Show file tree

Hide file tree

Showing 6 changed files with 78 additions and 3 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableAmb.java b/src/main/java/io/reactivex/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) {

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableCombineLatest.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableCombineLatest.java
@@ -49,7 +49,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)];

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableZip.java b/src/main/java/io/reactivex/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)];

diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableAmbTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableAmbTest.java
@@ -264,6 +264,23 @@ public void singleIterable() {
         .assertResult(1);
     }
 
+    /**
+     * Ensures that an ObservableSource implementation can be supplied that doesn't subclass Observable
+     */
+    @Test
+    public void singleIterableNotSubclassingObservable() {
+        final ObservableSource<Integer> s1 = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe (final Observer<? super Integer> observer) {
+                Observable.just(1).subscribe(observer);
+            }
+        };
+
+        Observable.amb(Collections.singletonList(s1))
+        .test()
+        .assertResult(1);
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void disposed() {

diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableCombineLatestTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableCombineLatestTest.java
@@ -787,6 +787,34 @@ public Object apply(Object[] a) throws Exception {
         .assertResult("[1, 2]");
     }
 
+    /**
+     * Ensures that an ObservableSource implementation can be supplied that doesn't subclass Observable
+     */
+    @Test
+    public void combineLatestIterableOfSourcesNotSubclassingObservable() {
+        final ObservableSource<Integer> s1 = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe (final Observer<? super Integer> observer) {
+                Observable.just(1).subscribe(observer);
+            }
+        };
+        final ObservableSource<Integer> s2 = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe (final Observer<? super Integer> observer) {
+                Observable.just(2).subscribe(observer);
+            }
+        };
+
+        Observable.combineLatest(Arrays.asList(s1, s2), new Function<Object[], Object>() {
+            @Override
+            public Object apply(Object[] a) throws Exception {
+                return Arrays.toString(a);
+            }
+        })
+        .test()
+        .assertResult("[1, 2]");
+    }
+
     @Test
     @SuppressWarnings("unchecked")
     public void combineLatestDelayErrorArrayOfSources() {
@@ -1216,4 +1244,5 @@ public Object apply(Object[] a) throws Exception {
         .awaitDone(5, TimeUnit.SECONDS)
         .assertFailure(TestException.class, 42);
     }
+
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableZipTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableZipTest.java
@@ -1350,6 +1350,34 @@ public Object apply(Object[] a) throws Exception {
         .assertResult("[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]");
     }
 
+    /**
+     * Ensures that an ObservableSource implementation can be supplied that doesn't subclass Observable
+     */
+    @Test
+    public void zipIterableNotSubclassingObservable() {
+        final ObservableSource<Integer> s1 = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe (final Observer<? super Integer> observer) {
+                Observable.just(1).subscribe(observer);
+            }
+        };
+        final ObservableSource<Integer> s2 = new ObservableSource<Integer>() {
+            @Override
+            public void subscribe (final Observer<? super Integer> observer) {
+                Observable.just(2).subscribe(observer);
+            }
+        };
+
+        Observable.zip(Arrays.asList(s1, s2), new Function<Object[], Object>() {
+            @Override
+            public Object apply(Object[] a) throws Exception {
+                return Arrays.toString(a);
+            }
+        })
+        .test()
+        .assertResult("[1, 2]");
+    }
+
     @Test
     public void dispose() {
         TestHelper.checkDisposed(Observable.zip(Observable.just(1), Observable.just(1), new BiFunction<Integer, Integer, Object>() {
@@ -1457,4 +1485,5 @@ public Object apply(Object[] a) throws Exception {
 
         assertEquals(0, counter.get());
     }
+
 }