@Test public void testRepetition() { int NUM = 10; final AtomicInteger count = new AtomicInteger(); int value = Observable.create(new OnSubscribeFunc<Integer>() { @Override public Subscription onSubscribe(Observer<? super Integer> o) { o.onNext(count.incrementAndGet()); o.onCompleted(); return Subscriptions.empty(); } }).repeat(Schedulers.computation()).take(NUM).toBlockingObservable().last(); assertEquals(NUM, value); }
public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) { return new OperationRepeat<T>(source, scheduler); }
OperationRepeat implements Observable.OnSubscribeFunc<T> { public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) { return new OperationRepeat<T>(source, scheduler); } }
OperationRepeat implements Observable.OnSubscribeFunc<T> { public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) { return new OperationRepeat<T>(source, scheduler); } private  OperationRepeat(Observable<T> source, Scheduler scheduler); }
OperationRepeat implements Observable.OnSubscribeFunc<T> { public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) { return new OperationRepeat<T>(source, scheduler); } private  OperationRepeat(Observable<T> source, Scheduler scheduler); static Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler); @Override Subscription onSubscribe(final Observer<? super T> observer); }
OperationRepeat implements Observable.OnSubscribeFunc<T> { public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) { return new OperationRepeat<T>(source, scheduler); } private  OperationRepeat(Observable<T> source, Scheduler scheduler); static Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler); @Override Subscription onSubscribe(final Observer<? super T> observer);  }