Scheduled and repeat API + retry refactoring

core/src/main/scala/ox/resilience/RetryPolicy.scala

@@ -1,5 +1,7 @@
 package ox.resilience
 
+import ox.scheduling.{Jitter, Schedule}
+
 import scala.concurrent.duration.*
 
 /** A policy that defines how to retry a failed operation.
@@ -88,7 +90,7 @@ object RetryPolicy:
  maxDelay: FiniteDuration = 1.minute,
  jitter: Jitter = Jitter.None
  ): RetryPolicy[E, T] =
- RetryPolicy(Schedule.Backoff(maxRetries, initialDelay, maxDelay, jitter))
+ RetryPolicy(Schedule.Exponential(maxRetries, initialDelay, maxDelay, jitter))
 
  /** Creates a policy that retries indefinitely, with an increasing delay (backoff) between subsequent attempts, using a default
  * [[ResultPolicy]].
@@ -111,4 +113,4 @@ object RetryPolicy:
  maxDelay: FiniteDuration = 1.minute,
  jitter: Jitter = Jitter.None
  ): RetryPolicy[E, T] =
- RetryPolicy(Schedule.Backoff.forever(initialDelay, maxDelay, jitter))
+ RetryPolicy(Schedule.Exponential.forever(initialDelay, maxDelay, jitter))

core/src/main/scala/ox/resilience/retry.scala

@@ -1,9 +1,8 @@
 package ox.resilience
 
-import ox.{EitherMode, ErrorMode, sleep}
+import ox.{EitherMode, ErrorMode}
+import ox.scheduling.*
 
-import scala.annotation.tailrec
-import scala.concurrent.duration.*
 import scala.util.Try
 
 /** Retries an operation returning a direct result until it succeeds or the policy decides to stop.
@@ -48,33 +47,12 @@ def retryEither[E, T](policy: RetryPolicy[E, T])(operation: => Either[E, T]): Ei
  * - the error `E` in context `F` as returned by the last attempt if the policy decides to stop.
  */
 def retryWithErrorMode[E, F[_], T](em: ErrorMode[E, F])(policy: RetryPolicy[E, T])(operation: => F[T]): F[T] =
- @tailrec
- def loop(attempt: Int, remainingAttempts: Option[Int], lastDelay: Option[FiniteDuration]): F[T] =
- def sleepIfNeeded =
- val delay = policy.schedule.nextDelay(attempt, lastDelay)
- if delay.toMillis > 0 then sleep(delay)
- delay
-
- operation match
- case v if em.isError(v) =>
- val error = em.getError(v)
- policy.onRetry(attempt, Left(error))
-
- if policy.resultPolicy.isWorthRetrying(error) && remainingAttempts.forall(_ > 0) then
- val delay = sleepIfNeeded
- loop(attempt + 1, remainingAttempts.map(_ - 1), Some(delay))
- else v
- case v =>
- val result = em.getT(v)
- policy.onRetry(attempt, Right(result))
-
- if !policy.resultPolicy.isSuccess(result) && remainingAttempts.forall(_ > 0) then
- val delay = sleepIfNeeded
- loop(attempt + 1, remainingAttempts.map(_ - 1), Some(delay))
- else v
-
- val remainingAttempts = policy.schedule match
- case finiteSchedule: Schedule.Finite => Some(finiteSchedule.maxRetries)
- case _ => None
-
- loop(1, remainingAttempts, None)
+ runScheduledWithErrorMode(em)(
+ RunScheduledConfig(
+ policy.schedule,
+ policy.onRetry,
+ shouldContinueOnError = policy.resultPolicy.isWorthRetrying,
+ shouldContinueOnResult = t => !policy.resultPolicy.isSuccess(t),
+ delayPolicy = DelayPolicy.SinceTheEndOfTheLastInvocation
+ )
+ )(operation)

core/src/main/scala/ox/resilience/Jitter.scala → core/src/main/scala/ox/scheduling/Jitter.scala

@@ -1,4 +1,4 @@
-package ox.resilience
+package ox.scheduling
 
 /** A random factor used for calculating the delay between subsequent retries when a backoff strategy is used for calculating the delay.
  *

core/src/main/scala/ox/scheduling/RepeatConfig.scala

@@ -0,0 +1,30 @@
+package ox.scheduling
+
+import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration.DurationInt
+
+case class RepeatConfig[E, T](
+ schedule: Schedule,
+ shouldContinueOnError: E => Boolean = (_: E) => false,
+ shouldContinueOnResult: T => Boolean = (_: T) => true
+)
+
+object RepeatConfig:
+ def immediate[E, T](repeats: Int): RepeatConfig[E, T] = RepeatConfig(Schedule.Immediate(repeats))
+ def immediateForever[E, T]: RepeatConfig[E, T] = RepeatConfig(Schedule.Immediate.forever)
+
+ def fixedRate[E, T](repeats: Int, delay: FiniteDuration): RepeatConfig[E, T] = RepeatConfig(Schedule.Delay(repeats, delay))
+ def fixedRateForever[E, T](delay: FiniteDuration): RepeatConfig[E, T] = RepeatConfig(Schedule.Delay.forever(delay))
+
+ def exponential[E, T](
+ repeats: Int,
+ firstDelay: FiniteDuration,
+ maxDelay: FiniteDuration = 1.minute,
+ jitter: Jitter = Jitter.None
+ ): RepeatConfig[E, T] = RepeatConfig(Schedule.Exponential(repeats, firstDelay, maxDelay, jitter))
+
+ def exponentialForever[E, T](
+ firstDelay: FiniteDuration,
+ maxDelay: FiniteDuration = 1.minute,
+ jitter: Jitter = Jitter.None
+ ): RepeatConfig[E, T] = RepeatConfig(Schedule.Exponential.forever(firstDelay, maxDelay, jitter))