Generic throttleFirst behavior

CoroutineThrottler.kt

import com.movile.faster.sdk.util.DispatcherProvider
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap

class CoroutineThrottler(
    private val dispatchers: DispatcherProvider,
    private val durationInMillis: Long = 300,
) {

    private val deferredCache = ConcurrentHashMap<String, Any>()

    /**
     * This implements the [throttleFirst](http://reactivex.io/documentation/operators/images/throttleFirst.png)
     * behavior for a given function.
     * This mechanism uses the provided key to identify the [[block]] function and prevent the same block from running
     * before the [[durationInMillis]] elapses.
     *
     * In short, this will effectively swallow all subsequent [[block]] for the [[durationInMillis]].
     *
     * Note: the first call to the method starts running immediately
     */
    @Suppress("UNCHECKED_CAST")
    fun <G : Any> throttleFirstAsync(key: String, block: suspend () -> G): Deferred<G> {

        val deferredRequest = CoroutineScope(dispatchers.io()).async(start = CoroutineStart.LAZY) { block() }
        val throttledRequest = deferredCache
            .putIfAbsent(key, deferredRequest)
            .let { it as? Deferred<G> }
            ?: deferredRequest

        CoroutineScope(dispatchers.default()).launch {
            delay(durationInMillis)
            deferredCache.remove(key)
        }

        return throttledRequest
    }
}