OperationSum { 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; } }); }  static Observable<Integer> sum(Observable<Integer> source); static Observable<Long> sumLongs(Observable<Long> source); static Observable<Float> sumFloats(Observable<Float> source); static Observable<Double> sumDoubles(Observable<Double> source);  }
@Test public void testSumOfAFewDoubles() throws Throwable { Observable<Double> src = Observable.from(0.0d, 1.0d, 0.5d); sumDoubles(src).subscribe(wd); verify(wd, times(1)).onNext(anyDouble()); verify(wd).onNext(1.5d); verify(wd, never()).onError(any(Throwable.class)); verify(wd, times(1)).onCompleted(); }
@Test public void testEmptySumDoubles() throws Throwable { Observable<Double> src = Observable.empty(); sumDoubles(src).subscribe(wd); verify(wd, times(1)).onNext(anyDouble()); verify(wd).onNext(0.0d); verify(wd, never()).onError(any(Throwable.class)); verify(wd, times(1)).onCompleted(); }