OperationAverage { public static Observable<Integer> average(Observable<Integer> source) { return source.reduce(new Tuple2<Integer>(0, 0), new Func2<Tuple2<Integer>, Integer, Tuple2<Integer>>() { @Override public Tuple2<Integer> call(Tuple2<Integer> accu, Integer next) { return new Tuple2<Integer>(accu.current + next, accu.count + 1); } }).map(new Func1<Tuple2<Integer>, Integer>() { @Override public Integer call(Tuple2<Integer> 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 testAverageOfAFewInts() throws Throwable { Observable<Integer> src = Observable.from(1, 2, 3, 4, 6); average(src).subscribe(w); verify(w, times(1)).onNext(anyInt()); verify(w).onNext(3); verify(w, never()).onError(any(Throwable.class)); verify(w, times(1)).onCompleted(); }
@Test public void testEmptyAverage() throws Throwable { Observable<Integer> src = Observable.empty(); average(src).subscribe(w); verify(w, never()).onNext(anyInt()); verify(w, times(1)).onError(isA(IllegalArgumentException.class)); verify(w, never()).onCompleted(); }