@Test public void someItemsAndFailure() { AtomicInteger count = new AtomicInteger(); TestSubscriber<Object> ts = new TestSubscriber<>(); Multi.defer(() -> { int c = count.incrementAndGet(); return c < 6 ? Multi.concat(Multi.just(c), Multi.error(new IOException())) : Multi.range(6, 5); }) .retry((e, n) -> e instanceof IOException) .subscribe(ts); ts .assertEmpty() .request(2) .assertValuesOnly(1, 2) .request(3) .assertValuesOnly(1, 2, 3, 4, 5) .request(5) .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { RetrySubscriber<T, U> parent = new RetrySubscriber<>(subscriber, whenFunction, source); subscriber.onSubscribe(parent); parent.retry(); }
MultiRetry implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { RetrySubscriber<T, U> parent = new RetrySubscriber<>(subscriber, whenFunction, source); subscriber.onSubscribe(parent); parent.retry(); } }
MultiRetry implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { RetrySubscriber<T, U> parent = new RetrySubscriber<>(subscriber, whenFunction, source); subscriber.onSubscribe(parent); parent.retry(); } MultiRetry(Multi<T> source,
               BiFunction<? super Throwable, ? super Long, ? extends Flow.Publisher<U>> whenFunction);   MultiRetry(Multi<T> source, long count);   MultiRetry(Multi<T> source, BiPredicate<? super Throwable, ? super Long> predicate); }
MultiRetry implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { RetrySubscriber<T, U> parent = new RetrySubscriber<>(subscriber, whenFunction, source); subscriber.onSubscribe(parent); parent.retry(); } MultiRetry(Multi<T> source,
               BiFunction<? super Throwable, ? super Long, ? extends Flow.Publisher<U>> whenFunction);   MultiRetry(Multi<T> source, long count);   MultiRetry(Multi<T> source, BiPredicate<? super Throwable, ? super Long> predicate); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiRetry implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { RetrySubscriber<T, U> parent = new RetrySubscriber<>(subscriber, whenFunction, source); subscriber.onSubscribe(parent); parent.retry(); } MultiRetry(Multi<T> source,
               BiFunction<? super Throwable, ? super Long, ? extends Flow.Publisher<U>> whenFunction);   MultiRetry(Multi<T> source, long count);   MultiRetry(Multi<T> source, BiPredicate<? super Throwable, ? super Long> predicate); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }