OnSubscribeRedoWithNext implements OnSubscribe<T> { public static <T> Observable<T> repeatOn(Observable<T> source, final Func1<Notification<T>, Observable<?>> on, boolean stopOnComplete) { final Func1<Observable<? extends Notification<T>>, Observable<?>> func1 = new Func1<Observable<? extends Notification<T>>, Observable<?>>() { @Override public Observable<?> call(final Observable<? extends Notification<T>> observable) { return observable.flatMap(on); } }; return create(new OnSubscribeRedoWithNext<>(source, func1, Schedulers.trampoline(), stopOnComplete)); } private  OnSubscribeRedoWithNext(Observable<T> source, Func1<? super Observable<? extends Notification<T>>, ? extends Observable<?>> f,
                                    Scheduler scheduler,
                                    boolean stopOnComplete); static Observable<T> repeatOn(Observable<T> source, final Func1<Notification<T>, Observable<?>> on, boolean stopOnComplete); static Observable<T> repeatOn(Observable<T> source, final Func1<Notification<T>, Observable<?>> on); static Observable<T> repeat(Observable<T> source, Func1<? super Observable<? extends Notification<T>>, ? extends Observable<?>> notificationHandler, boolean stopOnComplete); static Observable<T> repeat(Observable<T> source, Func1<? super Observable<? extends Notification<T>>, ? extends Observable<?>> notificationHandler); @Override void call(final Subscriber<? super T> child);  }
@Test public void testOnCompleted_onCompleted() throws Exception { returnError = false; OnSubscribeRedoWithNext.repeatOn(apiResponse, new Func1<Notification<Boolean>, Observable<?>>() { @Override public Observable<?> call(final Notification<Boolean> notification) { if (notification.isOnNext()) { final Boolean isError = notification.getValue(); return isError ? Observable.timer(1, TimeUnit.SECONDS, scheduler) : Observable.empty(); } else { return Observable.empty(); } } }, true) .subscribe(responseOnErrorObserver); verify(responseOnErrorObserver).onNext(false); verify(responseOnErrorObserver).onCompleted(); verifyNoMoreInteractions(responseOnErrorObserver); }