Schedule

sealed class Schedule<Input, Output>(source)

Retrying and repeating effects

A common demand when working with effects is to retry or repeat them when certain circumstances happen. Usually, the retrial or repetition does not happen right away; rather, it is done based on a policy. For instance, when fetching content from a network request, we may want to retry it when it fails, using an exponential backoff algorithm, for a maximum of 15 seconds or 5 attempts, whatever happens first.

Schedule allows you to define and compose powerful yet simple policies, which can be used to either repeat or retry computation.

The two core methods of running a schedule are:

  • retry: The effect is executed once, and if it fails, it will be reattempted based on the scheduling policy passed as an argument. It will stop if the effect ever succeeds, or the policy determines it should not be reattempted again.

  • repeat: The effect is executed once, and if it succeeds, it will be executed again based on the scheduling policy passed as an argument. It will stop if the effect ever fails, or the policy determines it should not be executed again. It will return the last internal state of the scheduling policy, or the error that happened running the effect.

Constructing a policy:

Constructing a simple schedule which recurs 10 times until it succeeds:

import arrow.fx.coroutines.*

fun <A> recurTenTimes() = Schedule.recurs<A>(10)

A more complex schedule

import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
import kotlin.time.ExperimentalTime
import arrow.fx.coroutines.*

@ExperimentalTime
fun <A> complexPolicy(): Schedule<A, List<A>> =
Schedule.exponential<A>(10.milliseconds).whileOutput { it < 60.seconds }
.andThen(Schedule.spaced<A>(60.seconds) and Schedule.recurs(100)).jittered()
.zipRight(Schedule.identity<A>().collect())

This policy will recur with exponential backoff as long as the delay is less than 60 seconds and then continue with a spaced delay of 60 seconds. The delay is also randomized slightly to avoid coordinated backoff from multiple services. Finally we also collect every input to the schedule and return it. When used with retry this will return a list of exceptions that occured on failed attempts.

Common use cases

Common use cases Once we have building blocks and ways to combine them, let’s see how we can use them to solve some use cases.

Repeating an effect and dealing with its result

When we repeat an effect, we do it as long as it keeps providing successful results and the scheduling policy tells us to keep recursing. But then, there is a question on what to do with the results provided by each iteration of the repetition.

There are at least 3 possible things we would like to do:

  • Discard all results; i.e., return Unit.

  • Discard all intermediate results and just keep the last produced result.

  • Keep all intermediate results.

Assuming we have a suspend effect in, and we want to repeat it 3 times after its first successful execution, we can do:

import arrow.fx.coroutines.*

suspend fun main(): Unit {
var counter = 0
//sampleStart
val res = Schedule.recurs<Unit>(3).repeat {
println("Run: ${counter++}")
}
//sampleEnd
println(res)
}

However, when running this new effect, its output will be the number of iterations it has performed, as stated in the documentation of the function. Also notice that we did not handle the error case, there are overloads repeatOrElse and repeatOrElseEither which offer that capability, repeat will just rethrow any error encountered.

If we want to discard the values provided by the repetition of the effect, we can combine our policy with Schedule.unit, using the zipLeft or zipRight combinators, which will keep just the output of one of the policies:

import arrow.fx.coroutines.*

suspend fun main(): Unit {
var counter = 0
//sampleStart
val res = (Schedule.unit<Unit>() zipLeft Schedule.recurs(3)).repeat {
println("Run: ${counter++}")
}
// equal to
val res2 = (Schedule.recurs<Unit>(3) zipRight Schedule.unit()).repeat {
println("Run: ${counter++}")
}
//sampleEnd
println(res)
println(res2)
}

Following the same strategy, we can zip it with the Schedule.identity policy to keep only the last provided result by the effect.

import arrow.fx.coroutines.*

suspend fun main(): Unit {
var counter = 0
//sampleStart
val res = (Schedule.identity<Int>() zipLeft Schedule.recurs(3)).repeat {
println("Run: ${counter++}"); counter
}
// equal to
val res2 = (Schedule.recurs<Int>(3) zipRight Schedule.identity<Int>()).repeat {
println("Run: ${counter++}"); counter
}
//sampleEnd
println(res)
println(res2)
}

Finally, if we want to keep all intermediate results, we can zip the policy with Schedule.collect:

import arrow.fx.coroutines.*

suspend fun main(): Unit {
var counter = 0
//sampleStart
val res = (Schedule.collect<Int>() zipLeft Schedule.recurs(3)).repeat {
println("Run: ${counter++}")
counter
}
// equal to
val res2 = (Schedule.recurs<Int>(3) zipRight Schedule.collect<Int>()).repeat {
println("Run: ${counter++}")
counter
}
//sampleEnd
println(res)
println(res2)
}

Repeating an effect until/while it produces a certain value

We can make use of the policies doWhile or doUntil to repeat an effect while or until its produced result matches a given predicate.

