Step specifications

Use the name of a concurrent step.

Create a step directly in the join.

catchErrors: takes the collection of errors as parameter to process them. In the example the number of previous errors to process is 10.

.catchErrors(10)

catchExhaustedContext detect when a coroutine context has been exhausted. It takes the full exhausted step context as a parameter. Available parameters for.catchExhaustedContext include:

delay adds a constant delay before executing the next step, keeping the same execution pace as the previous step. In the example, the delay between steps is set to 123 MS.

.delay(Duration.ofMillis(123))

pace ensures that the next step is executed at an expected pace, whatever the pace of the input is. The calculation applies independently to each minion. In the following examples, the pace is set to 10 MS.

.pace { pastPeriodMs -> pastPeriodMs + 10 }

.pace(Duration.ofMillis(10))

acceleratingPace reduces the waiting interval for each iteration; to slow down the pace, use an accelerator lower than 1. In the example, the start pace is set to 20 MS, the acceleration is set to 2.0; the end pace is set to 4 MS.

.acceleratingPace(20, 2.0, 4)

constantPace applies a constant pace to execute the successor steps. The constant pace is set to 20 MS in the example.

.constantPace(20)

validate validates the input. If errors are returned, the step context is marked as exhausted and can then only be processed by error processing steps. In the example, steps with an input specification that is not an even integer return an error and are marked as exhausted.

.validate{ if(it % 2 != 0) listOf(StepError("The input should be even")) else emptyList() }

filter removes the step contexts having an input not matching the filter specification. In the example, steps with an input specification that is not an even integer are filtered out.

.filter(is even)

verify validates assertions rather than the input data quality/safety that is validated by validate. In the example, .verify confirms that that the batteryLevel property of the foundedBatteryState is exactly the same as the batteryLevel property of the savedBatteryState. If the two values are not the same, an error is returned and the step is marked as exhausted.

.verify { result ->
    val savedBatteryState = result.first
    val foundedBatteryState = result.second
    foundedBatteryState.batteryLevel shouldBeExactly savedBatteryState.batteryLevel
}

map converts each input element into another element in the output. In the example, .map multiplies each of its input values (it) by 3.

.map { it * 3 }

mapWithContext provides the same capability as map but with the step context as an additional argument. In the example, .mapWithContext returns the value of its input (it) and the input value times 3.

.mapWithContext{ context, input -> context.minionId + "-" + input }

flatMap converts a collective input into a single element in the output, with a user-defined strategy. In the example, flatmap converts a collection from the property members of the input into a flow. Each record of members will be individually provided to the next steps..

.flatMap { input -> flowOf(*input.members.toTypedArray()) }

flatten converts a collective input into a single element in the output for a known collective type (array, iterable, etc.), keeping each item unchanged.

.map { it * 3 }
.map { arrayOf(it, it + 100) }
.flatten()

onEach executes one or several statements on each element of the input and forwards them unchanged to the output. The example prints the input of the .onEach step to the console.

.onEach { input -> { println(input) }

shelve caches the provided values in a shared cache. In the example, .shelve creates a new sequence by transforming each element of the original sequence into a map of key-value pairs ("value-1" to it + 1).

.shelve { input -> mapOf("value-1" to input + 1) }

unshelve extracts a specific value from the cache using the keys passed as arguments. In the example, unshelve<Int, Double>("value-1") takes two type arguments <Int, Double>, which specify the types of the input of the step and the one of the value with key value-1, retrieved from the cache.

.unshelve<Int, Double>("value-1")

stage groups a linear graph of steps.

.stage("my-stage") {
    .map { it * 3 }
    .map { arrayOf(it, it + 100) }
    .flatten()
}.configure {
    iterate(5)
}