From 67374198e9dc1e3e00dd56f5c53984af3cc1a721 Mon Sep 17 00:00:00 2001
From: George Campbell <gscampbell@netflix.com>
Date: Tue, 28 Jan 2014 00:15:02 -0800
Subject: [PATCH] Convert to scan to use lift

---
 rxjava-core/src/main/java/rx/Observable.java  |  26 +-
 .../main/java/rx/operators/OperationScan.java | 188 +++++------
 .../main/java/rx/operators/OperationSum.java  | 291 +++---------------
 rxjava-core/src/test/java/rx/ReduceTests.java |   2 +-
 .../java/rx/operators/OperationScanTest.java  |   6 +-
 .../java/rx/operators/OperationSumTest.java   |   4 +-
 6 files changed, 132 insertions(+), 385 deletions(-)

diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 544a60d712..a5a265fa8b 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -2529,14 +2529,6 @@ public final static <T> Observable<Boolean> sequenceEqual(Observable<? extends T
         return OperationSequenceEqual.sequenceEqual(first, second, equality);
     }
 
-    /**
-     * @deprecated use {@link #sumInteger}
-     */
-    @Deprecated
-    public final static Observable<Integer> sum(Observable<Integer> source) {
-        return OperationSum.sum(source);
-    }
-
     /**
      * Returns an Observable that emits the sum of all the Doubles emitted by the source Observable.
      * <p>
@@ -2583,7 +2575,7 @@ public final static Observable<Float> sumFloat(Observable<Float> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum.aspx">MSDN: Observable.Sum</a>
      */
     public final static Observable<Integer> sumInteger(Observable<Integer> source) {
-        return OperationSum.sum(source);
+        return OperationSum.sumIntegers(source);
     }
 
     /**
@@ -5519,7 +5511,7 @@ public final Observable<T> reduce(Func2<T, T, T> accumulator) {
          * It should use last() not takeLast(1) since it needs to emit an error if the sequence is
          * empty.
          */
-        return create(OperationScan.scan(this, accumulator)).last();
+        return scan(accumulator).last();
     }
 
     /**
@@ -5547,7 +5539,7 @@ public final Observable<T> reduce(Func2<T, T, T> accumulator) {
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
     public final <R> Observable<R> reduce(R initialValue, Func2<R, ? super T, R> accumulator) {
-        return create(OperationScan.scan(this, initialValue, accumulator)).takeLast(1);
+        return scan(initialValue, accumulator).takeLast(1);
     }
 
     /**
@@ -6148,7 +6140,7 @@ public final <U> Observable<T> sample(Observable<U> sampler) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665.aspx">MSDN: Observable.Scan</a>
      */
     public final Observable<T> scan(Func2<T, T, T> accumulator) {
-        return create(OperationScan.scan(this, accumulator));
+        return lift(OperationScan.scan(accumulator));
     }
 
     /**
@@ -6175,7 +6167,7 @@ public final Observable<T> scan(Func2<T, T, T> accumulator) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665.aspx">MSDN: Observable.Scan</a>
      */
     public final <R> Observable<R> scan(R initialValue, Func2<R, ? super T, R> accumulator) {
-        return create(OperationScan.scan(this, initialValue, accumulator));
+        return lift(OperationScan.scan(initialValue, accumulator));
     }
 
     /**
@@ -7079,7 +7071,7 @@ public final Observable<T> subscribeOn(Scheduler scheduler) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum.aspx">MSDN: Observable.Sum</a>
      */
     public final Observable<Double> sumDouble(Func1<? super T, Double> valueExtractor) {
-        return create(new OperationSum.SumDoubleExtractor<T>(this, valueExtractor));
+        return OperationSum.sumAtLeastOneDoubles(map(valueExtractor));
     }
 
     /**
@@ -7096,7 +7088,7 @@ public final Observable<Double> sumDouble(Func1<? super T, Double> valueExtracto
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum.aspx">MSDN: Observable.Sum</a>
      */
     public final Observable<Float> sumFloat(Func1<? super T, Float> valueExtractor) {
-        return create(new OperationSum.SumFloatExtractor<T>(this, valueExtractor));
+        return OperationSum.sumAtLeastOneFloats(map(valueExtractor));
     }
 
     /**
@@ -7113,7 +7105,7 @@ public final Observable<Float> sumFloat(Func1<? super T, Float> valueExtractor)
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum.aspx">MSDN: Observable.Sum</a>
      */
     public final Observable<Integer> sumInteger(Func1<? super T, Integer> valueExtractor) {
-        return create(new OperationSum.SumIntegerExtractor<T>(this, valueExtractor));
+        return OperationSum.sumAtLeastOneIntegers(map(valueExtractor));
     }
 
     /**
@@ -7130,7 +7122,7 @@ public final Observable<Integer> sumInteger(Func1<? super T, Integer> valueExtra
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum.aspx">MSDN: Observable.Sum</a>
      */
     public final Observable<Long> sumLong(Func1<? super T, Long> valueExtractor) {
-        return create(new OperationSum.SumLongExtractor<T>(this, valueExtractor));
+        return OperationSum.sumAtLeastOneLongs(map(valueExtractor));
     }
 
     /**
diff --git a/rxjava-core/src/main/java/rx/operators/OperationScan.java b/rxjava-core/src/main/java/rx/operators/OperationScan.java
index 3c58e053c5..8bbaeb0e59 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationScan.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationScan.java
@@ -15,10 +15,8 @@
  */
 package rx.operators;
 
