4.x: Streamable + repeat, repeatWhen, retry, retryWhen

src/main/java/io/reactivex/rxjava4/core/Streamable.java

@@ -865,10 +865,120 @@
         return RxJavaPlugins.onAssembly(new StreamableOnErrorResumeNext<>(this, fallbackMapper));
     }
 
+    /**
+     * Runs the upstream at most the given {@code count} times while it succeeds.
+     * <p>
+     * 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<T> 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.
+     * <p>
+     * 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<T> repeatWhen(Function<? super Long, ? extends CompletionStage<Boolean>> 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.
+     * <p>
+     * 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<T> 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<T> retry(Predicate<? super Throwable> 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.
+     * <p>
+     * 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<T> retry(BiPredicate<? super Long, ? super Throwable> 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.
+     * <p>
+     * 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<T> retryWhen(BiFunction<? super Long, ? super Throwable, ? extends CompletionStage<Boolean>> 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
      * @return the new {@Streamable} instance
      * @throws IllegalArgumentException if {@code count} is negative
      */
     @CheckReturnValue

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 {
     /**

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<T>(
+        Streamable<T> source,
+        Function<? super Long, ? extends CompletionStage<Boolean>> whenFunction
+)
+implements Streamable<T>, HasUpstreamStreamableSource<T> {
+
+    @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<T>
+    implements Streamer<T>, BiConsumer<Object, Throwable> {
+
+        final Streamable<T> source;
+
+        final StreamerCancellation downstreamCancellation;
+
+        final Function<? super Long, ? extends CompletionStage<Boolean>> whenFunction;
+
+        final AtomicInteger wipSource;
+
+        Streamer<T> currentStreamer;
+
+        CompletableFuture<Boolean> nextWaiter;
+
+        volatile int stage;
+
+        long completionCount;
+
+        Disposable whenFunctionCancel;
+
+        RepeatStreamer(Streamable<T> source, StreamerCancellation downstreamCancellation,
+                Function<? super Long, ? extends CompletionStage<Boolean>> 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<Boolean> 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<Void> finish() {
+            if (currentStreamer != null) {
+                return currentStreamer.finish();
+            }
+            return FINISHED;
+        }
+    }
+}