kotlin - Flowable concatMapSingle without prefetch to ignore clicks until processing finishes

Question

I want to handle clicks in such a way that they are ignored as long as I'm doing processing of some click that occurred.

I thought I could do it by utilizing the backpressure, like this:

private val clicks = PublishProcessor.create<Unit>()

// ...

clicks
    .onBackpressureDrop()
    .concatMapSingle(::handleClick, 0)

But this throws an error, because there's a requirement that concatMapSingle needs to prefetch at least one item, which makes it queue the click and process it immediately after I'm done processing, which is not what I want. I want to process the click only if there is no processing happening at the moment.

Is there some other operator I could use to achieve the desired effect?

Answer 1

Using flatMapSingle instead of concatMapSingle does the trick, as suggested by akarnokd on GitHub:

flatMap will only fetch the next upstream item if the current one actually completed

The last parameter is maxConcurrency, specifying the maximum number of active subscriptions to the SingleSources:

clicks
    .onBackpressureDrop()
    .flatMapSingle(::handleClick, false, 1)

In this instance, flatMapSingle subscribes to those Singles sequentially, so it doesn't change the semantics I got from concatMapSingle.