Deprecate old SFlux.retryWhen(predicate)

Added SFlux.retryWhen(Retry)

Author: sinwe

src/main/scala/reactor/core/scala/publisher/SFlux.scala

@@ -17,6 +17,7 @@ import reactor.util.Logger
 import reactor.util.concurrent.Queues
 import reactor.util.concurrent.Queues.{SMALL_BUFFER_SIZE, XS_BUFFER_SIZE}
 import reactor.util.function.{Tuple2, Tuple3, Tuple4, Tuple5, Tuple6}
+import reactor.util.retry.Retry
 
 import scala.collection.mutable
 import scala.concurrent.duration.Duration
@@ -504,7 +505,20 @@ trait SFlux[T] extends SFluxLike[T, SFlux] with MapablePublisher[T] with ScalaCo
     new ReactiveSFlux[SGroupedFlux[K, V]](jFluxOfGroupedFlux.map((jg: JGroupedFlux[K, V]) => SGroupedFlux(jg)))
   }
 
-  final def handle[R](handler: (T, SynchronousSink[R]) => Unit): SFlux[R] = SFlux.fromPublisher(coreFlux.handle[R](handler))
+  /**
+    * Handle the items emitted by this [[SFlux]] by calling a biconsumer with the
+    * output sink for each onNext. At most one [[SynchronousSink.next(Anyref)]]
+    * call must be performed and/or 0 or 1 [[SynchronousSink.error(Throwable)]] or
+    * [[SynchronousSink.complete()]].
+    *
+    * @param handler the handling [[Function2]]
+    * @tparam R the transformed type
+    * @reactor.errorMode This operator supports { @link #onErrorContinue(BiConsumer) resuming on errors} (including when
+    *                                                   fusion is enabled) when the { @link BiConsumer} throws an exception or if an error is signaled explicitly via
+    *                                                   { @link SynchronousSink#error(Throwable)}.
+    * @return a transformed [[SFlux]]
+    */
+  final def handle[R](handler: (T, SynchronousSink[R]) => Unit): SFlux[R] = coreFlux.handle[R](handler).asScala
 
   final def hasElement(value: T): SMono[Boolean] = new ReactiveSMono[JBoolean](coreFlux.hasElement(value)).map(Boolean2boolean)
 
@@ -596,13 +610,65 @@ trait SFlux[T] extends SFluxLike[T, SFlux] with MapablePublisher[T] with ScalaCo
 
   final def retry(numRetries: Long = Long.MaxValue, retryMatcher: Throwable => Boolean = (_: Throwable) => true): SFlux[T] = coreFlux.retry(numRetries, retryMatcher).asScala
 
+  @deprecated("Use retryWhen(Retry)")
   final def retryWhen(whenFactory: SFlux[Throwable] => Publisher[_]): SFlux[T] = {
     val func = new Function[JFlux[Throwable], Publisher[_]] {
       override def apply(t: JFlux[Throwable]): Publisher[_] = whenFactory(new ReactiveSFlux[Throwable](t))
     }
     coreFlux.retryWhen(func).asScala
   }
 
+  /**
+    * Retries this [[SFlux]] in response to signals emitted by a companion [[Publisher]].
+    * The companion is generated by the provided [[Retry]] instance, see {@link Retry#max(long)}, {@link Retry#maxInARow(long)}
+    * and {@link Retry#backoff(long, Duration)} for readily available strategy builders.
+    * <p>
+    * The operator generates a base for the companion, a [[SFlux]] of [[reactor.util.retry.Retry.RetrySignal]]
+    * which each give metadata about each retryable failure whenever this [[SFlux]] signals an error. The final companion
+    * should be derived from that base companion and emit data in response to incoming onNext (although it can emit less
+    * elements, or delay the emissions).
+    * <p>
+    * Terminal signals in the companion terminate the sequence with the same signal, so emitting an {@link Subscriber#onError(Throwable)}
+    * will fail the resulting [[SFlux]] with that same error.
+    * <p>
+    * <img class="marble" src="doc-files/marbles/retryWhenSpecForFlux.svg" alt="">
+    * <p>
+    * Note that the [[Retry.RetrySignal]] state can be transient and change between each source
+    * {@link org.reactivestreams.Subscriber#onError(Throwable) onError} or
+    * {@link org.reactivestreams.Subscriber#onNext(Object) onNext}. If processed with a delay,
+    * this could lead to the represented state being out of sync with the state at which the retry
+    * was evaluated. Map it to {@link Retry.RetrySignal#copy()} right away to mediate this.
+    * <p>
+    * Note that if the companion [[Publisher]] created by the {@code whenFactory}
+    * emits [[reactor.util.context.Context]] as trigger objects, these [[reactor.util.context.Context]] will be merged with
+    * the previous Context:
+    * <blockquote><pre>
+    * {@code
+    * Retry customStrategy = Retry.fromFunction(companion -> companion.handle((retrySignal, sink) -> {
+	 * 	    Context ctx = sink.currentContext();
+	 * 	    int rl = ctx.getOrDefault("retriesLeft", 0);
+	 * 	    if (rl > 0) {
+	 *		    sink.next(Context.of(
+	 *		        "retriesLeft", rl - 1,
+	 *		        "lastError", retrySignal.failure()
+	 *		    ));
+	 * 	    } else {
+    * 	        sink.error(Exceptions.retryExhausted("retries exhausted", retrySignal.failure()));
+    * }
+    * }));
+    * Flux<T> retried = originalFlux.retryWhen(customStrategy);
+    * }</pre>
+    * </blockquote>
+    *
+    * @param retrySpec the { @link Retry} strategy that will generate the companion { @link Publisher},
+    *                              given a { @link Flux} that signals each onError as a { @link reactor.util.retry.Retry.RetrySignal}.
+    * @return a { @link Flux} that retries on onError when a companion { @link Publisher} produces an onNext signal
+    * @see Retry.max(long)
+    * @see Retry.maxInARow(long)
+    * @see Retry.backoff(long, Duration)
+    */
+  final def retryWhen(retry: Retry): SFlux[T] = coreFlux.retryWhen(retry).asScala
+
   final def sample(timespan: Duration): SFlux[T] = coreFlux.sample(timespan).asScala
 
   final def sampleFirst(timespan: Duration): SFlux[T] = coreFlux.sampleFirst(timespan).asScala
