Added SMono.retryWhen(Retry) and deprecate all other SMono.retry and SMono.retryWhen

Author: sinwe

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

@@ -13,14 +13,14 @@ import reactor.core.{Disposable, Scannable => JScannable}
 import reactor.util.concurrent.Queues.SMALL_BUFFER_SIZE
 import reactor.util.context.Context
 import reactor.util.function.{Tuple2, Tuple3, Tuple4, Tuple5, Tuple6}
+import reactor.util.retry.Retry
 
 import scala.annotation.unchecked.uncheckedVariance
 import scala.jdk.CollectionConverters._
 import scala.concurrent.duration.Duration
 import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.reflect.ClassTag
 import scala.util.{Failure, Success, Try}
-
 import scala.language.implicitConversions
 
 
@@ -960,6 +960,7 @@ trait SMono[+T] extends SMonoLike[T] with MapablePublisher[T] with ScalaConverte
     * @return a re-subscribing [[SMono]] on onError up to the specified number of retries.
     *
     */
+  @deprecated("Use retryWhen(Retry)", since = "0.8.x")
   final def retry(numRetries: Long = Long.MaxValue, retryMatcher: Throwable => Boolean = (_: Throwable) => true): SMono[T] = coreMono.retry(numRetries, retryMatcher).asScala
 
   /**
@@ -976,13 +977,68 @@ trait SMono[+T] extends SMonoLike[T] with MapablePublisher[T] with ScalaConverte
     * @return a re-subscribing [[SMono]] on onError when the companion [[Publisher]] produces an
     *                                  onNext signal
     */
+  @deprecated("Use retryWhen(Retry)", since = "0.8.x")
   final def retryWhen(whenFactory: SFlux[Throwable] => Publisher[_]): SMono[T] = {
     val when = new Function[JFlux[Throwable], Publisher[_]] {
       override def apply(t: JFlux[Throwable]): Publisher[_] = whenFactory(new ReactiveSFlux[Throwable](t))
     }
     coreMono.retryWhen(when).asScala
   }
 
+  /**
+    * Retries this [[SMono]] in response to signals emitted by a companion [[Publisher]].
+    * The companion is generated by the provided [[Retry]] instance, see [[Retry.max(long)]], [[Retry.maxInARow(long)]]
+    * and [[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 [[SMono]] 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 [[Subscriber.onError(Throwable)]]
+    * will fail the resulting [[SMono]] with that same error.
+    * <p>
+    * <img class="marble" src="doc-files/marbles/retryWhenSpecForMono.svg" alt="">
+    * <p>
+    * Note that the [[Retry.RetrySignal]] state can be transient and change between each source
+    * [[org.reactivestreams.Subscriber.onError(Throwable) onError]] or
+    * [[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 [[Retry.RetrySignal.copy]] right away to mediate this.
+    * <p>
+    * Note that if the companion [[Publisher]] created by the <code>whenFactory</code>
+    * emits [[Context]] as trigger objects, these [[Context]] will be merged with
+    * the previous Context:
+    * <blockquote>
+    * <pre>
+    * <code>
+    * Retry customStrategy = SRetry.from(companion => companion.handle((retrySignal, sink) => {
+	 * 	    val ctx: Context = sink.currentContext();
+	 * 	    val rl: Int = 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()));
+    * }
+    * }));
+    * val retried: SMono[T] = originalMono.retryWhen(customStrategy);
+    * }
+    * </code>
+    * </pre>
+    * </blockquote>
+    *
+    * @param retrySpec the { @link Retry} strategy that will generate the companion [[Publisher]],
+    *                              given a [[SFlux]] that signals each onError as a { @link reactor.util.retry.Retry.RetrySignal}.
+    * @return a [[SMono]] that retries on onError when a companion [[Publisher]] produces an onNext signal
+    * @see Retry.max(long)
+    * @see Retry.maxInARow(long)
+    * @see Retry.backoff(long, Duration)
+    */
+  final def retryWhen(retrySpec: Retry): SMono[T] = coreMono.retryWhen(retrySpec).asScala
+
   /**
     * Expect exactly one item from this [[SMono]] source or signal
     * [[java.util.NoSuchElementException]] for an empty source.

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

@@ -22,6 +22,7 @@ import reactor.core.scheduler.{Scheduler, Schedulers}
 import reactor.test.scheduler.VirtualTimeScheduler
 import reactor.test.{StepVerifier, StepVerifierOptions}
 import reactor.util.context.Context
+import reactor.util.scala.retry.SRetry
 
 import scala.concurrent.Future
 import scala.concurrent.duration._
@@ -1126,6 +1127,12 @@ class SMonoTest extends AnyFreeSpec with Matchers with TestSupport with Idiomati
       counter.get() shouldBe 4
     }
 
+    ".retryWhen should retry according to the spec" in {
+      val sMono = SMono.error[Long](new RuntimeException("ex")).retryWhen(SRetry.from(_ => SMono.just(randomValue)))
+      StepVerifier.create(sMono)
+        .verifyComplete()
+    }
+
     ".single" - {
       "should enforce the existence of element" in {
         StepVerifier.create(just(randomValue).single())