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