@@ -863,7 +929,6 @@ object SFlux {
     * @tparam T the produced type
     * @return a [[reactor.core.publisher.FluxProcessor]] accepting publishers and producing T
     */
-  //  TODO: How to test these switchOnNext?
   def switchOnNext[T](mergedPublishers: Publisher[_ <: Publisher[_ <: T]]): SFlux[T] = SFlux.fromPublisher(JFlux.switchOnNext[T](mergedPublishers))
 
   def using[T, D](resourceSupplier: () => D, sourceSupplier: D => Publisher[_ <: T], resourceCleanup: D => Unit, eager: Boolean = false): SFlux[T] =

src/main/scala/reactor/util/scala/retry/SRetry.scala

@@ -0,0 +1,16 @@
+package reactor.util.scala.retry
+
+import org.reactivestreams.Publisher
+import reactor.core.scala.publisher.SFlux
+import reactor.util.retry.Retry
+import reactor.util.retry.Retry.RetrySignal
+
+object SRetry {
+  /**
+    * A wrapper around [[Function1]] to provide [[Retry]] by using lambda expressions.
+    *
+    * @param function the { @link Function} representing the desired { @link Retry} strategy as a lambda
+    * @return the { @link Retry} strategy adapted from the { @link Function}
+    */
+  final def from(function: SFlux[RetrySignal] => Publisher[_]): Retry = Retry.from(function)
+}

src/test/scala/reactor/core/scala/publisher/SFluxTest.scala

@@ -21,6 +21,7 @@ import reactor.core.scheduler.Schedulers
 import reactor.test.StepVerifier
 import reactor.test.scheduler.VirtualTimeScheduler
 import reactor.util.concurrent.Queues
+import reactor.util.scala.retry.SRetry
 
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
@@ -1770,6 +1771,8 @@ class SFluxTest extends AnyFreeSpec with Matchers with TableDrivenPropertyChecks
       }
     }
 
+/*
+//this is deprecated, should be removed on 0.7.x
     ".retryWhen should retry the companion publisher produces onNext signal" in {
       val counter = new AtomicInteger(0)
       val flux = SFlux.just(1, 2, 3).concatWith(SMono.raiseError(new RuntimeException("ex"))).retryWhen { _ =>
@@ -1781,6 +1784,20 @@ class SFluxTest extends AnyFreeSpec with Matchers with TableDrivenPropertyChecks
         .expectNext(1, 2, 3)
         .verifyComplete()
     }
+*/
+
+    ".retryWhen should retry according to the spec" in {
+      val counter = new AtomicInteger(0)
+      val sFlux = SFlux.just(1, 2, 3).concatWith(SMono.raiseError(new RuntimeException("ex")))
+        .retryWhen(SRetry.from(_ => {
+          if (counter.getAndIncrement() > 0) SMono.raiseError[Int](new RuntimeException("another ex"))
+          else SMono.just(1)
+        }))
+      StepVerifier.create(sFlux)
+        .expectNext(1, 2, 3)
+        .expectNext(1, 2, 3)
+        .verifyComplete()
+    }
 
     ".sample should emit the last value for given interval" in {
       StepVerifier.withVirtualTime(() => SFlux.just(1, 2, 3, 4, 5, 6).delayElements(1 second).sample(1500 milliseconds))