How can I transform a `Flow` that sends three stat...

# coroutines

[dave08]

How can I transform a Flow that sends three states Starting, Progress and Finished (from a sealed class), into a Flow that assures receiving Starting and Finished but conflates Progress for backpressure on latency of the collector...?

[diesieben07]

sealed class State {
    object Starting : State()

    data class Progress(val progress: Int) : State()

    object Finished : State()
}

fun Flow<State>.checkAndConflate(): Flow<State.Progress> {
    return flow<State.Progress> {
        var first = true
        var done = false
        collect {
            check(!done)
            when {
                first -> {
                    check(it is State.Starting)
                    first = false
                }
                it is State.Progress -> {
                    emit(it)
                }
                else -> {
                    check(it is State.Finished)
                    done = true
                }
            }
        }
    }.conflate()
}

[dave08]

Thanks! But I really need it to be Flow<State> as a result, only with the Progress being conflated... (there's some properties I need on Started and Finished...

[diesieben07]

well, should be easy to adapt my code to just re-emit those items as well

[diesieben07]

Ah, actually, not really. give me one second.

[diesieben07]

fun Flow<State>.checkAndConflate(): Flow<State> {
    val conflatedProgressFlow = flow<State> {
        var first = true
        var done = false
        collect {
            check(!done)
            when {
                first -> {
                    check(it is State.Starting)
                    first = false
                }
                it is State.Progress -> {
                    emit(it)
                }
                else -> {
                    check(it is State.Finished)
                    done = true
                }
            }
        }
    }.conflate()

    val startFlow = take(1)
    val finishedFlow = dropWhile { it !is State.Finished }

    return conflatedProgressFlow.onStart { emitAll(startFlow) }.onCompletion { emitAll(finishedFlow) }
}

Not sure if it's the optimal way to do this though...

[diesieben07]

Probably not

[diesieben07]

Scratch that, definitely not optimal, because it runs the flow three times

[dave08]

Yup... this problem is a result of Android's Notifications not being designed to receive too many updates... I really have to use delay in the receiver to skip some of the progress, but I need the first and last emissions to display start and finish notification states...

[dave08]

So far, I had to use broadcastIn and filter on multiple subscribers launched on different coroutines to get this to work... not too much cleaner than your solution 🙈...

[dave08]

Thanks for the attempt though 😉!

[diesieben07]

I have another solution, one second 😉

[Dominaezzz]

Would this operator help? https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/debounce.html

[dave08]

Only for the progress emissions... but the first and last ones shouldn't be debounced...

[dave08]

emit(1) was skipped over there.. and emit(5) would have been skipped if it would have some smaller delay...

[diesieben07]

fun Flow<State>.checkAndConflate(): Flow<State> {
    return channelFlow<State> {
        val progressChannel = Channel<State>(Channel.CONFLATED)
        launch {
            try {
                collect {
                    if (it is State.Progress) {
                        progressChannel.send(it)
                    } else {
                        [e-mail](it)
                    }
                }
            } finally {
                progressChannel.close()
                [e-mail]()
            }
        }
        progressChannel.consumeEach {
           [e-mail](it)
        }

        awaitClose()
    }.buffer(Channel.RENDEZVOUS)
}

This is what I came up with, but it breaks the sequence of elements

[diesieben07]

Revised again and it works now:

fun Flow<State>.checkAndConflate(): Flow<State> {
    return flow<State> {
        coroutineScope {
            val start = CompletableDeferred<State.Starting>()
            val finish = CompletableDeferred<State.Finished>()

            val progressChannel = produce<State.Progress>(capacity = Channel.CONFLATED) {
                collect {
                    when (it) {
                        is State.Starting -> start.complete(it)
                        is State.Progress -> send(it)
                        is State.Finished -> finish.complete(it)
                    }
                }
            }

            emit(start.await())
            progressChannel.consumeEach { emit(it) }
            emit(finish.await())
        }
    }
}

[dave08]

That's not bad 👌🏼, so to debounce the progress I would just add :

...
is State.Progress -> { delay(timoutMillis); send(it) }
...

[dave08]

Thanks!

[Dominaezzz]

I guess a debounceWhen(timeoutMillis) { ... } operator would be nice.