-import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
+import rx.Observable.Operator;
+import rx.Subscriber;
 import rx.util.functions.Func2;
 
 /**
@@ -36,7 +34,8 @@
  */
 public final class OperationScan {
     /**
-     * Applies an accumulator function over an observable sequence and returns each intermediate result with the specified source and accumulator.
+     * Applies an accumulator function over an observable sequence and returns each intermediate
+     * result with the specified source and accumulator.
      * 
      * @param sequence
      *            An observable sequence of elements to project.
@@ -45,124 +44,97 @@ public final class OperationScan {
      * @param accumulator
      *            An accumulator function to be invoked on each element from the sequence.
      * 
-     * @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,
+     * @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<R, ? super T, R> accumulator) {
-        return new Accumulator<T, R>(sequence, initialValue, accumulator);
+    public static <T, R> Operator<R, T> scan(final R initialValue, final Func2<R, ? super T, R> accumulator) {
+        return new Operator<R, T>() {
+            @Override
+            public Subscriber<T> call(final Subscriber<? super R> observer) {
+                observer.onNext(initialValue);
+                return new Subscriber<T>(observer) {
+                    private R value = initialValue;
+
+                    @Override
+                    public void onNext(T value) {
+                        try {
+                            this.value = accumulator.call(this.value, value);
+                        } catch (Throwable e) {
+                            observer.onError(e);
+                            observer.unsubscribe();
+                        }
+                        observer.onNext(this.value);
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        observer.onError(e);
+                    }
+
+                    @Override
+                    public void onCompleted() {
+                        observer.onCompleted();
+                    }
+                };
+            }
+        };
     }
 
     /**
-     * Applies an accumulator function over an observable sequence and returns each intermediate result with the specified source and accumulator.
+     * Applies an accumulator function over an observable sequence and returns each intermediate
+     * result with the specified source and accumulator.
      * 
      * @param sequence
      *            An observable sequence of elements to project.
      * @param accumulator
      *            An accumulator function to be invoked on each element from the sequence.
      * 
-     * @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>
+     * @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<T, T, T> accumulator) {
-        return new AccuWithoutInitialValue<T>(sequence, accumulator);
-    }
-
-    private static class AccuWithoutInitialValue<T> implements OnSubscribeFunc<T> {
-        private final Observable<? extends T> sequence;
-        private final Func2<T, T, T> accumulatorFunction;
-
-        private AccumulatingObserver<T, T> accumulatingObserver;
-
-        private AccuWithoutInitialValue(Observable<? extends T> sequence, Func2<T, T, T> accumulator) {
-            this.sequence = sequence;
-            this.accumulatorFunction = accumulator;
-        }
-
-        @Override
-        public Subscription onSubscribe(final Observer<? super T> observer) {
-            return sequence.subscribe(new Observer<T>() {
-
-                // has to be synchronized so that the initial value is always sent only once.
-                @Override
-                public synchronized void onNext(T value) {
-                    if (accumulatingObserver == null) {
-                        observer.onNext(value);
-                        accumulatingObserver = new AccumulatingObserver<T, T>(observer, value, accumulatorFunction);
-                    } else {
-                        accumulatingObserver.onNext(value);
+    public static <T> Operator<T, T> scan(final Func2<T, T, T> accumulator) {
+        return new Operator<T, T>() {
+            @Override
+            public Subscriber<T> call(final Subscriber<? super T> observer) {
+                return new Subscriber<T>(observer) {
+                    private boolean first = true;
+                    private T value;
+
+                    @Override
+                    public void onNext(T value) {
+                        if (first) {
+                            this.value = value;
+                            first = false;
+                        }
+                        else {
+                            try {
+                                this.value = accumulator.call(this.value, value);
+                            } catch (Throwable e) {
+                                observer.onError(e);
+                                observer.unsubscribe();
+                            }
+                        }
+                        observer.onNext(this.value);
                     }
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    observer.onError(e);
-                }
 
-                @Override
-                public void onCompleted() {
-                    observer.onCompleted();
-                }
-            });
-        }
-    }
-
-    private static class Accumulator<T, R> implements OnSubscribeFunc<R> {
-        private final Observable<? extends T> sequence;
-        private final R initialValue;
-        private final Func2<R, ? super T, R> accumulatorFunction;
-
-        private Accumulator(Observable<? extends T> sequence, R initialValue, Func2<R, ? super T, R> accumulator) {
-            this.sequence = sequence;
-            this.initialValue = initialValue;
-            this.accumulatorFunction = accumulator;
-        }
-
-        @Override
-        public Subscription onSubscribe(final Observer<? super R> observer) {
-            observer.onNext(initialValue);
-            return sequence.subscribe(new AccumulatingObserver<T, R>(observer, initialValue, accumulatorFunction));
-        }
-    }
-
-    private static class AccumulatingObserver<T, R> implements Observer<T> {
-        private final Observer<? super R> observer;
-        private final Func2<R, ? super T, R> accumulatorFunction;
-
-        private R acc;
-
-        private AccumulatingObserver(Observer<? super R> observer, R initialValue, Func2<R, ? super T, R> accumulator) {
-            this.observer = observer;
-            this.accumulatorFunction = accumulator;
-
-            this.acc = initialValue;
-        }
+                    @Override
+                    public void onError(Throwable e) {
+                        observer.onError(e);
+                    }
 
-        /**
-         * We must synchronize this because we can't allow
-         * multiple threads to execute the 'accumulatorFunction' at the same time because
-         * the accumulator code very often will be doing mutation of the 'acc' object such as a non-threadsafe HashMap
-         * 
-         * Because it's synchronized it's using non-atomic variables since everything in this method is single-threaded
-         */
-        @Override
-        public synchronized void onNext(T value) {
-            try {
-                acc = accumulatorFunction.call(acc, value);
-                observer.onNext(acc);
-            } catch (Throwable ex) {
-                observer.onError(ex);
+                    @Override
+                    public void onCompleted() {
+                        observer.onCompleted();
+                    }
+                };
             }
-        }
-
-        @Override
-        public void onError(Throwable e) {
-            observer.onError(e);
-        }
-
-        @Override
-        public void onCompleted() {
-            observer.onCompleted();
-        }
+        };
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSum.java b/rxjava-core/src/main/java/rx/operators/OperationSum.java
index 4a12335e42..65d4c8e4a9 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationSum.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationSum.java
@@ -16,291 +16,74 @@
 package rx.operators;
 
 import rx.Observable;
