added variance to Func1 (hopefully) everywhere...

Author: jmhofer

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

@@ -338,7 +338,8 @@ class UnitTestSuite extends JUnitSuite {
     @Test def testMap {
         val numbers = Observable.from(1, 2, 3, 4, 5, 6, 7, 8, 9)
         val mappedNumbers = ArrayBuffer.empty[Int]
-        numbers.map((x: Int) => x * x).subscribe((squareVal: Int) => {
+        val mapped: Observable[Int] = numbers map ((x: Int) => x * x)
+        mapped.subscribe((squareVal: Int) => {
             mappedNumbers.append(squareVal)
         })
         assertEquals(List(1, 4, 9, 16, 25, 36, 49, 64, 81), mappedNumbers.toList)
@@ -559,7 +560,7 @@ class UnitTestSuite extends JUnitSuite {
 
     @Test def testFilterInForComprehension {
         val doubler = (i: Int) => Observable.from(i, i)
-        val filteredObservable = for {
+        val filteredObservable: Observable[Int] = for {
             i: Int <- Observable.from(1, 2, 3, 4)
             j: Int <- doubler(i) if isOdd(i)
         } yield j

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

@@ -108,7 +108,7 @@ public class Observable<T> {
 
     private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
 
-    private final Func1<Observer<T>, Subscription> onSubscribe;
+    private final Func1<? super Observer<T>, ? extends Subscription> onSubscribe;
 
     /**
      * Observable with Function to execute when subscribed to.
@@ -119,7 +119,7 @@ public class Observable<T> {
      * @param onSubscribe
      *            {@link Func1} to be executed when {@link #subscribe(Observer)} is called.
      */
-    protected Observable(Func1<Observer<T>, Subscription> onSubscribe) {
+    protected Observable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
         this.onSubscribe = onSubscribe;
     }
 
@@ -157,7 +157,7 @@ protected Observable() {
      */
     public Subscription subscribe(Observer<T> observer) {
         // allow the hook to intercept and/or decorate
-        Func1<Observer<T>, Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
+        Func1<? super Observer<T>, ? extends Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
         // validate and proceed
         if (observer == null) {
             throw new IllegalArgumentException("observer can not be null");
@@ -461,7 +461,7 @@ public Subscription call(Observer<T> observer) {
      * @return an Observable that, when an {@link Observer} subscribes to it, will execute the given
      *         function
      */
-    public static <T> Observable<T> create(Func1<Observer<T>, Subscription> func) {
+    public static <T> Observable<T> create(Func1<? super Observer<T>, ? extends Subscription> func) {
         return new Observable<T>(func);
     }
 
@@ -1096,7 +1096,7 @@ public Observable<List<T>> buffer(Func0<? extends Observable<BufferClosing>> buf
      * @return
      *         An {@link Observable} which produces buffers which are created and emitted when the specified {@link Observable}s publish certain objects.
      */
-    public Observable<List<T>> buffer(Observable<BufferOpening> bufferOpenings, Func1<BufferOpening, Observable<BufferClosing>> bufferClosingSelector) {
+    public Observable<List<T>> buffer(Observable<BufferOpening> bufferOpenings, Func1<? super BufferOpening, ? extends Observable<BufferClosing>> bufferClosingSelector) {
         return create(OperationBuffer.buffer(this, bufferOpenings, bufferClosingSelector));
     }
 
@@ -1325,7 +1325,7 @@ public static <R> Observable<R> zip(Collection<Observable<?>> ws, FuncN<R> reduc
      * @return an Observable that emits only those items in the original Observable that the filter
      *         evaluates as {@code true}
      */
-    public Observable<T> filter(Func1<T, Boolean> predicate) {
+    public Observable<T> filter(Func1<? super T, Boolean> predicate) {
         return create(OperationFilter.filter(this, predicate));
     }
 
@@ -1360,7 +1360,7 @@ public Observable<T> finallyDo(Action0 action) {
      *         obtained from this transformation.
      * @see #mapMany(Func1)
      */
-    public <R> Observable<R> flatMap(Func1<T, Observable<R>> func) {
+    public <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<R>> func) {
         return mapMany(func);
     }
 
@@ -1374,7 +1374,7 @@ public <R> Observable<R> flatMap(Func1<T, Observable<R>> func) {
      *         evaluates as {@code true}
      * @see #filter(Func1)
      */
-    public Observable<T> where(Func1<T, Boolean> predicate) {
+    public Observable<T> where(Func1<? super T, Boolean> predicate) {
         return filter(predicate);
     }
 
@@ -1389,7 +1389,7 @@ public Observable<T> where(Func1<T, Boolean> predicate) {
      * @return an Observable that emits the items from the source Observable, transformed by the
      *         given function
      */
-    public <R> Observable<R> map(Func1<T, R> func) {
+    public <R> Observable<R> map(Func1<? super T, ? extends R> func) {
         return create(OperationMap.map(this, func));
     }
 
@@ -1410,7 +1410,7 @@ public <R> Observable<R> map(Func1<T, R> func) {
      *         obtained from this transformation.
      * @see #flatMap(Func1)
      */
-    public <R> Observable<R> mapMany(Func1<T, Observable<R>> func) {
+    public <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<R>> func) {
         return create(OperationMap.mapMany(this, func));
     }
 
@@ -1496,7 +1496,7 @@ public <T2> Observable<T2> dematerialize() {
      *            encounters an error
      * @return the original Observable, with appropriately modified behavior
      */
-    public Observable<T> onErrorResumeNext(final Func1<Throwable, Observable<T>> resumeFunction) {
+    public Observable<T> onErrorResumeNext(final Func1<? super Throwable, ? extends Observable<T>> resumeFunction) {
         return create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(this, resumeFunction));
     }
 
@@ -1581,7 +1581,7 @@ public Observable<T> onExceptionResumeNext(final Observable<T> resumeSequence) {
      *            Observable encounters an error
      * @return the original Observable with appropriately modified behavior
      */
-    public Observable<T> onErrorReturn(Func1<Throwable, T> resumeFunction) {
+    public Observable<T> onErrorReturn(Func1<? super Throwable, ? extends T> resumeFunction) {
         return create(OperationOnErrorReturn.onErrorReturn(this, resumeFunction));
     }
 
@@ -1800,7 +1800,7 @@ public <R> Observable<R> scan(R initialValue, Func2<R, T, R> accumulator) {
      * @return an Observable that emits <code>true</code> if all items emitted by the source
      *         Observable satisfy the predicate; otherwise, <code>false</code>
      */
-    public Observable<Boolean> all(Func1<T, Boolean> predicate) {
+    public Observable<Boolean> all(Func1<? super T, Boolean> predicate) {
         return create(OperationAll.all(this, predicate));
     }
 
@@ -1853,7 +1853,7 @@ public Observable<T> take(final int num) {
      * @return an Observable that emits the items from the source Observable so long as each item
      *         satisfies the condition defined by <code>predicate</code>
      */
-    public Observable<T> takeWhile(final Func1<T, Boolean> predicate) {
+    public Observable<T> takeWhile(final Func1<? super T, Boolean> predicate) {
         return create(OperationTakeWhile.takeWhile(this, predicate));
     }
 
@@ -1992,7 +1992,7 @@ public Observable<T> startWith(T... values) {
      *         unique key value and emits items representing items from the source Observable that
      *         share that key value
      */
-    public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
+    public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {
         return create(OperationGroupBy.groupBy(this, keySelector, elementSelector));
     }
 
@@ -2010,7 +2010,7 @@ public <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<T, K> keyS
      *         unique key value and emits items representing items from the source Observable that
      *         share that key value
      */
-    public <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<T, K> keySelector) {
+    public <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<? super T, ? extends K> keySelector) {
         return create(OperationGroupBy.groupBy(this, keySelector));
     }
 

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -60,7 +60,7 @@
  */
 public class BlockingObservable<T> extends Observable<T> {
 
-    protected BlockingObservable(Func1<Observer<T>, Subscription> onSubscribe) {
+    protected BlockingObservable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
     }
 
@@ -126,7 +126,7 @@ public static <T> T last(final Observable<T> source) {
      * @return the last item emitted by the {@link Observable} for which the predicate function
      *         returns <code>true</code>
      */
-    public static <T> T last(final Observable<T> source, final Func1<T, Boolean> predicate) {
+    public static <T> T last(final Observable<T> source, final Func1<? super T, Boolean> predicate) {
         return last(source.filter(predicate));
     }
 
@@ -166,7 +166,7 @@ public static <T> T lastOrDefault(Observable<T> source, T defaultValue) {
      * @return the last item emitted by an {@link Observable} that matches the predicate, or the
      *         default value if no matching item is emitted
      */
-    public static <T> T lastOrDefault(Observable<T> source, T defaultValue, Func1<T, Boolean> predicate) {
+    public static <T> T lastOrDefault(Observable<T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
         return lastOrDefault(source.filter(predicate), defaultValue);
     }
 
@@ -257,7 +257,7 @@ public static <T> T single(Observable<T> source) {
      *             if the {@link Observable} does not emit exactly one item that matches the
      *             predicate
      */
-    public static <T> T single(Observable<T> source, Func1<T, Boolean> predicate) {
+    public static <T> T single(Observable<T> source, Func1<? super T, Boolean> predicate) {
         return from(source).single(predicate);
     }
 
@@ -293,7 +293,7 @@ public static <T> T singleOrDefault(Observable<T> source, T defaultValue) {
      * @return the single item emitted by the source {@link Observable} that matches the predicate,
      *         or a default value if no such value is emitted
      */
-    public static <T> T singleOrDefault(Observable<T> source, T defaultValue, Func1<T, Boolean> predicate) {
+    public static <T> T singleOrDefault(Observable<T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
         return from(source).singleOrDefault(defaultValue, predicate);
     }
 
@@ -443,7 +443,7 @@ public T last() {
      *            a predicate function to evaluate items emitted by the {@link Observable}
      * @return the last item emitted by the {@link Observable} that matches the predicate
      */
-    public T last(final Func1<T, Boolean> predicate) {
+    public T last(final Func1<? super T, Boolean> predicate) {
         return last(this, predicate);
     }
 
@@ -487,7 +487,7 @@ public T lastOrDefault(T defaultValue) {
      * @return the last item emitted by the {@link Observable} that matches the predicate, or the
      *         default value if no matching items are emitted
      */
-    public T lastOrDefault(T defaultValue, Func1<T, Boolean> predicate) {
+    public T lastOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
         return lastOrDefault(this, defaultValue, predicate);
     }
 
@@ -542,7 +542,7 @@ public T single() {
      *            a predicate function to evaluate items emitted by the {@link Observable}
      * @return the single item emitted by the source {@link Observable} that matches the predicate
      */
-    public T single(Func1<T, Boolean> predicate) {
+    public T single(Func1<? super T, Boolean> predicate) {
         return _singleOrDefault(from(this.filter(predicate)), false, null);
     }
 
@@ -575,7 +575,7 @@ public T singleOrDefault(T defaultValue) {
      * @return the single item emitted by the {@link Observable} that matches the predicate, or the
      *         default value if no such items are emitted
      */
-    public T singleOrDefault(T defaultValue, Func1<T, Boolean> predicate) {
+    public T singleOrDefault(T defaultValue, Func1<? super T, Boolean> predicate) {
         return _singleOrDefault(from(this.filter(predicate)), true, defaultValue);
     }
 

rxjava-core/src/main/java/rx/observables/ConnectableObservable.java

@@ -37,7 +37,7 @@
 
 public abstract class ConnectableObservable<T> extends Observable<T> {
 
-    protected ConnectableObservable(Func1<Observer<T>, Subscription> onSubscribe) {
+    protected ConnectableObservable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
     }
 

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -32,7 +32,7 @@
 public class GroupedObservable<K, T> extends Observable<T> {
     private final K key;
 
-    public GroupedObservable(K key, Func1<Observer<T>, Subscription> onSubscribe) {
+    public GroupedObservable(K key, Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
         this.key = key;
     }

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

@@ -34,18 +34,18 @@
  */
 public class OperationAll {
 
-    public static <T> Func1<Observer<Boolean>, Subscription> all(Observable<T> sequence, Func1<T, Boolean> predicate) {
+    public static <T> Func1<Observer<Boolean>, Subscription> all(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
         return new AllObservable<T>(sequence, predicate);
     }
 
     private static class AllObservable<T> implements Func1<Observer<Boolean>, Subscription> {
         private final Observable<T> sequence;
-        private final Func1<T, Boolean> predicate;
+        private final Func1<? super T, Boolean> predicate;
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
 
-        private AllObservable(Observable<T> sequence, Func1<T, Boolean> predicate) {
+        private AllObservable(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
             this.sequence = sequence;
             this.predicate = predicate;
         }

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

@@ -107,7 +107,7 @@ public Subscription call(final Observer<List<T>> observer) {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<List<T>>, Subscription> buffer(final Observable<T> source, final Observable<BufferOpening> bufferOpenings, final Func1<BufferOpening, Observable<BufferClosing>> bufferClosingSelector) {
+    public static <T> Func1<Observer<List<T>>, Subscription> buffer(final Observable<T> source, final Observable<BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<BufferClosing>> bufferClosingSelector) {
         return new Func1<Observer<List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<List<T>> observer) {
@@ -489,7 +489,7 @@ private static class ObservableBasedMultiBufferCreator<T> implements BufferCreat
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
-        public ObservableBasedMultiBufferCreator(final OverlappingBuffers<T> buffers, Observable<BufferOpening> bufferOpenings, final Func1<BufferOpening, Observable<BufferClosing>> bufferClosingSelector) {
+        public ObservableBasedMultiBufferCreator(final OverlappingBuffers<T> buffers, Observable<BufferOpening> bufferOpenings, final Func1<? super BufferOpening, ? extends Observable<BufferClosing>> bufferClosingSelector) {
             subscription.wrap(bufferOpenings.subscribe(new Action1<BufferOpening>() {
                 @Override
                 public void call(BufferOpening opening) {

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

@@ -33,16 +33,16 @@
  */
 public final class OperationFilter<T> {
 
-    public static <T> Func1<Observer<T>, Subscription> filter(Observable<T> that, Func1<T, Boolean> predicate) {
+    public static <T> Func1<Observer<T>, Subscription> filter(Observable<T> that, Func1<? super T, Boolean> predicate) {
         return new Filter<T>(that, predicate);
     }
 
     private static class Filter<T> implements Func1<Observer<T>, Subscription> {
 
         private final Observable<T> that;
-        private final Func1<T, Boolean> predicate;
+        private final Func1<? super T, Boolean> predicate;
 
-        public Filter(Observable<T> that, Func1<T, Boolean> predicate) {
+        public Filter(Observable<T> that, Func1<? super T, Boolean> predicate) {
             this.that = that;
             this.predicate = predicate;
         }

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

@@ -48,7 +48,7 @@
  */
 public final class OperationGroupBy {
 
-    public static <K, T, R> Func1<Observer<GroupedObservable<K, R>>, Subscription> groupBy(Observable<T> source, final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
+    public static <K, T, R> Func1<Observer<GroupedObservable<K, R>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {
 
         final Observable<KeyValue<K, R>> keyval = source.map(new Func1<T, KeyValue<K, R>>() {
             @Override
@@ -63,7 +63,7 @@ public KeyValue<K, R> call(T t) {
         return new GroupBy<K, R>(keyval);
     }
 
-    public static <K, T> Func1<Observer<GroupedObservable<K, T>>, Subscription> groupBy(Observable<T> source, final Func1<T, K> keySelector) {
+    public static <K, T> Func1<Observer<GroupedObservable<K, T>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector) {
         return groupBy(source, keySelector, Functions.<T> identity());
     }
 
@@ -204,7 +204,7 @@ public void unsubscribe() {
 
         private final AtomicReference<Observer<T>> subscribedObserver;
 
-        public GroupedSubject(K key, Func1<Observer<T>, Subscription> onSubscribe, AtomicReference<Observer<T>> subscribedObserver) {
+        public GroupedSubject(K key, Func1<? super Observer<T>, ? extends Subscription> onSubscribe, AtomicReference<Observer<T>> subscribedObserver) {
             super(key, onSubscribe);
             this.subscribedObserver = subscribedObserver;
         }