Polish #4027

Merged RepeatSpec with Repeat to avoid exposing package private
RepeatSignal. Also, removed explicit contextWrite which is unnecessary
when there is no onErrorStop used like with Retry companion.

Fixes #3545

Signed-off-by: Dariusz Jędrzejczyk <[email protected]>

Author: chemicL

reactor-core/src/main/java/reactor/util/repeat/ImmutableRepeatSignal.java

@@ -21,12 +21,12 @@
 import java.time.Duration;
 
 /**
- * An immutable {@link reactor.util.repeat.Repeat.RepeatSignal} that can be used for
+ * An immutable {@link reactor.util.repeat.RepeatSpec.RepeatSignal} that can be used for
  * retained copies of mutable implementations.
  *
  * @author Daeho Kwon
  */
-final class ImmutableRepeatSignal implements Repeat.RepeatSignal {
+final class ImmutableRepeatSignal implements RepeatSpec.RepeatSignal {
 
 	private final long        iteration;
 	private final Long        companionValue;
@@ -64,7 +64,7 @@ public ContextView repeatContextView() {
 	}
 
 	@Override
-	public Repeat.RepeatSignal copy() {
+	public RepeatSpec.RepeatSignal copy() {
 		return this;
 	}
 

reactor-core/src/main/java/reactor/util/repeat/Repeat.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Predicate;
 
 import org.reactivestreams.Publisher;
@@ -46,17 +47,22 @@
  *   <li>Context propagation via {@link #withRepeatContext(ContextView)}</li>
  * </ul>
  * This strategy does not retry based on error signals, but rather repeats a successful sequence (until termination condition is met).
+ * <p>
+ * The companion {@code Flux<Long>} represents repeat signals, and each value corresponds to a repeat attempt.
+ * This class transforms the repeat signals into a {@link Publisher} that determines whether to trigger a repeat.
  *
  * @author Daeho Kwon
  */
-public final class RepeatSpec extends Repeat {
+public final class RepeatSpec implements Function<Flux<Long>, Publisher<?>> {
 
 	private static final Predicate<RepeatSignal> ALWAYS_TRUE    = __ -> true;
 	private static final Consumer<RepeatSignal>  NO_OP_CONSUMER = rs -> {
 	};
 	private static final RepeatSignal            TERMINATE      =
 			new ImmutableRepeatSignal(-1, -1L, Duration.ZERO, Context.empty());
 
+	final ContextView repeatContext;
+
 	final long                    maxRepeats;
 	final Predicate<RepeatSignal> repeatPredicate;
 	final Consumer<RepeatSignal>  beforeRepeatHook;
@@ -76,7 +82,7 @@ private RepeatSpec(ContextView repeatContext,
 			Duration fixedDelay,
 			Scheduler scheduler,
 			double jitterFactor) {
-		super(repeatContext);
+		this.repeatContext = repeatContext;
 		this.maxRepeats = maxRepeats;
 		this.repeatPredicate = repeatPredicate;
 		this.beforeRepeatHook = beforeRepeatHook;
@@ -135,12 +141,12 @@ public static RepeatSpec create(Predicate<RepeatSignal> predicate, long n) {
 	}
 
 	/**
-	 * Set the user provided {@link Repeat#repeatContext() context} that can be used to
+	 * Set the user provided {@link ContextView contextView} that can be used to
 	 * manipulate state on retries.
 	 *
 	 * @param repeatContext a new snapshot of user provided data
-	 * @return a new copy of the {@link RepeatSpec} which can either be further configured
-	 * or used as {@link Repeat}
+	 * @return a new copy of the {@link RepeatSpec} which can either be further
+	 * configured.
 	 */
 	public RepeatSpec withRepeatContext(ContextView repeatContext) {
 		return new RepeatSpec(repeatContext,
@@ -260,44 +266,50 @@ public RepeatSpec withScheduler(Scheduler scheduler) {
 	}
 
 	@Override
+	public Publisher<?> apply(Flux<Long> signals) {
+		return generateCompanion(signals);
+	}
+
+	/**
+	 * Generates the companion publisher responsible for reacting to incoming repeat signals,
+	 * effectively deciding whether to trigger another repeat cycle.
+	 *
+	 * @param signals the incoming repeat signals, where each {@link Long} value indicates the iteration index
+	 * @return the companion publisher that determines if a repeat should occur
+	 */
 	public Publisher<Long> generateCompanion(Flux<Long> signals) {
-		return Flux.deferContextual(cv -> signals.index()
-		                                         .contextWrite(cv)
-		                                         .map(tuple -> {
-			                                         long iter = tuple.getT1();
-			                                         long companionValue = tuple.getT2();
-
-			                                         RepeatSignal signal =
-					                                         new ImmutableRepeatSignal(
-							                                         iter,
-							                                         companionValue,
-							                                         calculateDelay(),
-							                                         repeatContext());
-
-			                                         if (iter >= maxRepeats || !repeatPredicate.test(
-					                                         signal)) {
-				                                         return TERMINATE;
-			                                         }
-			                                         return signal;
-		                                         })
-		                                         .takeWhile(signal -> signal != TERMINATE)
-		                                         .concatMap(signal -> {
-			                                         try {
-				                                         beforeRepeatHook.accept(signal);
-			                                         }
-			                                         catch (Throwable e) {
-				                                         return Mono.error(e);
-			                                         }
-
-			                                         Duration delay = signal.backoff();
-			                                         Mono<Long> trigger = delay.isZero() ?
-					                                         Mono.just(signal.companionValue()) :
-					                                         Mono.delay(delay, scheduler)
-					                                             .thenReturn(signal.companionValue());
-
-			                                         return trigger.doOnSuccess(v -> afterRepeatHook.accept(
-					                                         signal));
-		                                         }));
+		return signals.index()
+		              .map(tuple -> {
+			              long iter = tuple.getT1();
+			              long companionValue = tuple.getT2();
+
+			              RepeatSignal signal = new ImmutableRepeatSignal(iter,
+					              companionValue,
+					              calculateDelay(),
+					              this.repeatContext);
+
+			              if (iter >= maxRepeats || !repeatPredicate.test(signal)) {
+				              return TERMINATE;
+			              }
+			              return signal;
+		              })
+		              .takeWhile(signal -> signal != TERMINATE)
+		              .concatMap(signal -> {
+			              try {
+				              beforeRepeatHook.accept(signal);
+			              }
+			              catch (Throwable e) {
+				              return Mono.error(e);
+			              }
+
+			              Duration delay = signal.backoff();
+			              Mono<Long> trigger =
+					              delay.isZero() ? Mono.just(signal.companionValue()) :
+							              Mono.delay(delay, scheduler)
+							                  .thenReturn(signal.companionValue());
+
+			              return trigger.doOnSuccess(v -> afterRepeatHook.accept(signal));
+		              });
 	}
 
 	Duration calculateDelay() {
@@ -311,4 +323,45 @@ Duration calculateDelay() {
 		}
 		return actual;
 	}
+
+	/**
+	 * State information associated with each repeat signal, used in repeat strategies.
+	 */
+	public interface RepeatSignal {
+
+		/**
+		 * Returns the current iteration count, starting from 0.
+		 *
+		 * @return the iteration index
+		 */
+		long iteration();
+
+		/**
+		 * Returns the value from the companion publisher that triggered this repeat signal.
+		 *
+		 * @return the companion value
+		 */
+		Long companionValue();
+
+		/**
+		 * Returns the delay before the next repeat attempt.
+		 *
+		 * @return the backoff duration
+		 */
+		Duration backoff();
+
+		/**
+		 * Returns the read-only context associated with this repeat signal.
+		 *
+		 * @return the repeat context view
+		 */
+		ContextView repeatContextView();
+
+		/**
+		 * Returns an immutable copy of this {@link RepeatSignal}, capturing the current state.
+		 *
+		 * @return an immutable copy of the signal
+		 */
+		RepeatSignal copy();
+	}
 }

reactor-core/src/main/java/reactor/util/repeat/RepeatSpec.java

@@ -17,6 +17,7 @@
 package reactor.util.repeat;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.Test;
@@ -50,10 +51,10 @@ void onlyIfReplacesPredicate() {
 		                                  .onlyIf(signal -> signal.iteration() < 5)
 		                                  .onlyIf(signal -> signal.iteration() == 0);
 
-		Repeat.RepeatSignal acceptSignal =
+		RepeatSpec.RepeatSignal acceptSignal =
 				new ImmutableRepeatSignal(0, 123L, Duration.ofMillis(0), Context.empty());
 
-		Repeat.RepeatSignal rejectSignal =
+		RepeatSpec.RepeatSignal rejectSignal =
 				new ImmutableRepeatSignal(4, 123L, Duration.ofMillis(0), Context.empty());
 
 		assertThat(repeatSpec.repeatPredicate).accepts(acceptSignal)
@@ -64,7 +65,7 @@ void onlyIfReplacesPredicate() {
 	void doBeforeRepeatIsCumulative() {
 		AtomicInteger counter = new AtomicInteger();
 
-		Repeat.RepeatSignal dummySignal =
+		RepeatSpec.RepeatSignal dummySignal =
 				new ImmutableRepeatSignal(0, 0L, Duration.ZERO, Context.empty());
 
 		RepeatSpec repeatSpec = RepeatSpec.times(1)
@@ -80,7 +81,7 @@ void doBeforeRepeatIsCumulative() {
 	void doAfterRepeatIsCumulative() {
 		AtomicInteger counter = new AtomicInteger();
 
-		Repeat.RepeatSignal dummySignal =
+		RepeatSpec.RepeatSignal dummySignal =
 				new ImmutableRepeatSignal(0, 0L, Duration.ZERO, Context.empty());
 
 		RepeatSpec repeatSpec = RepeatSpec.times(1)
@@ -170,16 +171,16 @@ void jitterProducesDelayInExpectedRange() {
 
 	@Test
 	void repeatContextCanInfluencePredicate() {
-		RepeatSpec repeatSpec = RepeatSpec.times(5)
-		                                  .withRepeatContext(Context.of("stopAfterOne",
-				                                  true))
-		                                  .onlyIf(signal -> {
-			                                  boolean stopAfterOne =
-					                                  signal.repeatContextView()
-					                                        .getOrDefault("stopAfterOne",
-							                                        false);
-			                                  return !stopAfterOne || signal.iteration() == 0;
-		                                  });
+		RepeatSpec repeatSpec =
+				RepeatSpec.times(5)
+				          .withRepeatContext(Context.of("stopAfterOne", true))
+				          .onlyIf(signal -> {
+							  Boolean contextEntry =
+									  signal.repeatContextView()
+									        .getOrDefault("stopAfterOne", Boolean.FALSE);
+							  boolean stopAfterOne = Boolean.TRUE.equals(contextEntry);
+							  return !stopAfterOne || signal.iteration() == 0;
+						  });
 
 		AtomicInteger subscriptionCount = new AtomicInteger();
 		Flux<String> source = Flux.defer(() -> {