Flow onBackpressureReduce operator?

[shirozatou]

Use case

Fast data producer but slow collector, reduce information from each event without drop or suspend.
For example, send update events to channel, receive most recent result at long running task.

val f = flow {
    repeat(52) {
        emit(it)
        delay(100)
    }
}
f.onBackpressureReduce(::maxOf).collect {
    println(it)
    delay(1000)
}
// output: 0, 9, 19, 29, 39, 49, 51

The Shape of the API

// sample
fun <T : Any> Flow<T>.onBackpressureReduce(reducer: (T, T) -> T): Flow<T> {
    return channelFlow {
        var acc: T? = null
        collectLatest { value ->
            val v = acc?.let { reducer(it, value) } ?: value
            acc = v
            withContext(Dispatchers.Unconfined) {
                send(v)
                acc = null
            }
        }
    }.buffer(capacity = 0)
}

[qwwdfsad]

Hey, could you please share your product-specific use case? E.g. for which kind of events you want to apply this operator and what's the app logic flow that allows it/benefits from it?

This one has quite some parallels with #3884 and it would be interesting to see how it plays out together. Alternatively, it can be a channel BufferOverflow strategy

[shirozatou]

Hello, I can't share product use case, but this operator seems very common.

Like bus stop,
Reduce: passengers at bus stop.
Collector: scheduled bus, no waiting.

My use case is about handling duplicated endpoint request, accumulated parameter, one request each turn, but multiple callbacks.

How about this one?

    val logger = Channel<String>(0)
    launch {
        repeat(20) {
            logger.send("log-$it")
            delay(100)
        }
        logger.close()
    }
    //......

    logger.receiveAsFlow()
        .map { listOf(it) }
        .onBackpressureReduce { a, b -> a + b }
        .collect {
            // process as fast as possible.
            println("batch upload: $it")
            delay(500)
        }
//  batch upload: [log-0]
//  batch upload: [log-1, log-2, log-3, log-4, log-5]
//  batch upload: [log-6, log-7, log-8, log-9]
//  batch upload: [log-10, log-11, log-12, log-13, log-14]
//  batch upload: [log-15, log-16, log-17, log-18, log-19]

[shirozatou]

Alternatively, it can be a channel BufferOverflow strategy

SUSPEND: not upstream friendly,
DROP_*: data loss,

Operators based on transformLatest: not a choice for some uncancellable process.

[qwwdfsad]

What I've meanth is some new strategy, not the existing one, sorry for the confusion.
That operator seems to be the backpressure counterpart of natural batching, with batching happening on the side of the upstream.

but this operator seems very common.

Maybe you can share what you are using (or were using with another frameworks) instead?

[shirozatou]

What I've meanth is some new strategy

Good idea.

you can share what you are using

onBackpressureReduce

RxJava is what I am using, I'm ready to migrate to Flow which has multiplatform support.