OperationSum { 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; } }); }  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 testSumOfAFewFloats() throws Throwable { Observable<Float> src = Observable.from(1.0f); sumFloats(src).subscribe(wf); verify(wf, times(1)).onNext(anyFloat()); verify(wf).onNext(1.0f); verify(wf, never()).onError(any(Throwable.class)); verify(wf, times(1)).onCompleted(); }
@Test public void testEmptySumFloats() throws Throwable { Observable<Float> src = Observable.empty(); sumFloats(src).subscribe(wf); verify(wf, times(1)).onNext(anyFloat()); verify(wf).onNext(0.0f); verify(wf, never()).onError(any(Throwable.class)); verify(wf, times(1)).onCompleted(); }