OperationSum { 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; } }); }  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 testSumOfAFewLongs() throws Throwable { Observable<Long> src = Observable.from(1L, 2L, 3L, 4L, 5L); sumLongs(src).subscribe(wl); verify(wl, times(1)).onNext(anyLong()); verify(wl).onNext(15L); verify(wl, never()).onError(any(Throwable.class)); verify(wl, times(1)).onCompleted(); }
@Test public void testEmptySumLongs() throws Throwable { Observable<Long> src = Observable.empty(); sumLongs(src).subscribe(wl); verify(wl, times(1)).onNext(anyLong()); verify(wl).onNext(0L); verify(wl, never()).onError(any(Throwable.class)); verify(wl, times(1)).onCompleted(); }