Proposal kotlin flow repeatUntil operator

[jeziellago]

trafficstars

Hello folks 👋🏼

I have a pooling use-case and I'd like some operator for repeat operations.

My use-case

// returns Processing, Completed, or Failure
fun getTransformationStatus(): Flow<Message> { ... }

sealed class Message {
    object Processing : Message()
    object Completed : Message()
    object Failed : Message()
}

The Flow<Message> should complete with Completed or Failure status, otherwise must repeat (with delay() and some attempts).

We have the retry and retryWhen operators, but I don't have exceptions propagation to control my flow.

Proposal operator repeatUntil

Inspired by retryWhen operator, I created the extension repeatUntil.

fun <T> Flow<T>.repeatUntil(predicate: suspend (value: T, attempt: Long) -> Boolean): Flow<T> =
    flow {
        var attempt = 0L
        var shallRepeat: Boolean
        do {
            shallRepeat = false
            val value = [email protected]()
            if (value != null) {
                if (!predicate(value, attempt)) {
                    shallRepeat = true
                    attempt++
                } else {
                    emit(value)
                }
            }
        } while (shallRepeat)
    }

Using repeatUntil in my use-case:

getTransformationStatus()
    .repeatUntil { value, attempt ->
        if (value is Message.Completed ||
            value is Message.Failed ||
            attempt == MAX_ATTEMPT
        ) {
            true
        } else {
            delay(POOLING_INTERVAL)
            false
        }
    }.collect {
        println(it)
    }

It works well for me, but I'd like to know if there is a better way to do that.