generalized everything in Observable that deals with covariance of observables

Author: jmhofer

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/RxImplicits.scala

@@ -248,7 +248,7 @@ class UnitTestSuite extends JUnitSuite {
 
     @Test def testFlattenMerge {
         val observable = Observable.from(Observable.from(1, 2, 3))
-        val merged = Observable.merge(observable)
+        val merged = Observable.merge[Int](observable)
         assertSubscribeReceives(merged)(1, 2, 3)
     }
 

rxjava-core/src/main/java/rx/Observable.java

@@ -69,7 +69,7 @@ public final class OperationBuffer {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final Func0<? extends Observable<BufferClosing>> bufferClosingSelector) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final Func0<? extends Observable<? extends BufferClosing>> bufferClosingSelector) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -107,7 +107,7 @@ public Subscription call(final Observer<? super List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final Observable<BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<BufferClosing>> bufferClosingSelector) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final Observable<? extends BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<? extends BufferClosing>> bufferClosingSelector) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -135,7 +135,7 @@ public Subscription call(final Observer<? super List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<T> source, int count) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<? extends T> source, int count) {
         return buffer(source, count, count);
     }
 
@@ -162,7 +162,7 @@ public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observab
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final int count, final int skip) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final int count, final int skip) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -192,7 +192,7 @@ public Subscription call(final Observer<? super List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<T> source, long timespan, TimeUnit unit) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<? extends T> source, long timespan, TimeUnit unit) {
         return buffer(source, timespan, unit, Schedulers.threadPoolForComputation());
     }
 
@@ -217,7 +217,7 @@ public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observab
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -250,7 +250,7 @@ public Subscription call(final Observer<? super List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<T> source, long timespan, TimeUnit unit, int count) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<? extends T> source, long timespan, TimeUnit unit, int count) {
         return buffer(source, timespan, unit, count, Schedulers.threadPoolForComputation());
     }
 
@@ -278,7 +278,7 @@ public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observab
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -311,7 +311,7 @@ public Subscription call(final Observer<? super List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<T> source, long timespan, long timeshift, TimeUnit unit) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit) {
         return buffer(source, timespan, timeshift, unit, Schedulers.threadPoolForComputation());
     }
 
