Kotlin Channels Debounce & Throttle

channel.kt

import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.ReceiveChannel
import kotlinx.coroutines.experimental.channels.consumeEach
import kotlinx.coroutines.experimental.channels.produce
import kotlin.coroutines.experimental.CoroutineContext

fun <E> ReceiveChannel<E>.debounce(
    wait: Long = 50,
    context: CoroutineContext = DefaultDispatcher
): ReceiveChannel<E> = produce(context) {
    var lastTimeout: Job? = null
    consumeEach {
        lastTimeout?.cancel()
        lastTimeout = launch {
            delay(wait)
            send(it)
        }
    }
    lastTimeout?.join()
}

fun <E> ReceiveChannel<E>.throttle(
    wait: Long = 200,
    context: CoroutineContext = DefaultDispatcher
): ReceiveChannel<E> = produce(context) {
    var nextTime = 0L
    consumeEach {
        val curTime = System.currentTimeMillis()
        if (curTime >= nextTime) {
            nextTime = curTime + wait
            send(it)
        }
    }
}

main.kt

import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.ReceiveChannel
import kotlinx.coroutines.experimental.channels.consumeEach
import kotlinx.coroutines.experimental.channels.produce
import kotlin.coroutines.experimental.CoroutineContext

fun main(args: Array<String>) = runBlocking {
    val channel = produce<Int> {
        (0..100).forEach {
            send(it)
            delay(20)
        }
    }
    channel.throttle().consumeEach { println(it)  }
}

System.currentTimeMillis() is not reliable, could be modified by user at any time, SystemClock.uptimeMillis() is much better.