OperationAverage { public static Observable<Double> averageDoubles(Observable<Double> source) { return source.reduce(new Tuple2<Double>(0.0d, 0), new Func2<Tuple2<Double>, Double, Tuple2<Double>>() { @Override public Tuple2<Double> call(Tuple2<Double> accu, Double next) { return new Tuple2<Double>(accu.current + next, accu.count + 1); } }).map(new Func1<Tuple2<Double>, Double>() { @Override public Double call(Tuple2<Double> 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 testAverageOfAFewDoubles() throws Throwable { Observable<Double> src = Observable.from(1.0d, 2.0d); averageDoubles(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 testEmptyAverageDoubles() throws Throwable { Observable<Double> src = Observable.empty(); averageDoubles(src).subscribe(wd); verify(wd, never()).onNext(anyDouble()); verify(wd, times(1)).onError(isA(IllegalArgumentException.class)); verify(wd, never()).onCompleted(); }