Kotlin Coroutines - Asynchronously consume a sequence

Issue

I'm looking for a way to keep a Kotlin sequence that can produces values very quickly, from outpacing slower async consumers of its values. In the following code, if the async handleValue(it) cannot keep up with the rate that the sequence is producing values, the rate imbalance leads to buffering of produced values, and eventual out-of-memory errors.

getSequence().map { async {
  handleValue(it)
}}

I believe this is a classic producer/consumer "back-pressure" situation, and I'm trying to understand how to use Kotlin coroutines to deal with it.

Thanks for any suggestions :)

Solution

For my own reference, and to anyone else this may help, here's how I eventually solved this using Channels - https://kotlinlang.org/docs/channels.html#channel-basics

A producer coroutine:

fun itemChannel() : ReceiveChannel<MyItem> {
    return produce {
        while (moreItems()) {
            send(nextItem())   // <-- suspend until next 'receive()'
        }
    }
}

And a function to run multiple consumer coroutines, each reading off that channel:

fun itemConsumers() {
    runBlocking {
        val channel = itemChannel()
        repeat(numberOfConsumers) {
            launch {
                var more = true
                while (more) {
                    try {
                        val item = channel.receive()
                        // do stuff with item here...
                    } catch (ex: ClosedReceiveChannelException) {
                        more = false
                    }
                }
            }
        }
    }
}

The idea here is that the consumer receives off the channel within the coroutine, so the next receive() is not called until a consumer coroutine finishes handling the last item. This results in the desired back-pressure, as opposed to receiving from a sequence or flow in the main thread, and then passing the item into a coroutine to be consumed. In that scenario there is no back-pressure from the receiver, since the receive happens in a different coroutine than where the received item is consumed.

Answered By - pedorro