I'm looking to create a "queue" of some sort that ...

# coroutines

[Kieran Wallbanks]

I'm looking to create a "queue" of some sort that I can put objects into that get processed async. This sounds like a great usecase for a Flow/Channel, but I can't seem to find a way to replicate this exact logic:
• The objects will be processed when the queue has reached N elements.
• The objects will be processed when the last process was X milliseconds ago.
• The same object should never be processed more than once.
Does anyone have any pointers on this?

[ross_a]

Something I have kicking around

val channel = Channel<String>()

GlobalScope.launch {
    var bufferedItems = mutableListOf<String>()
    whileSelect {
        if (bufferedItems.isNotEmpty()) onTimeout(1000) {

            // TODO process items

            bufferedItems = mutableListOf()
            true
        }
        channel.onReceiveCatching { result ->
            result
                .onSuccess { item -> bufferedItems += item }
                .onFailure {
                    if (bufferedItems.isNotEmpty()) {
                        // TODO process items 
                        
                    }
                }
                .isSuccess
        }
    }
}

[Kieran Wallbanks]

I can't really see how this would help me solve my issue of processing the items when the queue reaches a certain size or when it's been a certain amount of time since the last process, unless I'm missing something?

[ross_a]

the onTimeout performs it after a period of silence

[ross_a]

and this minor tweak does it if the number of items is > 5

.onSuccess { item -> 
    bufferedItems += item
    if (bufferedItems.size > 5) {
        
        // TODO process items in bufferedItems
        
        bufferedItems = mutableListOf()
    }
}

[Kieran Wallbanks]

Ah I see! So the code in onTimeout will just run the block every X milliseconds and the code in the onSuccess will process as well at that time. I'm also assuming this is all concurrent enough to not hit any issues if the onTimeout and onSuccess blocks call at the same time?

[ross_a]

Sorry I thought it was the last event (debounce) rather than last process

[ross_a]

whileSelect will choose only one of the paths to run at a time yeah

[ross_a]

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.selects/select.html

[Kieran Wallbanks]

That's very snazzy! And then I assume that when the channel is closed, the onFailure branch will trigger that does the final cleanup and then stops the while?

[ross_a]

yes correct

[ross_a]

You can do similar things to produce rather advanced custom Flow operators quite simply, for example that example was stolen from my operator:

/**
 * Buffers all elements emitted during a debounced window and emits them as a [List] once the debounce timer expires.
 */
fun <T : Any?> Flow<T>.bufferingDebounce(
    timeoutMillis: Long
): Flow<List<T>> = channelFlow {
    val itemChannel = produceIn(this)
    var bufferedItems = mutableListOf<T>()
    whileSelect {
        if (bufferedItems.isNotEmpty()) onTimeout(timeoutMillis) {
            send(bufferedItems)
            bufferedItems = mutableListOf()
            true
        }
        itemChannel.onReceiveCatching { result ->
            result
                .onSuccess { item -> bufferedItems += item }
                .onFailure { if (bufferedItems.isNotEmpty()) send(bufferedItems) }
                .isSuccess
        }
    }
}

[Kieran Wallbanks]

That's really cool - thank you for your help!

[ross_a]

👍

[debop]

Above code, timeoutMillis applied on each items. not debounded windows for multiple items.

/**
 * debounced window 내에 발생한 모든 요소를 버퍼링하고, 디바운스 타이머가 만료되면 [List]로 발행합니다.
 *
 *

* val source = flow {
* emit(1)
* delay(110)
* emit(2)
* delay(90)
* emit(3)
* delay(110)
* emit(4)
* delay(90)
* }
* val buffered = source.bufferingDebounce(200.milliseconds) // [1, 2], [3, 4]
*

*
 * @param timeout 디바운스 타임아웃
 * @return 디바운스된 [List] 요소를 발행하는 [Flow]
 */
fun <T: Any?> Flow<T>.bufferingDebounce(timeout: Duration): Flow<List<T>> = channelFlow {
    val itemChannel = produceIn(this)
    var bufferedItems = mutableListOf<T>()
    var deboundedTimeout = timeout

    whileSelect {
        var prevTimeMs = System.currentTimeMillis()
        if (bufferedItems.isNotEmpty()) {
            onTimeout(deboundedTimeout) {
                send(bufferedItems)
                bufferedItems = mutableListOf()
                deboundedTimeout = timeout
                true
            }
        }
        itemChannel.onReceiveCatching { result ->
            val receiveTimeMs = System.currentTimeMillis()
            deboundedTimeout -= (receiveTimeMs - prevTimeMs).milliseconds
            prevTimeMs = receiveTimeMs
            result
                .onSuccess { item -> bufferedItems.add(item) }
                .onFailure { if (bufferedItems.isNotEmpty()) send(bufferedItems) }
                .isSuccess
        }
    }
}

[debop]

Test code

@Test
    fun `debounced window 내에 발생한 모든 요소를 버퍼링하고, 디바운스 타이머가 만료되면 List로 발행합니다`() = runBlocking<Unit> {
        val source = flow {
            emit(1)
            delay(110)
            emit(2)
            delay(90)
            emit(3)
            delay(110)
            emit(4)
            delay(90)
        }
        val buffered = source.bufferingDebounce(200.milliseconds)  // [1, 2], [3, 4]
        val itemLists = buffered.toList()
        log.debug { "itemLists=$itemLists" }
        itemLists shouldHaveSize 2 shouldBeEqualTo listOf(listOf(1, 2), listOf(3, 4))
    }