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);  }
@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); }
@Test public void testRepeatTake() { Observable<Integer> xs = Observable.from(1, 2); Object[] ys = xs.repeat(Schedulers.newThread()).take(4).toList().toBlockingObservable().last().toArray(); assertArrayEquals(new Object[] { 1, 2, 1, 2 }, ys); }
@Test public void testNoStackOverFlow() { Observable.from(1).repeat(Schedulers.newThread()).take(100000).toBlockingObservable().last(); }