import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*
import java.time.Instant
import kotlin.time.Duration.Companion.milliseconds

suspend fun main() = coroutineScope {
    // First we'll look at basic rate limiting. Suppose
    // we want to limit our handling of incoming requests.
    // We'll serve these requests off a channel of the
    // same name.
    val requests = Channel<Int>(5)
    for (i in 1..5) {
        requests.send(i)
    }
    requests.close()

    // This limiter flow will emit a value every 200 milliseconds.
    // This is the regulator in our rate limiting scheme.
    val limiter = flow {
        while (true) {
            emit(Unit)
            delay(200.milliseconds)
        }
    }

    // By collecting from the limiter flow before serving each request,
    // we limit ourselves to 1 request every 200 milliseconds.
    launch {
        requests.consumeAsFlow().collect { req ->
            limiter.first()
            println("request $req ${Instant.now()}")
        }
    }

    // We may want to allow short bursts of requests in
    // our rate limiting scheme while preserving the
    // overall rate limit. We can accomplish this by
    // buffering our limiter channel. This burstyLimiter
    // channel will allow bursts of up to 3 events.
    val burstyLimiter = Channel<Unit>(3)

    // Fill up the channel to represent allowed bursting.
    repeat(3) {
        burstyLimiter.send(Unit)
    }

    // Every 200 milliseconds we'll try to add a new
    // value to burstyLimiter, up to its limit of 3.
    launch {
        while (true) {
            delay(200.milliseconds)
            burstyLimiter.trySend(Unit)
        }
    }

    // Now simulate 5 more incoming requests. The first
    // 3 of these will benefit from the burst capability
    // of burstyLimiter.
    val burstyRequests = Channel<Int>(5)
    for (i in 1..5) {
        burstyRequests.send(i)
    }
    burstyRequests.close()

    burstyRequests.consumeAsFlow().collect { req ->
        burstyLimiter.receive()
        println("request $req ${Instant.now()}")
    }
}