added variance to Action*, too

Author: jmhofer

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

@@ -247,7 +247,7 @@ private Subscription protectivelyWrapAndSubscribe(Observer<T> o) {
         return subscription.wrap(subscribe(new SafeObserver<T>(subscription, o)));
     }
 
-    public Subscription subscribe(final Action1<T> onNext) {
+    public Subscription subscribe(final Action1<? super T> onNext) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -278,11 +278,11 @@ public void onNext(T args) {
         });
     }
 
-    public Subscription subscribe(final Action1<T> onNext, Scheduler scheduler) {
+    public Subscription subscribe(final Action1<? super T> onNext, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext);
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError) {
+    public Subscription subscribe(final Action1<? super T> onNext, final Action1<? super Throwable> onError) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -316,11 +316,11 @@ public void onNext(T args) {
         });
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, Scheduler scheduler) {
+    public Subscription subscribe(final Action1<? super T> onNext, final Action1<? super Throwable> onError, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError);
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
+    public Subscription subscribe(final Action1<? super T> onNext, final Action1<? super Throwable> onError, final Action0 onComplete) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -357,7 +357,7 @@ public void onNext(T args) {
         });
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, final Action0 onComplete, Scheduler scheduler) {
+    public Subscription subscribe(final Action1<? super T> onNext, final Action1<? super Throwable> onError, final Action0 onComplete, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError, onComplete);
     }
 

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

@@ -355,7 +355,7 @@ private Subscription protectivelyWrapAndSubscribe(Observer<T> o) {
      * @throws RuntimeException
      *             if an error occurs
      */
-    public void forEach(final Action1<T> onNext) {
+    public void forEach(final Action1<? super T> onNext) {
         final CountDownLatch latch = new CountDownLatch(1);
         final AtomicReference<Throwable> exceptionFromOnError = new AtomicReference<Throwable>();
 

rxjava-core/src/main/java/rx/util/functions/Functions.java

@@ -253,7 +253,7 @@ public Void call(Object... args) {
      * @param f
      * @return {@link FuncN}
      */
-    public static <T0> FuncN<Void> fromAction(final Action1<T0> f) {
+    public static <T0> FuncN<Void> fromAction(final Action1<? super T0> f) {
         return new FuncN<Void>() {
 
             @SuppressWarnings("unchecked")
@@ -275,7 +275,7 @@ public Void call(Object... args) {
      * @param f
      * @return {@link FuncN}
      */
-    public static <T0, T1> FuncN<Void> fromAction(final Action2<T0, T1> f) {
+    public static <T0, T1> FuncN<Void> fromAction(final Action2<? super T0, ? super T1> f) {
         return new FuncN<Void>() {
 
             @SuppressWarnings("unchecked")
@@ -297,7 +297,7 @@ public Void call(Object... args) {
      * @param f
      * @return {@link FuncN}
      */
-    public static <T0, T1, T2> FuncN<Void> fromAction(final Action3<T0, T1, T2> f) {
+    public static <T0, T1, T2> FuncN<Void> fromAction(final Action3<? super T0, ? super T1, ? super T2> f) {
         return new FuncN<Void>() {
 
             @SuppressWarnings("unchecked")