import arrow.fx.coroutines.*

suspend fun main(): Unit {
var counter = 0
//sampleStart
val res = Schedule.doWhile<Int>{ it <= 3 }.repeat {
println("Run: ${counter++}"); counter
}
//sampleEnd
println(res)
}

Exponential backoff retries

A common algorithm to retry effectful operations, as network requests, is the exponential backoff algorithm. There is a scheduling policy that implements this algorithm and can be used as:

import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.ExperimentalTime
import arrow.fx.coroutines.*

@ExperimentalTime
val exponential = Schedule.exponential<Unit>(250.milliseconds)

Types

Link copied to clipboard
object Companion
Link copied to clipboard
data class Decision<out A, out B>(val cont: Boolean, val delayInNanos: Double, val state: A, val finish: Eval<B>)

A single decision. Contains the decision to continue, the delay, the new state and the (lazy) result of a Schedule.

Functions

Link copied to clipboard
infix fun <A : Input, B> and(other: Schedule<A, B>): Schedule<A, Pair<Output, B>>

Combines two schedules. Continues only when both continue and chooses the maximum delay.

Link copied to clipboard
abstract infix fun <A : Input, B> andThen(other: Schedule<A, B>): Schedule<A, Either<Output, B>>

Executes one schedule after the other. When the first schedule ends, it continues with the second.

Link copied to clipboard
abstract fun <A : Input> check(pred: suspend (input: A, output: Output) -> Boolean): Schedule<A, Output>

Conditionally checks on both the input and the output whether or not to continue.

Link copied to clipboard
abstract infix fun <A, B> choose(other: Schedule<A, B>): Schedule<Either<Input, A>, Either<Output, B>>

Combines two schedules with different input and output and conditionally choose between the two. Continues when the chosen schedule continues and uses the chosen schedules delay.

Link copied to clipboard

Accumulates the results of every execution into a list.

Link copied to clipboard
fun <A : Input, B, C> combine(other: Schedule<A, B>, zipContinue: (cont: Boolean, otherCont: Boolean) -> Boolean, zipDuration: (duration: Duration, otherDuration: Duration) -> Duration, zip: (Output, B) -> C): Schedule<A, C>

Combines with another schedule by combining the result and the delay of the Decision with the zipContinue, zipDuration and a zip functions

Link copied to clipboard
abstract fun <A : Input, B, C> combineNanos(other: Schedule<A, B>, zipContinue: (cont: Boolean, otherCont: Boolean) -> Boolean, zipDuration: (duration: Double, otherDuration: Double) -> Double, zip: (Output, B) -> C): Schedule<A, C>

Combines with another schedule by combining the result and the delay of the Decision with the functions zipContinue, zipDuration and a zip function

Link copied to clipboard
infix fun <B> compose(other: Schedule<B, Input>): Schedule<B, Output>

Infix variant of pipe with reversed order.

Link copied to clipboard
fun <B> const(b: B): Schedule<Input, B>

Changes the result of a Schedule to always be b.

Link copied to clipboard
abstract fun <B> contramap(f: suspend (B) -> Input): Schedule<B, Output>

Changes the input of the schedule. May alter a schedule's decision if it is based on input.

Link copied to clipboard
fun delay(f: suspend (duration: Duration) -> Duration): Schedule<Input, Output>
Link copied to clipboard
fun delayedNanos(f: suspend (duration: Double) -> Double): Schedule<Input, Output>
Link copied to clipboard
fun <B, C> dimap(f: suspend (B) -> Input, g: (Output) -> C): Schedule<B, C>
Link copied to clipboard
fun <C> fold(initial: C, f: suspend (acc: C, output: Output) -> C): Schedule<Input, C>

Non-effectful version of foldM.

Link copied to clipboard
abstract fun <C> foldLazy(initial: suspend () -> C, f: suspend (acc: C, output: Output) -> C): Schedule<Input, C>

Accumulates the results of a schedule by folding over them effectfully.

Link copied to clipboard
abstract fun forever(): Schedule<Input, Output>

Always retries a schedule regardless of the decision made prior to invoking this method.

Link copied to clipboard
fun jittered(genRand: suspend () -> Double): Schedule<Input, Output>
@JvmName(name = "jitteredDuration")
fun jittered(genRand: suspend () -> Duration): Schedule<Input, Output>

fun jittered(random: Random = Random.Default): Schedule<Input, Output>

Add random jitter to a schedule.

Link copied to clipboard
abstract fun logInput(f: suspend (input: Input) -> Unit): Schedule<Input, Output>

Runs an effectful handler on every input. Does not alter the decision.

Link copied to clipboard
abstract fun logOutput(f: suspend (output: Output) -> Unit): Schedule<Input, Output>

Runs an effectful handler on every output. Does not alter the decision.

Link copied to clipboard
abstract fun <B> map(f: (output: Output) -> B): Schedule<Input, B>