@@ -339,7 +339,7 @@ public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(Observab
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler) {
+    public static <T> Func1<Observer<? super List<T>>, Subscription> buffer(final Observable<? extends T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler) {
         return new Func1<Observer<? super List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<? super List<T>> observer) {
@@ -444,10 +444,10 @@ public void stop() {
     private static class ObservableBasedSingleBufferCreator<T> implements BufferCreator<T> {
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
-        private final Func0<? extends Observable<BufferClosing>> bufferClosingSelector;
+        private final Func0<? extends Observable<? extends BufferClosing>> bufferClosingSelector;
         private final NonOverlappingBuffers<T> buffers;
 
-        public ObservableBasedSingleBufferCreator(NonOverlappingBuffers<T> buffers, Func0<? extends Observable<BufferClosing>> bufferClosingSelector) {
+        public ObservableBasedSingleBufferCreator(NonOverlappingBuffers<T> buffers, Func0<? extends Observable<? extends BufferClosing>> bufferClosingSelector) {
             this.buffers = buffers;
             this.bufferClosingSelector = bufferClosingSelector;
 
@@ -456,7 +456,7 @@ public ObservableBasedSingleBufferCreator(NonOverlappingBuffers<T> buffers, Func
         }
 
         private void listenForBufferEnd() {
-            Observable<BufferClosing> closingObservable = bufferClosingSelector.call();
+            Observable<? extends BufferClosing> closingObservable = bufferClosingSelector.call();
             closingObservable.subscribe(new Action1<BufferClosing>() {
                 @Override
                 public void call(BufferClosing closing) {
@@ -489,12 +489,12 @@ private static class ObservableBasedMultiBufferCreator<T> implements BufferCreat
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
-        public ObservableBasedMultiBufferCreator(final OverlappingBuffers<T> buffers, Observable<BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<BufferClosing>> bufferClosingSelector) {
+        public ObservableBasedMultiBufferCreator(final OverlappingBuffers<T> buffers, Observable<? extends BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<? extends BufferClosing>> bufferClosingSelector) {
             subscription.wrap(bufferOpenings.subscribe(new Action1<BufferOpening>() {
                 @Override
                 public void call(BufferOpening opening) {
                     final Buffer<T> buffer = buffers.createBuffer();
-                    Observable<BufferClosing> closingObservable = bufferClosingSelector.call(opening);
+                    Observable<? extends BufferClosing> closingObservable = bufferClosingSelector.call(opening);
 
                     closingObservable.subscribe(new Action1<BufferClosing>() {
                         @Override

rxjava-core/src/main/java/rx/operators/OperationBuffer.java

@@ -38,6 +38,11 @@
 import rx.util.functions.Func2;
 import rx.util.functions.Func3;
 import rx.util.functions.Func4;
+import rx.util.functions.Func5;
+import rx.util.functions.Func6;
+import rx.util.functions.Func7;
+import rx.util.functions.Func8;
+import rx.util.functions.Func9;
 import rx.util.functions.FuncN;
 import rx.util.functions.Functions;
 
@@ -62,7 +67,7 @@ public class OperationCombineLatest {
      *          The aggregation function used to combine the source observable values.
      * @return A function from an observer to a subscription. This can be used to create an observable from.
      */
-    public static <T0, T1, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<T0> w0, Observable<T1> w1, Func2<? super T0, ? super T1, ? extends R> combineLatestFunction) {
+    public static <T0, T1, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<T1> w1, Func2<? super T0, ? super T1, ? extends R> combineLatestFunction) {
         Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
         a.addObserver(new CombineObserver<R, T0>(a, w0));
         a.addObserver(new CombineObserver<R, T1>(a, w1));
@@ -72,7 +77,8 @@ public static <T0, T1, R> Func1<Observer<? super R>, Subscription> combineLatest
     /**
      * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
      */
-    public static <T0, T1, T2, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Func3<? super T0, ? super T1, ? super T2, ? extends R> combineLatestFunction) {
+    public static <T0, T1, T2, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, 
+            Func3<? super T0, ? super T1, ? super T2, ? extends R> combineLatestFunction) {
         Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
         a.addObserver(new CombineObserver<R, T0>(a, w0));
         a.addObserver(new CombineObserver<R, T1>(a, w1));
@@ -83,7 +89,8 @@ public static <T0, T1, T2, R> Func1<Observer<? super R>, Subscription> combineLa
     /**
      * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
      */
-    public static <T0, T1, T2, T3, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Observable<T3> w3, Func4<? super T0, ? super T1, ? super T2, ? super T3, ? extends R> combineLatestFunction) {
+    public static <T0, T1, T2, T3, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, 
+            Func4<? super T0, ? super T1, ? super T2, ? super T3, ? extends R> combineLatestFunction) {
         Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
         a.addObserver(new CombineObserver<R, T0>(a, w0));
         a.addObserver(new CombineObserver<R, T1>(a, w1));
@@ -92,12 +99,92 @@ public static <T0, T1, T2, T3, R> Func1<Observer<? super R>, Subscription> combi
         return a;
     }
 
+    /**
+     * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
+     */
+    public static <T0, T1, T2, T3, T4, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, Observable<? extends T4> w4, 
+            Func5<? super T0, ? super T1, ? super T2, ? super T3, ? super T4, ? extends R> combineLatestFunction) {
+        Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
+        a.addObserver(new CombineObserver<R, T0>(a, w0));
+        a.addObserver(new CombineObserver<R, T1>(a, w1));
+        a.addObserver(new CombineObserver<R, T2>(a, w2));
+        a.addObserver(new CombineObserver<R, T3>(a, w3));
+        a.addObserver(new CombineObserver<R, T4>(a, w4));
+        return a;
+    }
+
+    /**
+     * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
+     */
+    public static <T0, T1, T2, T3, T4, T5, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, Observable<? extends T4> w4, Observable<? extends T5> w5, 
+            Func6<? super T0, ? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> combineLatestFunction) {
+        Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
+        a.addObserver(new CombineObserver<R, T0>(a, w0));
+        a.addObserver(new CombineObserver<R, T1>(a, w1));
+        a.addObserver(new CombineObserver<R, T2>(a, w2));
+        a.addObserver(new CombineObserver<R, T3>(a, w3));
+        a.addObserver(new CombineObserver<R, T4>(a, w4));
+        a.addObserver(new CombineObserver<R, T5>(a, w5));
+        return a;
+    }
+
+    /**
+     * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
+     */
+    public static <T0, T1, T2, T3, T4, T5, T6, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, Observable<? extends T4> w4, Observable<? extends T5> w5, Observable<? extends T6> w6, 
+            Func7<? super T0, ? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> combineLatestFunction) {
+        Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
+        a.addObserver(new CombineObserver<R, T0>(a, w0));
+        a.addObserver(new CombineObserver<R, T1>(a, w1));
+        a.addObserver(new CombineObserver<R, T2>(a, w2));
+        a.addObserver(new CombineObserver<R, T3>(a, w3));
+        a.addObserver(new CombineObserver<R, T4>(a, w4));
+        a.addObserver(new CombineObserver<R, T5>(a, w5));
+        a.addObserver(new CombineObserver<R, T6>(a, w6));
+        return a;
+    }
+
+    /**
+     * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
+     */
+    public static <T0, T1, T2, T3, T4, T5, T6, T7, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, Observable<? extends T4> w4, Observable<? extends T5> w5, Observable<? extends T6> w6, Observable<? extends T7> w7, 
+            Func8<? super T0, ? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> combineLatestFunction) {
+        Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
+        a.addObserver(new CombineObserver<R, T0>(a, w0));
+        a.addObserver(new CombineObserver<R, T1>(a, w1));
+        a.addObserver(new CombineObserver<R, T2>(a, w2));
+        a.addObserver(new CombineObserver<R, T3>(a, w3));
+        a.addObserver(new CombineObserver<R, T4>(a, w4));
+        a.addObserver(new CombineObserver<R, T5>(a, w5));
+        a.addObserver(new CombineObserver<R, T6>(a, w6));
+        a.addObserver(new CombineObserver<R, T7>(a, w7));
+        return a;
+    }
+
+    /**
+     * @see #combineLatest(Observable w0, Observable w1, Func2 combineLatestFunction)
+     */
+    public static <T0, T1, T2, T3, T4, T5, T6, T7, T8, R> Func1<Observer<? super R>, Subscription> combineLatest(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Observable<? extends T3> w3, Observable<? extends T4> w4, Observable<? extends T5> w5, Observable<? extends T6> w6, Observable<? extends T7> w7, Observable<? extends T8> w8, 
+            Func9<? super T0, ? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> combineLatestFunction) {
+        Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
+        a.addObserver(new CombineObserver<R, T0>(a, w0));
+        a.addObserver(new CombineObserver<R, T1>(a, w1));
+        a.addObserver(new CombineObserver<R, T2>(a, w2));
+        a.addObserver(new CombineObserver<R, T3>(a, w3));
+        a.addObserver(new CombineObserver<R, T4>(a, w4));
+        a.addObserver(new CombineObserver<R, T5>(a, w5));
+        a.addObserver(new CombineObserver<R, T6>(a, w6));
+        a.addObserver(new CombineObserver<R, T7>(a, w7));
+        a.addObserver(new CombineObserver<R, T8>(a, w8));
+        return a;
+    }
+
     private static class CombineObserver<R, T> implements Observer<T> {
-        final Observable<T> w;
+        final Observable<? extends T> w;
         final Aggregator<R> a;
         private Subscription subscription;
 
-        public CombineObserver(Aggregator<R> a, Observable<T> w) {
+        public CombineObserver(Aggregator<R> a, Observable<? extends T> w) {
             this.a = a;
             this.w = w;
         }