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; } }); }  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 testSumOfAFewInts() throws Throwable { Observable<Integer> src = Observable.from(1, 2, 3, 4, 5); sum(src).subscribe(w); verify(w, times(1)).onNext(anyInt()); verify(w).onNext(15); verify(w, never()).onError(any(Throwable.class)); verify(w, times(1)).onCompleted(); }
@Test public void testEmptySum() throws Throwable { Observable<Integer> src = Observable.empty(); sum(src).subscribe(w); verify(w, times(1)).onNext(anyInt()); verify(w).onNext(0); verify(w, never()).onError(any(Throwable.class)); verify(w, times(1)).onCompleted(); }