hi guys i'm trying to implement rx-like `throttle`...

# coroutines

[deviant]

hi guys
i'm trying to implement rx-like throttle feature with coroutines

suspend fun <T> ReceiveChannel<T>.consumeThrottled(throttling: Long, action: (T) -> Unit) {
    var timestamp = 0L
    for (data in this) {
        if (timestamp + throttling < System.currentTimeMillis()) {
            timestamp = System.currentTimeMillis()
            action(data)
        }
    }
}

is there a proper way?

[deviant]

Your function won’t send the last element

it will be sent if throttled time is passed, which is fine for my case.

btw i like solution with select clause 👍

[deviant]

another one:

fun <T> ReceiveChannel<T>.throttle(throttling: Long) = produce(Unconfined) {
    val original = this@throttle
    var t: Job? = null
    for (item in original) {
        if (t == null || t.isCompleted) {
            send(item)
            t = launch(context) { delay(throttling) }
        }
    }
}