Extension class to convert ReactiveX operators into Coroutines

OperatorExtensions.kt

import kotlinx.coroutines.*

fun throttleClick(scope: CoroutineScope = CoroutineScope(Dispatchers.Main), clickAction: (Unit) -> Unit): (Unit) -> Unit {
    return throttleFirst(1_000, scope, clickAction)
}

/**
 * Processes input data and passes the first data to [action] and skips all new data for the next [skipMs].
 */
fun <T> throttleFirst(skipMs: Long = 700L, scope: CoroutineScope = CoroutineScope(Dispatchers.Main), action: (T) -> Unit): (T) -> Unit {
    var throttleJob: Job? = null
    return { param: T ->
        if (throttleJob?.isCompleted != false) { //if null or has completed
            throttleJob = scope.launch {
                action(param)
                delay(skipMs)
            }
        }
    }
}

/**
 * Processes input data and passes the latest data to [action] in the end of every time window with length of [intervalMs].
 */
fun <T> throttleLatest(intervalMs: Long = 700L, scope: CoroutineScope = CoroutineScope(Dispatchers.Main), action: (T) -> Unit): (T) -> Unit {
    var throttleJob: Job? = null
    var latestParam: T
    return { param: T ->
        latestParam = param
        if (throttleJob?.isCompleted != false) { //if null or has completed
            throttleJob = scope.launch {
                delay(intervalMs)
                action(latestParam)
            }
        }
    }
}

/**
 * Processes input data and passes it to [action] only if there's no new data for at least [waitMs]
 */
fun <T> debounce(waitMs: Long = 700L, scope: CoroutineScope = CoroutineScope(Dispatchers.Main), action: (T) -> Unit): (T) -> Unit {
    var debounceJob: Job? = null
    return { param: T ->
        debounceJob?.cancel()
        debounceJob = scope.launch {
            delay(waitMs)
            action(param)
        }
    }
}

/**
 * Returns a function that emits 0 after the [initialDelay] and ever increasing [Int] numbers
 * after each [period] of time thereafter, on a specified [scope], until cancellation.
 */
fun interval(initialDelay: Long = 0L, period: Long = 700L, scope: CoroutineScope = CoroutineScope(Dispatchers.Main), callback: (Int) -> Unit): Job {
    var value = 0
    return scope.launch {
        delay(initialDelay)
        while (isActive) {
            callback(value++)
            delay(period)
        }
    }
}