Flow.transformWhile operator

[elizarov]

Kotlin flow provides Flow<T>.takeWhile { ... } operator that emits values while the predicate returns true and then completes the flow without emitting the last value (where the predicate returned true). Its behavior, in this respect, is consistent with dropWhile operator.

In light of the design of shared flows (see #2034) that never complete by themselves there is a use-case where the matching value needs to be emitted, too. For example, consider a shared flow of download progress messages that ends with a message that download was done. We want to use this shared flow in a way that it completes when the download is done but also emits this last value because it also contains additional information. See also the similar request in #1816 (comment)

Naming challenges

Both RxJava and Project Reactor provide takeUntil operators for this use-case so that you can write takeUntil { it.isTheLastOne() }. However, the word "Until" has a completely different meaning in Koltin's core libraries. It is used to indicate a range up to, but excluding the last element. Moreover, Kotlin (the language) does not have any kind of repeat ... until predicate loop that can serve as inspiration. Kotlin consistently uses while for that purpose in both pre-condition while(predicate) { ... } and post-condition do { ... } while(predicate) forms. So takeUntil name is a no-go for Kotlin.

RxJS provides takeWhile(predicate, inclusive = true) operator for this use-case. However, we feel that accepting boolean here provides misplaced flexibility, since in all the uses it is likely going to be just constant and dedicated version of takeWhile operator would have worked better.

We cannot find a Kotliny name for this kind of use-case-specific operator.

Flow-style flexiblity

Flow-truncation operators (like takeWhile) are quite difficult to write correctly, so it would be wrong to leave the use-case of truncating a flow and emitting the last element to be unmet. However, we do have a similar problem in other places of Kotlin Flow design where some operators (like mapLatest) are too inflexible for some specific use-cass. We are aiming to minimize the number of operators in the library, and we want to provide both a flexible and easy-to-use operator. Hence we have transformXxx operators (like transformLatest) that accept a block of code that can make a decision of what and when to emit.

So, the proposal is to take the same road here and provide transformWhile operator:

fun <T, R> Flow<T>.transformWhile(
    transform: suspend FlowCollector<R>.(value: T) -> Boolean
): Flow<R>

The transformWhile operator generalizes takeWhile just like transform operator generalizes both filter/map and transformLatest generalizes mapLatest:

takeWhile(predicate) = 
    transformWhile { 
        if (predicate) { emit(it); true }
        else false
    }

Using this transformWhile operator one can complete a flow on the last value and emit it, too, by writing:

flow.transformWhile { emit(it); it.isNotTheLastOne() }