OperationAverage { public static Observable<Float> averageFloats(Observable<Float> source) { return source.reduce(new Tuple2<Float>(0.0f, 0), new Func2<Tuple2<Float>, Float, Tuple2<Float>>() { @Override public Tuple2<Float> call(Tuple2<Float> accu, Float next) { return new Tuple2<Float>(accu.current + next, accu.count + 1); } }).map(new Func1<Tuple2<Float>, Float>() { @Override public Float call(Tuple2<Float> result) { if (result.count == 0) { throw new IllegalArgumentException("Sequence contains no elements"); } return result.current / result.count; } }); }  static Observable<Integer> average(Observable<Integer> source); static Observable<Long> averageLongs(Observable<Long> source); static Observable<Float> averageFloats(Observable<Float> source); static Observable<Double> averageDoubles(Observable<Double> source);  }
@Test public void testAverageOfAFewFloats() throws Throwable { Observable<Float> src = Observable.from(1.0f, 2.0f); averageFloats(src).subscribe(wf); verify(wf, times(1)).onNext(anyFloat()); verify(wf).onNext(1.5f); verify(wf, never()).onError(any(Throwable.class)); verify(wf, times(1)).onCompleted(); }
@Test public void testEmptyAverageFloats() throws Throwable { Observable<Float> src = Observable.empty(); averageFloats(src).subscribe(wf); verify(wf, never()).onNext(anyFloat()); verify(wf, times(1)).onError(isA(IllegalArgumentException.class)); verify(wf, never()).onCompleted(); }