added variance to Func2, too

Author: jmhofer

rxjava-contrib/rxjava-android/src/main/java/rx/android/concurrency/HandlerThreadScheduler.java

@@ -1,11 +1,13 @@
 package rx.android.concurrency;
 
 import android.os.Handler;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
+
 import rx.Scheduler;
 import rx.Subscription;
 import rx.operators.SafeObservableSubscription;
@@ -39,7 +41,7 @@ public HandlerThreadScheduler(Handler handler) {
      * See {@link #schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)}
      */
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         return schedule(state, action, 0L, TimeUnit.MILLISECONDS);
     }
 
@@ -56,7 +58,7 @@ public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscr
      * @return A Subscription from which one can unsubscribe from.
      */
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delayTime, TimeUnit unit) {
         final SafeObservableSubscription subscription = new SafeObservableSubscription();
         final Scheduler _scheduler = this;
         handler.postDelayed(new Runnable() {
@@ -76,6 +78,7 @@ public static final class UnitTest {
         public void shouldScheduleImmediateActionOnHandlerThread() {
             final Handler handler = mock(Handler.class);
             final Object state = new Object();
+            @SuppressWarnings("unchecked")
             final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
 
             Scheduler scheduler = new HandlerThreadScheduler(handler);
@@ -94,6 +97,7 @@ public void shouldScheduleImmediateActionOnHandlerThread() {
         public void shouldScheduleDelayedActionOnHandlerThread() {
             final Handler handler = mock(Handler.class);
             final Object state = new Object();
+            @SuppressWarnings("unchecked")
             final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
 
             Scheduler scheduler = new HandlerThreadScheduler(handler);

rxjava-contrib/rxjava-swing/src/main/java/rx/concurrency/SwingScheduler.java

@@ -55,7 +55,7 @@ private SwingScheduler() {
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         final AtomicReference<Subscription> sub = new AtomicReference<Subscription>();
         EventQueue.invokeLater(new Runnable() {
             @Override
@@ -75,7 +75,7 @@ public void call() {
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long dueTime, TimeUnit unit) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long dueTime, TimeUnit unit) {
         final AtomicReference<Subscription> sub = new AtomicReference<Subscription>();
         long delay = unit.toMillis(dueTime);
         assertThatTheDelayIsValidForTheSwingTimer(delay);
@@ -113,7 +113,7 @@ public void call() {
     }
 
     @Override
-    public <T> Subscription schedulePeriodically(T state, final Func2<Scheduler, T, Subscription> action, long initialDelay, long period, TimeUnit unit) {
+    public <T> Subscription schedulePeriodically(T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long initialDelay, long period, TimeUnit unit) {
         final AtomicReference<Timer> timer = new AtomicReference<Timer>();
 
         final long delay = unit.toMillis(period);

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

@@ -918,7 +918,7 @@ public static <T> Observable<T> from(Future<T> future, long timeout, TimeUnit un
      *            Observable
      * @return an Observable that emits the zipped results
      */
-    public static <R, T0, T1> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Func2<T0, T1, R> reduceFunction) {
+    public static <R, T0, T1> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Func2<? super T0, ? super T1, ? extends R> reduceFunction) {
         return create(OperationZip.zip(w0, w1, reduceFunction));
     }
 
@@ -964,7 +964,7 @@ public Boolean call(T first, T second) {
      * @return an Observable that emits Booleans that indicate whether the corresponding items
      *         emitted by the source Observables are equal
      */
-    public static <T> Observable<Boolean> sequenceEqual(Observable<T> first, Observable<T> second, Func2<T, T, Boolean> equality) {
+    public static <T> Observable<Boolean> sequenceEqual(Observable<T> first, Observable<T> second, Func2<? super T, ? super T, Boolean> equality) {
         return zip(first, second, equality);
     }
 
@@ -1041,7 +1041,7 @@ public static <R, T0, T1, T2, T3> Observable<R> zip(Observable<T0> w0, Observabl
      *            The aggregation function used to combine the source observable values.
      * @return An Observable that combines the source Observables with the given combine function
      */
-    public static <R, T0, T1> Observable<R> combineLatest(Observable<T0> w0, Observable<T1> w1, Func2<T0, T1, R> combineFunction) {
+    public static <R, T0, T1> Observable<R> combineLatest(Observable<T0> w0, Observable<T1> w1, Func2<? super T0, ? super T1, ? extends R> combineFunction) {
         return create(OperationCombineLatest.combineLatest(w0, w1, combineFunction));
     }
 
@@ -1606,7 +1606,7 @@ public Observable<T> onErrorReturn(Func1<? super Throwable, ? extends T> resumeF
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229154(v%3Dvs.103).aspx">MSDN: Observable.Aggregate</a>
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
-    public Observable<T> reduce(Func2<T, T, T> accumulator) {
+    public Observable<T> reduce(Func2<? super T, ? super T, ? extends T> accumulator) {
         return create(OperationScan.scan(this, accumulator)).takeLast(1);
     }
 
@@ -1663,7 +1663,7 @@ public ConnectableObservable<T> publish() {
      * 
      * @see #reduce(Func2)
      */
-    public Observable<T> aggregate(Func2<T, T, T> accumulator) {
+    public Observable<T> aggregate(Func2<? super T, ? super T, ? extends T> accumulator) {
         return reduce(accumulator);
     }
 
@@ -1690,7 +1690,7 @@ public Observable<T> aggregate(Func2<T, T, T> accumulator) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229154(v%3Dvs.103).aspx">MSDN: Observable.Aggregate</a>
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
-    public <R> Observable<R> reduce(R initialValue, Func2<R, T, R> accumulator) {
+    public <R> Observable<R> reduce(R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
         return create(OperationScan.scan(this, initialValue, accumulator)).takeLast(1);
     }
 
@@ -1701,7 +1701,7 @@ public <R> Observable<R> reduce(R initialValue, Func2<R, T, R> accumulator) {
      * 
      * @see #reduce(Object, Func2)
      */
-    public <R> Observable<R> aggregate(R initialValue, Func2<R, T, R> accumulator) {
+    public <R> Observable<R> aggregate(R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
         return reduce(initialValue, accumulator);
     }
 
@@ -1724,7 +1724,7 @@ public <R> Observable<R> aggregate(R initialValue, Func2<R, T, R> accumulator) {
      * @return an Observable that emits the results of each call to the accumulator function
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
-    public Observable<T> scan(Func2<T, T, T> accumulator) {
+    public Observable<T> scan(Func2<? super T, ? super T, ? extends T> accumulator) {
         return create(OperationScan.scan(this, accumulator));
     }
 
@@ -1785,7 +1785,7 @@ public Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler) {
      * @return an Observable that emits the results of each call to the accumulator function
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
-    public <R> Observable<R> scan(R initialValue, Func2<R, T, R> accumulator) {
+    public <R> Observable<R> scan(R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
         return create(OperationScan.scan(this, initialValue, accumulator));
     }
 
@@ -1870,7 +1870,7 @@ public Observable<T> takeWhile(final Func1<? super T, Boolean> predicate) {
      * @return an Observable that emits items from the source Observable so long as the predicate
      *         continues to return <code>true</code> for each item, then completes
      */
-    public Observable<T> takeWhileWithIndex(final Func2<T, Integer, Boolean> predicate) {
+    public Observable<T> takeWhileWithIndex(final Func2<? super T, ? super Integer, Boolean> predicate) {
         return create(OperationTakeWhile.takeWhileWithIndex(this, predicate));
     }
 
@@ -1956,7 +1956,7 @@ public Observable<List<T>> toSortedList() {
      *            an Integer that indicates their sort order
      * @return an Observable that emits the items from the source Observable in sorted order
      */
-    public Observable<List<T>> toSortedList(Func2<T, T, Integer> sortFunction) {
+    public Observable<List<T>> toSortedList(Func2<? super T, ? super T, Integer> sortFunction) {
         return create(OperationToObservableSortedList.toSortedList(this, sortFunction));
     }
 

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

@@ -62,7 +62,7 @@ public abstract class Scheduler {
      *            Action to schedule.
      * @return a subscription to be able to unsubscribe from action.
      */
-    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);
+    public abstract <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action);
 
     /**
      * Schedules a cancelable action to be executed in delayTime.
@@ -77,7 +77,7 @@ public abstract class Scheduler {
      *            Time unit of the delay time.
      * @return a subscription to be able to unsubscribe from action.
      */
-    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);
+    public abstract <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delayTime, TimeUnit unit);
 
     /**
      * Schedules a cancelable action to be executed periodically.
@@ -96,7 +96,7 @@ public abstract class Scheduler {
      *            The time unit the interval above is given in.
      * @return A subscription to be able to unsubscribe from action.
      */
-    public <T> Subscription schedulePeriodically(T state, final Func2<Scheduler, T, Subscription> action, long initialDelay, long period, TimeUnit unit) {
+    public <T> Subscription schedulePeriodically(T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long initialDelay, long period, TimeUnit unit) {
         final long periodInNanos = unit.toNanos(period);
         final AtomicBoolean complete = new AtomicBoolean();
 
@@ -140,7 +140,7 @@ public void call() {
      *            Time the action is to be executed. If in the past it will be executed immediately.
      * @return a subscription to be able to unsubscribe from action.
      */
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, Date dueTime) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, Date dueTime) {
         long scheduledTime = dueTime.getTime();
         long timeInFuture = scheduledTime - now();
         if (timeInFuture <= 0) {
@@ -162,7 +162,7 @@ public Subscription schedule(final Action0 action) {
         return schedule(null, new Func2<Scheduler, Void, Subscription>() {
 
             @Override
-            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
+            public Subscription call(Scheduler scheduler, Void state) {
                 action.call();
                 return Subscriptions.empty();
             }
@@ -180,7 +180,7 @@ public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit
         return schedule(null, new Func2<Scheduler, Void, Subscription>() {
 
             @Override
-            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
+            public Subscription call(Scheduler scheduler, Void state) {
                 action.call();
                 return Subscriptions.empty();
             }
@@ -204,7 +204,7 @@ public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @Suppr
     public Subscription schedulePeriodically(final Action0 action, long initialDelay, long period, TimeUnit unit) {
         return schedulePeriodically(null, new Func2<Scheduler, Void, Subscription>() {
             @Override
-            public Subscription call(@SuppressWarnings("unused") Scheduler scheduler, @SuppressWarnings("unused") Void state) {
+            public Subscription call(Scheduler scheduler, Void state) {
                 action.call();
                 return Subscriptions.empty();
             }

rxjava-core/src/main/java/rx/concurrency/CurrentThreadScheduler.java

@@ -47,14 +47,14 @@ private CurrentThreadScheduler() {
     private final AtomicInteger counter = new AtomicInteger(0);
 
     @Override
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);
         enqueue(discardableAction, now());
         return discardableAction;
     }
 
     @Override
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long dueTime, TimeUnit unit) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long dueTime, TimeUnit unit) {
         long execTime = now() + unit.toMillis(dueTime);
 
         DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, new SleepingAction<T>(action, this, execTime));

rxjava-core/src/main/java/rx/concurrency/DiscardableAction.java

@@ -27,13 +27,13 @@
  * Combines standard {@link Subscription#unsubscribe()} functionality with ability to skip execution if an unsubscribe occurs before the {@link #call()} method is invoked.
  */
 /* package */class DiscardableAction<T> implements Func1<Scheduler, Subscription>, Subscription {
-    private final Func2<Scheduler, T, Subscription> underlying;
+    private final Func2<? super Scheduler, ? super T, ? extends Subscription> underlying;
     private final T state;
 
     private final SafeObservableSubscription wrapper = new SafeObservableSubscription();
     private final AtomicBoolean ready = new AtomicBoolean(true);
 
-    public DiscardableAction(T state, Func2<Scheduler, T, Subscription> underlying) {
+    public DiscardableAction(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> underlying) {
         this.state = state;
         this.underlying = underlying;
     }

rxjava-core/src/main/java/rx/concurrency/ExecutorScheduler.java

@@ -45,7 +45,7 @@ public ExecutorScheduler(ScheduledExecutorService executor) {
     }
 
     @Override
-    public <T> Subscription schedulePeriodically(final T state, final Func2<Scheduler, T, Subscription> action, long initialDelay, long period, TimeUnit unit) {
+    public <T> Subscription schedulePeriodically(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long initialDelay, long period, TimeUnit unit) {
         if (executor instanceof ScheduledExecutorService) {
             final CompositeSubscription subscriptions = new CompositeSubscription();
 
@@ -66,7 +66,7 @@ public void run() {
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delayTime, TimeUnit unit) {
         final DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);
         final Scheduler _scheduler = this;
         // all subscriptions that may need to be unsubscribed
@@ -113,7 +113,7 @@ public void run() {
     }
 
     @Override
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         final DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);
         final Scheduler _scheduler = this;
         // all subscriptions that may need to be unsubscribed

rxjava-core/src/main/java/rx/concurrency/ImmediateScheduler.java

@@ -41,12 +41,12 @@ private ImmediateScheduler() {
     }
 
     @Override
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         return action.call(this, state);
     }
 
     @Override
-    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long dueTime, TimeUnit unit) {
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long dueTime, TimeUnit unit) {
         // since we are executing immediately on this thread we must cause this thread to sleep
         long execTime = now() + unit.toMillis(dueTime);
 

rxjava-core/src/main/java/rx/concurrency/NewThreadScheduler.java

@@ -36,7 +36,7 @@ public static NewThreadScheduler getInstance() {
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
         final SafeObservableSubscription subscription = new SafeObservableSubscription();
         final Scheduler _scheduler = this;
 
@@ -53,7 +53,7 @@ public void run() {
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delay, TimeUnit unit) {
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delay, TimeUnit unit) {
         // we will use the system scheduler since it doesn't make sense to launch a new Thread and then sleep
         // we will instead schedule the event then launch the thread after the delay has passed
         final Scheduler _scheduler = this;

rxjava-core/src/main/java/rx/concurrency/SleepingAction.java

@@ -20,11 +20,11 @@
 import rx.util.functions.Func2;
 
 /* package */class SleepingAction<T> implements Func2<Scheduler, T, Subscription> {
-    private final Func2<Scheduler, T, Subscription> underlying;
+    private final Func2<? super Scheduler, ? super T, ? extends Subscription> underlying;
     private final Scheduler scheduler;
     private final long execTime;
 
-    public SleepingAction(Func2<Scheduler, T, Subscription> underlying, Scheduler scheduler, long execTime) {
+    public SleepingAction(Func2<? super Scheduler, ? super T, ? extends Subscription> underlying, Scheduler scheduler, long execTime) {
         this.underlying = underlying;
         this.scheduler = scheduler;
         this.execTime = execTime;