Changes the output of a schedule. Does not alter the decision of the schedule.

Link copied to clipboard

Changes the delay of a resulting Decision based on the Output and the produced delay.

Link copied to clipboard
abstract fun modifyNanos(f: suspend (Output, Double) -> Double): Schedule<Input, Output>
Link copied to clipboard
abstract operator fun not(): Schedule<Input, Output>

Inverts the decision of a schedule.

Link copied to clipboard
infix fun <A : Input, B> or(other: Schedule<A, B>): Schedule<A, Pair<Output, B>>

Combines two schedules. Continues if one continues and chooses the minimum delay.

Link copied to clipboard
abstract infix fun <B> pipe(other: Schedule<Output, B>): Schedule<Input, B>

Composes this schedule with the other schedule by piping the output of this schedule into the input of the other.

Link copied to clipboard
suspend fun repeat(fa: suspend () -> Input): Output

Runs this effect once and, if it succeeded, decide using the provided policy if the effect should be repeated and if so, with how much delay. Returns the last output from the policy or raises an error if a repeat failed.

Link copied to clipboard
suspend fun repeatAsFlow(fa: suspend () -> Input): Flow<Output>

Runs this effect and emits the output, if it succeeded, decide using the provided policy if the effect should be repeated and emitted, if so, with how much delay. This will raise an error if a repeat failed.

Link copied to clipboard
suspend fun repeatOrElse(fa: suspend () -> Input, orElse: suspend (Throwable, Output?) -> Output): Output

Runs this effect once and, if it succeeded, decide using the provided policy if the effect should be repeated and if so, with how much delay. Also offers a function to handle errors if they are encountered during repetition.

Link copied to clipboard
suspend fun repeatOrElseAsFlow(fa: suspend () -> Input, orElse: suspend (Throwable, Output?) -> Output): Flow<Output>

Runs this effect and emits the output, if it succeeded, decide using the provided policy if the effect should be repeated and emitted, if so, with how much delay. Also offers a function to handle errors if they are encountered during repetition.

Link copied to clipboard
abstract suspend fun <C> repeatOrElseEither(fa: suspend () -> Input, orElse: suspend (Throwable, Output?) -> C): Either<C, Output>
Link copied to clipboard
abstract suspend fun <C> repeatOrElseEitherAsFlow(fa: suspend () -> Input, orElse: suspend (Throwable, Output?) -> C): Flow<Either<C, Output>>
Link copied to clipboard
fun <A : Input> untilInput(f: suspend (A) -> Boolean): Schedule<A, Output>

untilInput(f) = whileInput(f).not()

Link copied to clipboard
fun untilOutput(f: suspend (Output) -> Boolean): Schedule<Input, Output>

untilOutput(f) = whileOutput(f).not()

Link copied to clipboard
Link copied to clipboard
fun <A : Input> whileInput(f: suspend (A) -> Boolean): Schedule<A, Output>

Continues or stops the schedule based on the input.

Link copied to clipboard
fun whileOutput(f: suspend (Output) -> Boolean): Schedule<Input, Output>

Continues or stops the schedule based on the output.

Link copied to clipboard
infix fun <A, B> zip(other: Schedule<A, B>): Schedule<Pair<Input, A>, Pair<Output, B>>
abstract fun <A, B, C> zip(other: Schedule<A, B>, f: (Output, B) -> C): Schedule<Pair<Input, A>, C>

Combines two with different input and output using and. Continues when both continue and uses the maximum delay.

Link copied to clipboard
infix fun <A : Input, B> zipLeft(other: Schedule<A, B>): Schedule<A, Output>

Combines two schedules with and but throws away the right schedule's result.

Link copied to clipboard
infix fun <A : Input, B> zipRight(other: Schedule<A, B>): Schedule<A, B>

Combines two schedules with and but throws away the left schedule's result.

Extensions

Link copied to clipboard
suspend fun <A, B> Schedule<Throwable, B>.retry(fa: suspend () -> A): A

Runs an effect and, if it fails, decide using the provided policy if the effect should be retried and if so, with how much delay. Returns the result of the effect if if it was successful or re-raises the last error encountered when the schedule ends.

Link copied to clipboard
suspend fun <A, B> Schedule<Throwable, B>.retryOrElse(fa: suspend () -> A, orElse: suspend (Throwable, B) -> A): A

Runs an effect and, if it fails, decide using the provided policy if the effect should be retried and if so, with how much delay. Also offers a function to handle errors if they are encountered during retrial.

Link copied to clipboard
suspend fun <A, B, C> Schedule<Throwable, B>.retryOrElseEither(fa: suspend () -> A, orElse: suspend (Throwable, B) -> C): Either<C, A>

Runs an effect and, if it fails, decide using the provided policy if the effect should be retried and if so, with how much delay. Also offers a function to handle errors if they are encountered during retrial.