From 89f3218d84ef679c661b01822d110477f2b06ab1 Mon Sep 17 00:00:00 2001
From: Ben Christensen <benjchristensen@netflix.com>
Date: Tue, 10 Sep 2013 21:36:15 -0700
Subject: [PATCH 1/2] Remove covariance of scan/reduce since consumer/producer
 are the same <T, T, T>

See https://github.com/Netflix/RxJava/issues/360#issuecomment-24203016
---
 rxjava-core/src/main/java/rx/Observable.java  |  6 +--
 .../main/java/rx/operators/OperationScan.java |  2 +-
 .../src/test/java/rx/CovarianceTest.java      | 30 +++++++++++++++
 rxjava-core/src/test/java/rx/ReduceTests.java | 38 ++++++++++---------
 4 files changed, 55 insertions(+), 21 deletions(-)

diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 2ac0f42754..41d40e6217 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -3021,7 +3021,7 @@ public Observable<T> onErrorReturn(Func1<Throwable, ? extends T> resumeFunction)
      * @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<? super T, ? super T, ? extends T> accumulator) {
+    public Observable<T> reduce(Func2<T, T, T> accumulator) {
         return create(OperationScan.scan(this, accumulator)).takeLast(1);
     }
 
@@ -3166,7 +3166,7 @@ public ConnectableObservable<T> publish() {
      * 
      * @see #reduce(Func2)
      */
-    public Observable<T> aggregate(Func2<? super T, ? super T, ? extends T> accumulator) {
+    public Observable<T> aggregate(Func2<T, T, T> accumulator) {
         return reduce(accumulator);
     }
 
@@ -3227,7 +3227,7 @@ public <R> Observable<R> aggregate(R initialValue, Func2<? super R, ? super T, ?
      * @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<? super T, ? super T, ? extends T> accumulator) {
+    public Observable<T> scan(Func2<T, T, T> accumulator) {
         return create(OperationScan.scan(this, accumulator));
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationScan.java b/rxjava-core/src/main/java/rx/operators/OperationScan.java
index fae739dfd8..b71c804729 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationScan.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationScan.java
@@ -70,7 +70,7 @@ public static <T, R> OnSubscribeFunc<R> scan(Observable<? extends T> sequence, R
      * @return An observable sequence whose elements are the result of accumulating the output from the list of Observables.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v=vs.103).aspx">Observable.Scan(TSource) Method (IObservable(TSource), Func(TSource, TSource, TSource))</a>
      */
-    public static <T> OnSubscribeFunc<T> scan(Observable<? extends T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
+    public static <T> OnSubscribeFunc<T> scan(Observable<? extends T> sequence, Func2<T, T, T> accumulator) {
         return new AccuWithoutInitialValue<T>(sequence, accumulator);
     }
 
diff --git a/rxjava-core/src/test/java/rx/CovarianceTest.java b/rxjava-core/src/test/java/rx/CovarianceTest.java
index 69110b6c6a..01f2030c52 100644
--- a/rxjava-core/src/test/java/rx/CovarianceTest.java
+++ b/rxjava-core/src/test/java/rx/CovarianceTest.java
@@ -4,6 +4,8 @@
 
 import org.junit.Test;
 
+import rx.util.functions.Func2;
+
 /**
  * Test super/extends of generics.
  * 
@@ -21,6 +23,25 @@ public void testCovarianceOfFrom() {
         // Observable.<HorrorMovie>from(new Movie()); // may not compile
     }
 
+    @Test
+    public void testSortedList() {
+        Func2<Media, Media, Integer> SORT_FUNCTION = new Func2<Media, Media, Integer>() {
+
+            @Override
+            public Integer call(Media t1, Media t2) {
+                return 1;
+            }
+        };
+
+        // this one would work without the covariance generics
+        Observable<Media> o = Observable.from(new Movie(), new TVSeason(), new Album());
+        o.toSortedList(SORT_FUNCTION);
+
+        // this one would NOT work without the covariance generics
+        Observable<Movie> o2 = Observable.from(new Movie(), new ActionMovie(), new HorrorMovie());
+        o2.toSortedList(SORT_FUNCTION);
+    }
+
     /*
      * Most tests are moved into their applicable classes such as [Operator]Tests.java
      */
@@ -34,6 +55,15 @@ static class Movie extends Media {
     static class HorrorMovie extends Movie {
     }
 
+    static class ActionMovie extends Movie {
+    }
+
+    static class Album extends Media {
+    }
+
+    static class TVSeason extends Media {
+    }
+
     static class Rating {
     }
 
diff --git a/rxjava-core/src/test/java/rx/ReduceTests.java b/rxjava-core/src/test/java/rx/ReduceTests.java
index 1f7ea5550c..b812ba3638 100644
--- a/rxjava-core/src/test/java/rx/ReduceTests.java
+++ b/rxjava-core/src/test/java/rx/ReduceTests.java
@@ -25,6 +25,7 @@ public Integer call(Integer t1, Integer t2) {
         assertEquals(6, value);
     }
 
+    @SuppressWarnings("unused")
     @Test
     public void reduceWithObjects() {
         Observable<Movie> horrorMovies = Observable.<Movie> from(new HorrorMovie());
@@ -41,9 +42,15 @@ public Movie call(Movie t1, Movie t2) {
         Observable<Movie> reduceResult2 = horrorMovies.reduce(chooseSecondMovie);
     }
 
+    /**
+     * Reduce consumes and produces T so can't do covariance.
+     * 
+     * https://github.com/Netflix/RxJava/issues/360#issuecomment-24203016
+     */
+    @SuppressWarnings("unused")
     @Test
     public void reduceWithCovariantObjects() {
-        Observable<HorrorMovie> horrorMovies = Observable.from(new HorrorMovie());
+        Observable<Movie> horrorMovies = Observable.<Movie> from(new HorrorMovie());
 
         Func2<Movie, Movie, Movie> chooseSecondMovie =
                 new Func2<Movie, Movie, Movie>() {
@@ -52,24 +59,25 @@ public Movie call(Movie t1, Movie t2) {
                     }
                 };
 
-        Observable<Movie> reduceResult = Observable.create(OperationScan.scan(horrorMovies, chooseSecondMovie)).takeLast(1);
-
-        //TODO this isn't compiling
-        //        Observable<Movie> reduceResult2 = horrorMovies.reduce(chooseSecondMovie);
+        Observable<Movie> reduceResult2 = horrorMovies.reduce(chooseSecondMovie);
     }
 
+    /**
+     * Reduce consumes and produces T so can't do covariance.
+     * 
+     * https://github.com/Netflix/RxJava/issues/360#issuecomment-24203016
+     */
     @Test
     public void reduceCovariance() {
-        Observable<HorrorMovie> horrorMovies = Observable.from(new HorrorMovie());
-
-        // do something with horrorMovies, relying on the fact that all are HorrorMovies
-        // and not just any Movies...
-
-        // pass it to library (works because it takes Observable<? extends Movie>)
+        // must type it to <Movie>
+        Observable<Movie> horrorMovies = Observable.<Movie> from(new HorrorMovie());
         libraryFunctionActingOnMovieObservables(horrorMovies);
     }
 
-    public void libraryFunctionActingOnMovieObservables(Observable<? extends Movie> obs) {
+    /*
+     * This accepts <Movie> instead of <? super Movie> since `reduce` can't handle covariants
+     */
+    public void libraryFunctionActingOnMovieObservables(Observable<Movie> obs) {
         Func2<Movie, Movie, Movie> chooseSecondMovie =
                 new Func2<Movie, Movie, Movie>() {
                     public Movie call(Movie t1, Movie t2) {
@@ -77,11 +85,7 @@ public Movie call(Movie t1, Movie t2) {
                     }
                 };
 
-        Observable<Movie> reduceResult = Observable.create(OperationScan.scan(obs, chooseSecondMovie)).takeLast(1);
-
-        //TODO this isn't compiling
-        //        Observable<Movie> reduceResult2 = obs.reduce(chooseSecondMovie);
-        // do something with reduceResult...
+        obs.reduce(chooseSecondMovie);
     }
 
 }

From 3501fe74ad2588757e1b4d72ca321990a70610dd Mon Sep 17 00:00:00 2001
From: Ben Christensen <benjchristensen@netflix.com>
Date: Wed, 11 Sep 2013 09:52:16 -0700
Subject: [PATCH 2/2] Simplify reduce/scan generics

Updates based on discussion at https://github.com/Netflix/RxJava/pull/369#issuecomment-24255958
---
 rxjava-core/src/main/java/rx/Observable.java       |  6 +++---
 .../src/main/java/rx/operators/OperationScan.java  | 14 +++++++-------
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 41d40e6217..e1e1e59eea 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -3193,7 +3193,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<? super R, ? super T, ? extends R> accumulator) {
+    public <R> Observable<R> reduce(R initialValue, Func2<R, ? super T, R> accumulator) {
         return create(OperationScan.scan(this, initialValue, accumulator)).takeLast(1);
     }
 
@@ -3204,7 +3204,7 @@ public <R> Observable<R> reduce(R initialValue, Func2<? super R, ? super T, ? ex
      * 
      * @see #reduce(Object, Func2)
      */
-    public <R> Observable<R> aggregate(R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+    public <R> Observable<R> aggregate(R initialValue, Func2<R, ? super T, R> accumulator) {
         return reduce(initialValue, accumulator);
     }
 
@@ -3288,7 +3288,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<? super R, ? super T, ? extends R> accumulator) {
+    public <R> Observable<R> scan(R initialValue, Func2<R, ? super T, R> accumulator) {
         return create(OperationScan.scan(this, initialValue, accumulator));
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationScan.java b/rxjava-core/src/main/java/rx/operators/OperationScan.java
index b71c804729..513646b8ef 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationScan.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationScan.java
@@ -55,7 +55,7 @@ public final class OperationScan {
      * @return An observable sequence whose elements are the result of accumulating the output from the list of Observables.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212007%28v=vs.103%29.aspx">Observable.Scan(TSource, TAccumulate) Method (IObservable(TSource), TAccumulate, Func(TAccumulate, TSource, TAccumulate))</a>
      */
-    public static <T, R> OnSubscribeFunc<R> scan(Observable<? extends T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+    public static <T, R> OnSubscribeFunc<R> scan(Observable<? extends T> sequence, R initialValue, Func2<R, ? super T, R> accumulator) {
         return new Accumulator<T, R>(sequence, initialValue, accumulator);
     }
 
@@ -76,11 +76,11 @@ public static <T> OnSubscribeFunc<T> scan(Observable<? extends T> sequence, Func
 
     private static class AccuWithoutInitialValue<T> implements OnSubscribeFunc<T> {
         private final Observable<? extends T> sequence;
-        private final Func2<? super T, ? super T, ? extends T> accumulatorFunction;
+        private final Func2<T, T, T> accumulatorFunction;
         
         private AccumulatingObserver<T, T> accumulatingObserver;
         
-        private AccuWithoutInitialValue(Observable<? extends T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
+        private AccuWithoutInitialValue(Observable<? extends T> sequence, Func2<T, T, T> accumulator) {
             this.sequence = sequence;
             this.accumulatorFunction = accumulator;
         }
@@ -116,9 +116,9 @@ public void onCompleted() {
     private static class Accumulator<T, R> implements OnSubscribeFunc<R> {
         private final Observable<? extends T> sequence;
         private final R initialValue;
-        private final Func2<? super R, ? super T, ? extends R> accumulatorFunction;
+        private final Func2<R, ? super T, R> accumulatorFunction;
 
-        private Accumulator(Observable<? extends T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+        private Accumulator(Observable<? extends T> sequence, R initialValue, Func2<R, ? super T, R> accumulator) {
             this.sequence = sequence;
             this.initialValue = initialValue;
             this.accumulatorFunction = accumulator;
@@ -133,11 +133,11 @@ public Subscription onSubscribe(final Observer<? super R> observer) {
 
     private static class AccumulatingObserver<T, R> implements Observer<T> {
         private final Observer<? super R> observer;
-        private final Func2<? super R, ? super T, ? extends R> accumulatorFunction;
+        private final Func2<R, ? super T, R> accumulatorFunction;
 
         private R acc;
 
-        private AccumulatingObserver(Observer<? super R> observer, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+        private AccumulatingObserver(Observer<? super R> observer, R initialValue, Func2<R, ? super T, R> accumulator) {
             this.observer = observer;
             this.accumulatorFunction = accumulator;