-import rx.Observer;
-import rx.Subscription;
-import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
 /**
  * A few operators for implementing the sum operation.
  * 
- * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum%28v=vs.103%29.aspx">MSDN: Observable.Sum</a>
+ * @see <a
+ *      href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.sum%28v=vs.103%29.aspx">MSDN:
+ *      Observable.Sum</a>
  */
 public final class OperationSum {
-    public static Observable<Integer> sum(Observable<Integer> source) {
-        return source.reduce(0, new Func2<Integer, Integer, Integer>() {
-            @Override
-            public Integer call(Integer accu, Integer next) {
-                return accu + next;
-            }
-        });
+
+    public static Observable<Integer> sumIntegers(Observable<Integer> source) {
+        return source.reduce(0, ACCUM_INT);
     }
 
     public static Observable<Long> sumLongs(Observable<Long> source) {
-        return source.reduce(0L, new Func2<Long, Long, Long>() {
-            @Override
-            public Long call(Long accu, Long next) {
-                return accu + next;
-            }
-        });
+        return source.reduce(0l, ACCUM_LONG);
     }
 
     public static Observable<Float> sumFloats(Observable<Float> source) {
-        return source.reduce(0.0f, new Func2<Float, Float, Float>() {
-            @Override
-            public Float call(Float accu, Float next) {
-                return accu + next;
-            }
-        });
+        return source.reduce(0.0f, ACCUM_FLOAT);
     }
 
     public static Observable<Double> sumDoubles(Observable<Double> source) {
-        return source.reduce(0.0d, new Func2<Double, Double, Double>() {
-            @Override
-            public Double call(Double accu, Double next) {
-                return accu + next;
-            }
-        });
+        return source.reduce(0.0d, ACCUM_DOUBLE);
     }
 
-    /**
-     * Compute the sum by extracting integer values from the source via an
-     * extractor function.
-     * 
-     * @param <T>
-     *            the source value type
-     */
-    public static final class SumIntegerExtractor<T> implements Observable.OnSubscribeFunc<Integer> {
-        final Observable<? extends T> source;
-        final Func1<? super T, Integer> valueExtractor;
-
-        public SumIntegerExtractor(Observable<? extends T> source, Func1<? super T, Integer> valueExtractor) {
-            this.source = source;
-            this.valueExtractor = valueExtractor;
-        }
-
-        @Override
-        public Subscription onSubscribe(Observer<? super Integer> t1) {
-            return source.subscribe(new SumObserver(t1));
-        }
-
-        /** Computes the average. */
-        private final class SumObserver implements Observer<T> {
-            final Observer<? super Integer> observer;
-            int sum;
-            boolean hasValue;
-
-            public SumObserver(Observer<? super Integer> observer) {
-                this.observer = observer;
-            }
-
-            @Override
-            public void onNext(T args) {
-                sum += valueExtractor.call(args);
-                hasValue = true;
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                observer.onError(e);
-            }
-
-            @Override
-            public void onCompleted() {
-                if (hasValue) {
-                    try {
-                        observer.onNext(sum);
-                    } catch (Throwable t) {
-                        observer.onError(t);
-                        return;
-                    }
-                    observer.onCompleted();
-                } else {
-                    observer.onError(new IllegalArgumentException("Sequence contains no elements"));
-                }
-            }
-
-        }
+    public static Observable<Integer> sumAtLeastOneIntegers(Observable<Integer> source) {
+        return source.reduce(ACCUM_INT);
     }
 
-    /**
-     * Compute the sum by extracting long values from the source via an
-     * extractor function.
-     * 
-     * @param <T>
-     *            the source value type
-     */
-    public static final class SumLongExtractor<T> implements Observable.OnSubscribeFunc<Long> {
-        final Observable<? extends T> source;
-        final Func1<? super T, Long> valueExtractor;
-
-        public SumLongExtractor(Observable<? extends T> source, Func1<? super T, Long> valueExtractor) {
-            this.source = source;
-            this.valueExtractor = valueExtractor;
-        }
-
-        @Override
-        public Subscription onSubscribe(Observer<? super Long> t1) {
-            return source.subscribe(new SumObserver(t1));
-        }
-
-        /** Computes the average. */
-        private final class SumObserver implements Observer<T> {
-            final Observer<? super Long> observer;
-            long sum;
-            boolean hasValue;
-
-            public SumObserver(Observer<? super Long> observer) {
-                this.observer = observer;
-            }
-
-            @Override
-            public void onNext(T args) {
-                sum += valueExtractor.call(args);
-                hasValue = true;
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                observer.onError(e);
-            }
-
-            @Override
-            public void onCompleted() {
-                if (hasValue) {
-                    try {
-                        observer.onNext(sum);
-                    } catch (Throwable t) {
-                        observer.onError(t);
-                        return;
-                    }
-                    observer.onCompleted();
-                } else {
-                    observer.onError(new IllegalArgumentException("Sequence contains no elements"));
-                }
-            }
-
-        }
+    public static Observable<Long> sumAtLeastOneLongs(Observable<Long> source) {
+        return source.reduce(ACCUM_LONG);
     }
 
-    /**
-     * Compute the sum by extracting float values from the source via an
-     * extractor function.
-     * 
-     * @param <T>
-     *            the source value type
-     */
-    public static final class SumFloatExtractor<T> implements Observable.OnSubscribeFunc<Float> {
-        final Observable<? extends T> source;
-        final Func1<? super T, Float> valueExtractor;
+    public static Observable<Float> sumAtLeastOneFloats(Observable<Float> source) {
+        return source.reduce(ACCUM_FLOAT);
+    }
 
-        public SumFloatExtractor(Observable<? extends T> source, Func1<? super T, Float> valueExtractor) {
-            this.source = source;
-            this.valueExtractor = valueExtractor;
-        }
+    public static Observable<Double> sumAtLeastOneDoubles(Observable<Double> source) {
+        return source.reduce(ACCUM_DOUBLE);
+    }
 
+    private static final Func2<Integer, Integer, Integer> ACCUM_INT = new Func2<Integer, Integer, Integer>() {
         @Override
-        public Subscription onSubscribe(Observer<? super Float> t1) {
-            return source.subscribe(new SumObserver(t1));
+        public Integer call(Integer accu, Integer next) {
+            return accu + next;
         }
+    };
 
-        /** Computes the average. */
-        private final class SumObserver implements Observer<T> {
-            final Observer<? super Float> observer;
-            float sum;
-            boolean hasValue;
-
-            public SumObserver(Observer<? super Float> observer) {
-                this.observer = observer;
-            }
-
-            @Override
-            public void onNext(T args) {
-                sum += valueExtractor.call(args);
-                hasValue = true;
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                observer.onError(e);
-            }
-
-            @Override
-            public void onCompleted() {
-                if (hasValue) {
-                    try {
-                        observer.onNext(sum);
-                    } catch (Throwable t) {
-                        observer.onError(t);
-                        return;
-                    }
-                    observer.onCompleted();
-                } else {
-                    observer.onError(new IllegalArgumentException("Sequence contains no elements"));
-                }
-            }
-
-        }
-    }
-
-    /**
-     * Compute the sum by extracting float values from the source via an
-     * extractor function.
-     * 
-     * @param <T>
-     *            the source value type
-     */
-    public static final class SumDoubleExtractor<T> implements Observable.OnSubscribeFunc<Double> {
-        final Observable<? extends T> source;
-        final Func1<? super T, Double> valueExtractor;
-
-        public SumDoubleExtractor(Observable<? extends T> source, Func1<? super T, Double> valueExtractor) {
-            this.source = source;
-            this.valueExtractor = valueExtractor;
+    private static final Func2<Long, Long, Long> ACCUM_LONG = new Func2<Long, Long, Long>() {
+        @Override
+        public Long call(Long accu, Long next) {
+            return accu + next;
         }
+    };
 
+    private static final Func2<Float, Float, Float> ACCUM_FLOAT = new Func2<Float, Float, Float>() {
         @Override
-        public Subscription onSubscribe(Observer<? super Double> t1) {
-            return source.subscribe(new SumObserver(t1));
+        public Float call(Float accu, Float next) {
+            return accu + next;
         }
+    };
 
-        /** Computes the average. */
-        private final class SumObserver implements Observer<T> {
-            final Observer<? super Double> observer;
-            double sum;
-            boolean hasValue;
-
-            public SumObserver(Observer<? super Double> observer) {
-                this.observer = observer;
-            }
-
-            @Override
-            public void onNext(T args) {
-                sum += valueExtractor.call(args);
-                hasValue = true;
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                observer.onError(e);
-            }
-
-            @Override
-            public void onCompleted() {
-                if (hasValue) {
-                    try {
-                        observer.onNext(sum);
-                    } catch (Throwable t) {
-                        observer.onError(t);
-                        return;
-                    }
-                    observer.onCompleted();
-                } else {
-                    observer.onError(new IllegalArgumentException("Sequence contains no elements"));
-                }
-            }
-
+    private static final Func2<Double, Double, Double> ACCUM_DOUBLE = new Func2<Double, Double, Double>() {
+        @Override
+        public Double call(Double accu, Double next) {
+            return accu + next;
         }
-    }
-
+    };
 }
diff --git a/rxjava-core/src/test/java/rx/ReduceTests.java b/rxjava-core/src/test/java/rx/ReduceTests.java
index 20d1b8811c..8704f80d61 100644
--- a/rxjava-core/src/test/java/rx/ReduceTests.java
+++ b/rxjava-core/src/test/java/rx/ReduceTests.java
@@ -52,7 +52,7 @@ public Movie call(Movie t1, Movie t2) {
                     }
                 };
 
-        Observable<Movie> reduceResult = Observable.create(OperationScan.scan(horrorMovies, chooseSecondMovie)).takeLast(1);
+        Observable<Movie> reduceResult = horrorMovies.lift(OperationScan.scan(chooseSecondMovie)).takeLast(1);
 
         Observable<Movie> reduceResult2 = horrorMovies.reduce(chooseSecondMovie);
     }
diff --git a/rxjava-core/src/test/java/rx/operators/OperationScanTest.java b/rxjava-core/src/test/java/rx/operators/OperationScanTest.java
index 47a3274e52..2e1613353b 100644
--- a/rxjava-core/src/test/java/rx/operators/OperationScanTest.java
+++ b/rxjava-core/src/test/java/rx/operators/OperationScanTest.java
@@ -41,7 +41,7 @@ public void testScanIntegersWithInitialValue() {
 
         Observable<Integer> observable = Observable.from(1, 2, 3);
 
-        Observable<String> m = Observable.create(scan(observable, "", new Func2<String, Integer, String>() {
+        Observable<String> m = observable.lift(scan("", new Func2<String, Integer, String>() {
 
             @Override
             public String call(String s, Integer n) {
@@ -68,7 +68,7 @@ public void testScanIntegersWithoutInitialValue() {
 
         Observable<Integer> observable = Observable.from(1, 2, 3);
 
-        Observable<Integer> m = Observable.create(scan(observable, new Func2<Integer, Integer, Integer>() {
+        Observable<Integer> m = observable.lift(scan(new Func2<Integer, Integer, Integer>() {
 
             @Override
             public Integer call(Integer t1, Integer t2) {
@@ -95,7 +95,7 @@ public void testScanIntegersWithoutInitialValueAndOnlyOneValue() {
 
         Observable<Integer> observable = Observable.from(1);
 
-        Observable<Integer> m = Observable.create(scan(observable, new Func2<Integer, Integer, Integer>() {
+        Observable<Integer> m = observable.lift(scan(new Func2<Integer, Integer, Integer>() {
 
             @Override
             public Integer call(Integer t1, Integer t2) {
diff --git a/rxjava-core/src/test/java/rx/operators/OperationSumTest.java b/rxjava-core/src/test/java/rx/operators/OperationSumTest.java
index 478158eb21..7253e6196f 100644
--- a/rxjava-core/src/test/java/rx/operators/OperationSumTest.java
+++ b/rxjava-core/src/test/java/rx/operators/OperationSumTest.java
@@ -39,7 +39,7 @@ public class OperationSumTest {
     @Test
     public void testSumOfAFewInts() throws Throwable {
         Observable<Integer> src = Observable.from(1, 2, 3, 4, 5);
-        sum(src).subscribe(w);
+        sumIntegers(src).subscribe(w);
 
         verify(w, times(1)).onNext(anyInt());
         verify(w).onNext(15);
@@ -50,7 +50,7 @@ public void testSumOfAFewInts() throws Throwable {
     @Test
     public void testEmptySum() throws Throwable {
         Observable<Integer> src = Observable.empty();
-        sum(src).subscribe(w);
+        sumIntegers(src).subscribe(w);
 
         verify(w, times(1)).onNext(anyInt());
         verify(w).onNext(0);