Proposal: Flow.collectLatest

[BoxResin]

Function signature

suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit): Unit

Behavior

val flow = flowOf(1, 2, 3, 4, 5)
flow.collectLatest { value: Int -> // This lambda would be canceled if a new value is emitted although it hasn't been finished.
    delay(100)
    println(value) // So only '5' will be printed.
}

[BoxResin]

Possible Implementation I think:

suspend inline fun <T> Flow<T>.collectLatest(crossinline action: suspend (value: T) -> Unit) {
    var job: Job? = null
    val scope = CoroutineScope(coroutineContext)

    this.collect { value: T ->
        job?.cancelAndJoin()
        job = scope.launch { action(value) }
    }
    job?.join()
}

[LouisCAD]

I have used such a thing for a long time (also have a version for channels), with a longer name, and the ability to skip equals:
https://github.com/LouisCAD/Splitties/blob/develop/samples/android-app/src/androidMain/kotlin/com/example/splitties/extensions/coroutines/Flow.kt#L18-L35

I need it a lot of cases, so I'd vote for it coming here.

[elizarov]

What's the use-case? I get the example of usage, but where would you actually use it?

[BoxResin]

@LouisCAD
The ability to skip equals may be replaced by Flow.distinctUntilChanged().
In combination with it, you can do the same thing.

For example:

val flow = flowOf(1, 2, 2, 3).delayEach(100)
flow.distinctUntilChanged().collectLatest {
    // Do something
}

[LouisCAD]

@BoxResin Yes, but it's a little less efficient, and may force an additional indentation level of the lambda.

[LouisCAD]

@elizarov Here's an example: https://github.com/LouisCAD/Splitties/blob/develop/samples/android-app/src/androidMain/kotlin/com/example/splitties/main/MainActivity.kt#L68-L90

Can also be helpful if you want to have a loop running while in a certain state that is emitted from the flow.

[BoxResin]

@elizarov Here's another example:

object Config {
    val school: Flow<School?> = ...
}

class DailyMealViewModel(val date: TimePoint) : CoroutineScope {
    private val job = SupervisorJob()
    override val context = Dispatchers.Main + job

    val content = MutableLiveData<DailyMealViewContent>()

    init {
        // `this.launch { ... }` would be canceled when `this.onCleared()` is called.
        this.launch {
            Config.school.collectLatest { school: School? ->
                // This function will be canceled when a new school emitted.
                this.onSchoolChanged(school)
            }
        }
    }

    private suspend fun onSchoolChanged(school: School?) {
        if (school == null) {
            this.content.value = DailyMealViewContent.Error(
                message = "Please select school",
                button = "Select" to this::onClickSchoolSetting
            )
            return
        }

        MealDatabase.observe(school.code, this.date).collectLatest { dailyMeal: DailyMeal? ->
            // This function will be canceled when the `dailyMeal` in database changed.
            this.onDailyMealChanged(dailyMeal)
        }
    }

    private suspend fun onDailyMealChanged(dailyMeal: DailyMeal?) {
        if (dailyMeal == null) {
            this.content.value = DailyMealViewContent.Error(
                message = "Please download meal data",
                button = "Download" to this::onClickDownloadMeal
            )
            return
        }
        
        // Something to deal with `dailyMeal`
        ...
    }

    override fun onCleared() {
        this.job.cancel()
    }
}

[elizarov]

We've recently added .conflate() operator and it seems that collectLatest { ... } is almost equivalent to conflate().collect { ... }. See the following example in playground: https://pl.kotl.in/JcBskQvFE

There:

    val flow = flowOf(1, 2, 3, 4, 5)
    flow.conflate().collect { value: Int -> 
        delay(100)
        println(value)
    }    

prints 1 and 5. It does not rely on the cancellation of the collecting lambda though. Still, might it cover some of the use-cases?

[elizarov]

Another observise, the requested behavior for collectLatest seems very much alike to switchMap but for a suspending function call, as opposed to the switching on the flow...

[BoxResin]

@elizarov Yes, I considered switchMap at first. Its behavior is very similar to collectLatest but the use-case looks weird.

suspend fun main() {
    val flow = flowOf(1, 2, 3, 4, 5)
    flow.switchMap { it -> flowOf(it) }.collect { value: Int -> 
        delay(100)
        println(value) // So only '5' will be printed.
    }    
}

I just want to take the 'switching' behavior of switchMap without 'mapping'.

[elizarov]

Thanks. That is an interesting observation and it raises the whole set of issues with respect to naming. I wonder if we've making a mistaking of providing switchMap operator in the first place. Here I've recorded by thoughts: #1335

[BoxResin]

@elizarov
I found a more common use-case.

val accountFlow: Flow<Account?> // null value means that user has signed out.

class Account {
    val emailFlow: Flow<String>
    val nicknameFlow: Flow<String>
    val phoneNumberFlow: Flow<String>
}

// Suppose this code is in a suspend lambda.
accountFlow.collectLatest { account: Account? ->
    if (account != null) {
        showAccountUI()

        // This lambda should be canceled when a new account is emitted.
        account.emailFlow.collect { email: String -> /* Redraw email UI */ }

        // This lambda should be canceled when a new account is emitted.
        account.nicknameFlow.collect { nickname: String -> /* Redraw nickname UI */ }

        // This lambda should be canceled when a new account is emitted.
        account.phoneNumberFlow.collect { phoneNumber: String -> /* Redraw phone number UI */ }
    } else {
        showEmptyAccountUI()
    }
}

The accountFlow may emit null value, so accountFlow.switchMap { it.emailFlow }.collect { ... } is not available in this case.

[qwwdfsad]

@BoxResin please note that Flow (as opposed to Rx) supports nulls, so using switchMap is still possible:

suspend fun Flow<T>.collectLatest(action: suspend (value: T) -> Unit) = switchMap { flowOf(it) }.collect(action)