From 99215f5583a65aaad0a783922a6a6abb2620dafd Mon Sep 17 00:00:00 2001 From: akarnokd Date: Sun, 12 Jul 2026 12:26:47 +0200 Subject: [PATCH] 4.x: Streamable + repeat, repeatWhen, retry, retryWhen --- .../io/reactivex/rxjava4/core/Streamable.java | 110 +++++++++++ .../rxjava4/disposables/Disposable.java | 2 +- .../streamable/StreamableRepeat.java | 153 +++++++++++++++ .../operators/streamable/StreamableRetry.java | 155 +++++++++++++++ .../streamable/StreamableRepeatTest.java | 151 +++++++++++++++ .../streamable/StreamableRetryTest.java | 176 ++++++++++++++++++ .../validators/CheckParamValidationTest.java | 2 + 7 files changed, 748 insertions(+), 1 deletion(-) create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeat.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetry.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeatTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetryTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 390523a5020..1b9f45e1257 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -865,6 +865,116 @@ default Streamable onErrorResumeNext(@NonNull Function(this, fallbackMapper)); } + /** + * Runs the upstream at most the given {@code count} times while it succeeds. + *

+ * So a {@code repeat(1)} will try to consume the upstream once. + * @param count the number of retries if the upstream fails + * @return the new {@code Streamable} instance + * @throws IllegalArgumentException if {@code count} is negative + */ + @CheckReturnValue + @NonNull + default Streamable repeat(long count) { + if (count < 0) { + throw new IllegalArgumentException("count >= 0 required but it was " + count); + } + if (count == 0) { + return empty(); + } + return RxJavaPlugins.onAssembly(new StreamableRepeat<>(this, + v -> v + 1 < count ? Streamer.NEXT_TRUE : Streamer.NEXT_FALSE)); + } + + /** + * Repeats the upstream when the given function signals {@code true} via the + * {@link CompletionStage} for the count how many times the upstream was streamed. + *

+ * The first repeat run will present 0 for the function. + * @param whenFunction the function to call with the run index, + * it should signal {@code true} to repeat the source, {@code false} + * to complete without error or complete exceptionally to become the + * failure result of the sequence. + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code whenFunction} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable repeatWhen(Function> whenFunction) { + Objects.requireNonNull(whenFunction, "whenFunction is null"); + return RxJavaPlugins.onAssembly(new StreamableRepeat<>(this, whenFunction)); + } + + /** + * Retries at most the given {@code count} times the upstream if it fails with any error. + *

+ * So a {@code retry(1)} will try to consume the upstream twice. + * @param count the number of retries if the upstream fails + * @return the new {@code Streamable} instance + * @throws IllegalArgumentException if {@code count} is negative + */ + @CheckReturnValue + @NonNull + default Streamable retry(long count) { + if (count < 0) { + throw new IllegalArgumentException("count >= 0 required but it was " + count); + } + return RxJavaPlugins.onAssembly(new StreamableRetry<>(this, + (v, e) -> v < count ? Streamer.NEXT_TRUE : CompletableFuture.failedStage(e))); + } + + /** + * Retries the upstream if the given predicate returns {@code true} for the + * failure {@link Throwable} of the last streaming of the upstream. + * @param predicate the p + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code predicate} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable retry(Predicate predicate) { + Objects.requireNonNull(predicate, "predicate is null"); + return RxJavaPlugins.onAssembly(new StreamableRetry<>(this, + (_, e) -> predicate.test(e) ? Streamer.NEXT_TRUE : CompletableFuture.failedStage(e))); + } + + /** + * Retries the upstream if the given predicate returns {@code true} for the + * failure count and {@link Throwable} of the last streaming of the upstream. + *

+ * The first failure run will present 0 for the predicate. + * @param predicate the function to call with the failure index and {@code Throwable} + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code predicate} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable retry(BiPredicate predicate) { + Objects.requireNonNull(predicate, "predicate is null"); + return RxJavaPlugins.onAssembly(new StreamableRetry<>(this, + (v, e) -> predicate.test(v, e) ? Streamer.NEXT_TRUE : CompletableFuture.failedStage(e))); + } + + /** + * Retries the upstream when the given predicate signals {@code true} via the + * {@link CompletionStage} for the failure count and {@link Throwable} of the last + * streaming of the upstream. + *

+ * The first failure run will present 0 for the predicate. + * @param whenFunction the function to call with the failure index and {@code Throwable}, + * it should signal {@code true} to retry the source, {@code false} + * to complete without error or complete exceptionally to become the + * failure result of the sequence. + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code whenFunction} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable retryWhen(BiFunction> whenFunction) { + Objects.requireNonNull(whenFunction, "whenFunction is null"); + return RxJavaPlugins.onAssembly(new StreamableRetry<>(this, whenFunction)); + } + /** * Skips the first {@code count} items and relays the rest to the downstream. * @param count the number of items to skip diff --git a/src/main/java/io/reactivex/rxjava4/disposables/Disposable.java b/src/main/java/io/reactivex/rxjava4/disposables/Disposable.java index c18fcc85c65..7b2e01750f6 100644 --- a/src/main/java/io/reactivex/rxjava4/disposables/Disposable.java +++ b/src/main/java/io/reactivex/rxjava4/disposables/Disposable.java @@ -23,7 +23,7 @@ import io.reactivex.rxjava4.internal.functions.Functions; /** - * Represents a disposable resource. + * Represents a disposable resource or ongoing task. */ public interface Disposable { /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeat.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeat.java new file mode 100644 index 00000000000..a34bf0f3628 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeat.java @@ -0,0 +1,153 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableRepeat( + Streamable source, + Function> whenFunction +) +implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + var streamer = new RepeatStreamer<>(source, cancellation, whenFunction); + streamer.retrySource(); + return streamer; + } + + static final class RepeatStreamer + implements Streamer, BiConsumer { + + final Streamable source; + + final StreamerCancellation downstreamCancellation; + + final Function> whenFunction; + + final AtomicInteger wipSource; + + Streamer currentStreamer; + + CompletableFuture nextWaiter; + + volatile int stage; + + long completionCount; + + Disposable whenFunctionCancel; + + RepeatStreamer(Streamable source, StreamerCancellation downstreamCancellation, + Function> whenFunction) { + this.source = source; + this.downstreamCancellation = downstreamCancellation; + this.whenFunction = whenFunction; + this.wipSource = new AtomicInteger(); + this.stage = -1; + } + + void retrySource() { + if (wipSource.getAndIncrement() != 0) { + return; + } + do { + // FIXME some operators don't clean up their StreamerCancellations so we hand out clean ones for now + var innerCanceller = downstreamCancellation.derive(); + currentStreamer = source.stream(innerCanceller); + if (stage == 0) { + stage = 1; + currentStreamer.next().whenComplete(this); + } + } while (wipSource.decrementAndGet() != 0); + } + + @Override + public @NonNull CompletionStage next() { + nextWaiter = new CompletableFuture<>(); + stage = 1; + currentStreamer.next().whenComplete(this); + return nextWaiter; + } + + @Override + public void accept(Object t, Throwable u) { + if (stage == 1) { + if (u != null) { + nextWaiter.completeExceptionally(u); + } else + if ((Boolean)t) { + nextWaiter.complete(true); + } else { + var streamer = currentStreamer; + currentStreamer = null; + stage = 2; + streamer.finish().whenComplete(this); + + } + } else + if (stage == 2) { + if (u != null) { + nextWaiter.completeExceptionally(u); + } else { + try { + var cs = whenFunction.apply(completionCount++); + whenFunctionCancel = Disposable.fromAction(() -> cs.toCompletableFuture().cancel(true)); + downstreamCancellation.add(whenFunctionCancel); + stage = 3; + cs.whenComplete(this); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextWaiter.completeExceptionally(ex); + } + } + } else { // stage 3 + downstreamCancellation.delete(whenFunctionCancel); + whenFunctionCancel = null; + var cf = nextWaiter; + if (u != null) { + cf.completeExceptionally(u); + } else + if ((Boolean)t){ + stage = 0; + retrySource(); + } else { + cf.complete(false); + } + } + } + + @Override + public @NonNull T current() { + return currentStreamer.current(); + } + + @Override + public @NonNull CompletionStage finish() { + if (currentStreamer != null) { + return currentStreamer.finish(); + } + return FINISHED; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetry.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetry.java new file mode 100644 index 00000000000..387a42e14db --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetry.java @@ -0,0 +1,155 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.BiFunction; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableRetry( + Streamable source, + BiFunction> whenFunction +) +implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + var streamer = new RetryStreamer<>(source, cancellation, whenFunction); + streamer.retrySource(); + return streamer; + } + + static final class RetryStreamer + implements Streamer, BiConsumer { + + final Streamable source; + + final StreamerCancellation downstreamCancellation; + + final BiFunction> whenFunction; + + final AtomicInteger wipSource; + + Streamer currentStreamer; + + CompletableFuture nextWaiter; + + volatile int stage; + + long failureCount; + + Disposable whenFunctionCancel; + + Throwable currentThrowable; + + RetryStreamer(Streamable source, StreamerCancellation downstreamCancellation, + BiFunction> whenFunction) { + this.source = source; + this.downstreamCancellation = downstreamCancellation; + this.whenFunction = whenFunction; + this.wipSource = new AtomicInteger(); + this.stage = -1; + } + + void retrySource() { + if (wipSource.getAndIncrement() != 0) { + return; + } + do { + // FIXME some operators don't clean up their StreamerCancellations so we hand out clean ones for now + var innerCanceller = downstreamCancellation.derive(); + currentStreamer = source.stream(innerCanceller); + if (stage == 0) { + stage = 1; + currentStreamer.next().whenComplete(this); + } + } while (wipSource.decrementAndGet() != 0); + } + + @Override + public @NonNull CompletionStage next() { + nextWaiter = new CompletableFuture<>(); + stage = 1; + currentStreamer.next().whenComplete(this); + return nextWaiter; + } + + @Override + public void accept(Object t, Throwable u) { + if (stage == 1) { + if (u != null) { + currentThrowable = u; + var streamer = currentStreamer; + currentStreamer = null; + stage = 2; + streamer.finish().whenComplete(this); + } else { + nextWaiter.complete((Boolean)t); + } + } else + if (stage == 2) { + if (u != null) { + u.addSuppressed(currentThrowable); + } else { + u = currentThrowable; + } + currentThrowable = null; + try { + var cs = whenFunction.apply(failureCount++, u); + whenFunctionCancel = Disposable.fromAction(() -> cs.toCompletableFuture().cancel(true)); + downstreamCancellation.add(whenFunctionCancel); + stage = 3; + cs.whenComplete(this); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + ex.addSuppressed(u); + nextWaiter.completeExceptionally(ex); + } + } else { // stage 3 + downstreamCancellation.delete(whenFunctionCancel); + whenFunctionCancel = null; + var cf = nextWaiter; + if (u != null) { + cf.completeExceptionally(u); + } else + if ((Boolean)t){ + stage = 0; + retrySource(); + } else { + cf.complete(false); + } + } + } + + @Override + public @NonNull T current() { + return currentStreamer.current(); + } + + @Override + public @NonNull CompletionStage finish() { + if (currentStreamer != null) { + return currentStreamer.finish(); + } + return FINISHED; + } + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeatTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeatTest.java new file mode 100644 index 00000000000..5644698c1b6 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRepeatTest.java @@ -0,0 +1,151 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableRepeatTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.just(1) + .repeat(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 1, 1, 1, 1); + } + + @Test + public void zero() throws Throwable { + Streamable.just(1) + .repeat(0) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void one() throws Throwable { + Streamable.just(1) + .repeat(1) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void empty() throws Throwable { + Streamable.empty() + .repeat(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void error() throws Throwable { + Streamable.error(new TestException()) + .repeat(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void finishCrash() throws Throwable { + StreamableFailingFinish.MAIN_COMPLETES + .repeat(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void normalDeferred() throws Throwable { + var counter = new AtomicInteger(); + Streamable.defer(() -> Streamable.just(counter.incrementAndGet())) + .repeat(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void normalDontRepeat() throws Throwable { + Streamable.range(1, 5) + .repeatWhen(_ -> CompletableFuture.completedStage(false)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void failDontRepeat() throws Throwable { + Streamable.range(1, 5) + .repeatWhen(_ -> CompletableFuture.failedStage(new TestException())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 1, 2, 3, 4, 5); + } + + @Test + public void timeoutWhen() throws Throwable { + var ts = Streamable.empty() + .repeatWhen(_ -> new CompletableFuture<>()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + Thread.sleep(100); + + ts.cancel(); + } + + @Test + public void functionCrashes() throws Throwable { + Streamable.just(1) + .repeatWhen(_ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 1); + } + + @Test + public void functionCrashesDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.just(1) + .repeatWhen(_ -> { throw new TestException(); }) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 1); + }); + } + + @Test + public void functionCrashesEmptyDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.empty() + .repeatWhen(_ -> { throw new TestException(); }) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + }); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetryTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetryTest.java new file mode 100644 index 00000000000..2627a3bac09 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRetryTest.java @@ -0,0 +1,176 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.io.IOException; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableRetryTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .retry(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void errorDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.error(new TestException()) + .retry(5) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + }); + } + + @Test + public void error() throws Throwable { + Streamable.error(new TestException()) + .retry(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void firstErrors() throws Throwable { + AtomicInteger counter = new AtomicInteger(); + Streamable.defer(() -> { + if (counter.getAndIncrement() == 0) { + return Streamable.error(new TestException()); + } + return Streamable.range(1, 5); + }) + .retry(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void firstErrorsDebug() throws Throwable { + withCachedExecutor(exec -> { + AtomicInteger counter = new AtomicInteger(); + Streamable.defer(() -> { + if (counter.getAndIncrement() == 0) { + return Streamable.error(new TestException()); + } + return Streamable.range(1, 5); + }) + .retry(5) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + }); + } + + @Test + public void functionCrashes() throws Throwable { + Streamable.error(new TestException()) + .retry(_ -> { throw new IOException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(IOException.class) + .assertError(e -> e.getSuppressed()[0] instanceof TestException); + } + + @Test + public void errorComplete() throws Throwable { + Streamable.error(new TestException()) + .retryWhen((_, _) -> CompletableFuture.completedStage(false)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void errorAndFinishCrash() throws Throwable { + StreamableFailingFinish.MAIN_FAILS + .retryWhen((_, _) -> CompletableFuture.completedStage(false)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void timeoutWhen() throws Throwable { + var ts = Streamable.error(new TestException()) + .retryWhen((_, _) -> new CompletableFuture<>()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + Thread.sleep(100); + + ts.cancel(); + } + + @Test + public void retryPredicateFalse() { + Streamable.error(new TestException()) + .retry(_ -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void retryBiPredicateFalse() { + Streamable.error(new TestException()) + .retry((_, _) -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void retryBiPredicateTrueOnce() { + AtomicInteger counter = new AtomicInteger(); + Streamable.defer(() -> { + if (counter.getAndIncrement() == 0) { + return Streamable.error(new TestException()); + } + return Streamable.range(1, 5); + }) + .retry((c, _) -> c < 1) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void retryPredicateTrueOnce() { + AtomicInteger counter = new AtomicInteger(); + Streamable.defer(() -> { + if (counter.getAndIncrement() == 0) { + return Streamable.error(new TestException()); + } + return Streamable.range(1, 5); + }) + .retry(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index 1d7d0abce75..bbc0c0b575c 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -598,6 +598,8 @@ public void checkStreamable() { addOverride(new ParamOverride(Streamable.class, 0, ParamMode.ANY, "timer", Long.TYPE, TimeUnit.class, ExecutorService.class)); addOverride(new ParamOverride(Streamable.class, 0, ParamMode.ANY, "timeout", Long.TYPE, TimeUnit.class, Scheduler.class, Streamable.class)); addOverride(new ParamOverride(Streamable.class, 0, ParamMode.ANY, "delay", Long.TYPE, TimeUnit.class, Scheduler.class)); + addOverride(new ParamOverride(Streamable.class, 0, ParamMode.NON_NEGATIVE, "repeat", Long.TYPE)); + addOverride(new ParamOverride(Streamable.class, 0, ParamMode.NON_NEGATIVE, "retry", Long.TYPE)); // -----------------------------------------------------------------------------------