@Test public void testTransformer() throws Exception { Callable<String> mockCallable = (Callable<String>) mock(Callable.class); when(mockCallable.call()).thenReturn("value"); PublishSubject<String> source = PublishSubject.create(); TestObserver<String> testSubscriber = Observable.fromCallable(mockCallable) .compose(RepeatWhenEmits.<String>from(source)) .test(); for (int i = 1; i < 10; i++) { verify(mockCallable, times(i)).call(); testSubscriber.assertValueCount(i); source.onNext("event"); } }
@Nonnull static <T> RepeatWhenEmits<T> from(@Nonnull Observable source) { return new RepeatWhenEmits<>(checkNotNull(source)); }
RepeatWhenEmits implements ObservableTransformer<T, T> { @Nonnull static <T> RepeatWhenEmits<T> from(@Nonnull Observable source) { return new RepeatWhenEmits<>(checkNotNull(source)); } }
RepeatWhenEmits implements ObservableTransformer<T, T> { @Nonnull static <T> RepeatWhenEmits<T> from(@Nonnull Observable source) { return new RepeatWhenEmits<>(checkNotNull(source)); } private  RepeatWhenEmits(@Nonnull Observable source); }
RepeatWhenEmits implements ObservableTransformer<T, T> { @Nonnull static <T> RepeatWhenEmits<T> from(@Nonnull Observable source) { return new RepeatWhenEmits<>(checkNotNull(source)); } private  RepeatWhenEmits(@Nonnull Observable source); @Override ObservableSource<T> apply(Observable<T> upstream); }
RepeatWhenEmits implements ObservableTransformer<T, T> { @Nonnull static <T> RepeatWhenEmits<T> from(@Nonnull Observable source) { return new RepeatWhenEmits<>(checkNotNull(source)); } private  RepeatWhenEmits(@Nonnull Observable source); @Override ObservableSource<T> apply(Observable<T> upstream);  }