Remove newly added aggregate methods

- avoiding methods with `index`
- resultSelector overload is same as `reduce(seed, accumulator).map(resultSelector)`

Author: benjchristensen

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

@@ -5327,7 +5327,7 @@ public Observable<T> aggregate(Func2<T, T, T> accumulator) {
     public <R> Observable<R> reduce(R initialValue, Func2<R, ? super T, R> accumulator) {
         return create(OperationScan.scan(this, initialValue, accumulator)).takeLast(1);
     }
-
+    
     /**
      * Synonymous with <code>reduce()</code>.
      * <p>
@@ -5341,49 +5341,6 @@ public <R> Observable<R> aggregate(R initialValue, Func2<R, ? super T, R> accumu
         return reduce(initialValue, accumulator);
     }
 
-    /**
-     * Create an Observable that aggregates the source values with the given accumulator
-     * function and projects the final result via the resultselector.
-     * <p>
-     * Works like the {@link #aggregate(java.lang.Object, rx.util.functions.Func2)} projected
-     * with {@link #map(rx.util.functions.Func1)} without the overhead of some helper
-     * operators.
-     * @param <U> the intermediate (accumulator) type
-     * @param <V> the result type
-     * @param seed the initial value of the accumulator
-     * @param accumulator the function that takes the current accumulator value,
-     *                    the current emitted value and returns a (new) accumulated value.
-     * @param resultSelector the selector to project the final value of the accumulator
-     * @return an Observable that aggregates the source values with the given accumulator
-     *         function and projects the final result via the resultselector
-     */
-    public <U, V> Observable<V> aggregate(
-            U seed, Func2<U, ? super T, U> accumulator, 
-            Func1<? super U, ? extends V> resultSelector) {
-        return create(new OperationAggregate.AggregateSelector<T, U, V>(this, seed, accumulator, resultSelector));
-    }
-    
-    /**
-     * Create an Observable that aggregates the source values with the given indexed accumulator
-     * function and projects the final result via the indexed resultselector.
-     * 
-     * @param <U> the intermediate (accumulator) type
-     * @param <V> the result type
-     * @param seed the initial value of the accumulator
-     * @param accumulator the function that takes the current accumulator value,
-     *                    the current emitted value and returns a (new) accumulated value.
-     * @param resultSelector the selector to project the final value of the accumulator, where
-     *                       the second argument is the total number of elements accumulated
-     * @return an Observable that aggregates the source values with the given indexed accumulator
-     * function and projects the final result via the indexed resultselector.
-     */
-    public <U, V> Observable<V> aggregateIndexed(
-            U seed, Func3<U, ? super T, ? super Integer, U> accumulator,
-        Func2<? super U, ? super Integer, ? extends V> resultSelector
-    ) {
-        return create(new OperationAggregate.AggregateIndexedSelector<T, U, V>(this, seed, accumulator, resultSelector));
-    }
-
     /**
      * Returns an Observable that applies a function of your choosing to the
      * first item emitted by a source Observable, then feeds the result of that

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

@@ -45,41 +45,13 @@ public Integer call(Integer t1, Integer t2) {
     @Test
     public void testAggregateAsIntSum() {
 
-        Observable<Integer> result = Observable.from(1, 2, 3, 4, 5).aggregate(0, sum, Functions.<Integer>identity());
+        Observable<Integer> result = Observable.from(1, 2, 3, 4, 5).reduce(0, sum).map(Functions.<Integer>identity());
 
         result.subscribe(observer);
 
         verify(observer).onNext(1 + 2 + 3 + 4 + 5);
         verify(observer).onCompleted();
         verify(observer, never()).onError(any(Throwable.class));
-        
-    }
-    
-    @Test
-    public void testAggregateIndexedAsAverage() {
-        Func3<Integer, Integer, Integer, Integer> sumIndex = new Func3<Integer, Integer, Integer, Integer>() {
-            @Override
-            public Integer call(Integer acc, Integer value, Integer index) {
-                return acc + (index + 1) + value;
-            }
-        };
-        Func2<Integer, Integer, Integer> selectIndex = new Func2<Integer, Integer, Integer>() {
-            @Override
-            public Integer call(Integer t1, Integer count) {
-                return t1 + count;
-            }
-            
-        };
-        
-        Observable<Integer> result = Observable.from(1, 2, 3, 4, 5)
-            .aggregateIndexed(0, sumIndex, selectIndex);
-        
-        result.subscribe(observer);
-        
-        verify(observer).onNext(2 + 4 + 6 + 8 + 10 + 5);
-        verify(observer).onCompleted();
-        verify(observer, never()).onError(any(Throwable.class));
-        
     }
 
     static class CustomException extends RuntimeException { }
@@ -88,7 +60,7 @@ static class CustomException extends RuntimeException { }
     public void testAggregateAsIntSumSourceThrows() {
         Observable<Integer> result = Observable.concat(Observable.from(1, 2, 3, 4, 5),
             Observable.<Integer>error(new CustomException()))
-            .aggregate(0, sum, Functions.<Integer>identity());
+            .reduce(0, sum).map(Functions.<Integer>identity());
 
         result.subscribe(observer);
 
@@ -107,7 +79,7 @@ public Integer call(Integer t1, Integer t2) {
         };
 
         Observable<Integer> result = Observable.from(1, 2, 3, 4, 5)
-            .aggregate(0, sumErr, Functions.<Integer>identity());
+            .reduce(0, sumErr).map(Functions.<Integer>identity());
 
         result.subscribe(observer);
 
@@ -128,7 +100,7 @@ public Integer call(Integer t1) {
         };
 
         Observable<Integer> result = Observable.from(1, 2, 3, 4, 5)
-            .aggregate(0, sum, error);
+            .reduce(0, sum).map(error);
 
         result.subscribe(observer);