reactor-core/src/main/java/reactor/core/publisher/Flux.java
public final Flux<T> repeatWhen(Function<Flux<Long>, ? extends Publisher<?>> repeatFactory) {
		return onAssembly(new FluxRepeatWhen<>(this, repeatFactory));
	}
public final Flux<T> repeatWhen(Function<Flux<Long>, ? extends Publisher<?>> repeatFactory) {
		return onAssembly(new FluxRepeatWhen<>(this, repeatFactory));
	}
public final Flux < T > repeatWhen ( Function < Flux < Long > , ? extends Publisher < ? > > repeatFactory ) { return onAssembly ( new FluxRepeatWhen <> ( this , repeatFactory ) ) ; }
Repeatedly subscribe to this {@link Flux} when a companion sequence emits elements in
response to the flux completion signal. Any terminal signal from the companion
sequence will terminate the resulting {@link Flux} with the same signal immediately.
<p>If the companion sequence signals when this {@link Flux} is active, the repeat
attempt is suppressed.
<p>
<img class="marble" src="doc-files/marbles/repeatWhenForFlux.svg" alt="">
<p>
Note that if the companion {@link Publisher} created by the {@code repeatFactory}
emits {@link Context} as trigger objects, these {@link Context} will REPLACE the
operator's own Context. <strong>Please be careful there</strong>: replacing the
Context means that some keys you don't own could be removed, breaking libraries
that depend on them. As a result, the recommended approach is to always create such
a {@link Context} trigger by starting from the original Context (ensuring the trigger
contains all the keys from the original, unless you absolutely know you want to
remove one of these keys):
<pre><code>
.repeatWhen(emittedEachAttempt -> emittedEachAttempt
.flatMap(e -> Mono.subscriberContext().map(ctx -> Tuples.of(e, ctx)))
.flatMap(t2 -> {
long lastEmitted = t2.getT1();
Context ctx = t2.getT2();
int rl = ctx.getOrDefault("repeatsLeft", 0);
if (rl > 0) {
// /!\ THE ctx.put HERE IS THE ESSENTIAL PART /!\
return Mono.just(ctx.put("repeatsLeft", rl - 1)
.put("emitted", lastEmitted));
} else {
return Mono.<Context>error(new IllegalStateException("repeats exhausted"));
}
})
)
</code></pre>

@param repeatFactory the {@link Function} that returns the associated {@link Publisher}
companion, given a {@link Flux} that signals each onComplete as a {@link Long}
representing the number of source elements emitted in the latest attempt.

@return a {@link Flux} that repeats on onComplete when the companion {@link Publisher} produces an
onNext signal
Repeatedly subscribe to this { @link Flux } when a companion sequence emits elements in response to the flux completion signal . Any terminal signal from the companion sequence will terminate the resulting { @link Flux } with the same signal immediately . <p > If the companion sequence signals when this { @link Flux } is active the repeat attempt is suppressed . <p > <img class = marble src = doc - files / marbles / repeatWhenForFlux . svg alt = > <p > Note that if the companion { @link Publisher } created by the { @code repeatFactory } emits { @link Context } as trigger objects these { @link Context } will REPLACE the operator s own Context . <strong > Please be careful there< / strong > : replacing the Context means that some keys you don t own could be removed breaking libraries that depend on them . As a result the recommended approach is to always create such a { @link Context } trigger by starting from the original Context ( ensuring the trigger contains all the keys from the original unless you absolutely know you want to remove one of these keys ) : <pre > <code > . repeatWhen ( emittedEachAttempt - > emittedEachAttempt . flatMap ( e - > Mono . subscriberContext () . map ( ctx - > Tuples . of ( e ctx ))) . flatMap ( t2 - > { long lastEmitted = t2 . getT1 () ; Context ctx = t2 . getT2 () ; int rl = ctx . getOrDefault ( repeatsLeft 0 ) ; if ( rl > 0 ) { // / ! \ THE ctx . put HERE IS THE ESSENTIAL PART / ! \ return Mono . just ( ctx . put ( repeatsLeft rl - 1 ) . put ( emitted lastEmitted )) ; } else { return Mono . <Context > error ( new IllegalStateException ( repeats exhausted )) ; } } ) ) < / code > < / pre >