the synchronous function gradually iterates over a stream during a given timeInterval.

gradualStreamLoop.kt

fun <T> gradualLoop(timeInterval: Long, stream: Stream<T>, streamElementsCount: Long, action: (T) -> Unit) {
    println("total elements: ${streamElementsCount}")
    if (streamElementsCount > 0) {
        var accumulatingOverdue: Long = 0L
        var iterationLenFactMillis: Long = 0L // used to compensate iteration duration, so that we don't sleep through the end of the timeInterval
        val iterationMaxAvgLenMillis: Long = timeInterval / streamElementsCount // maximum sleep duration

        println("max iteration duration must be: ${iterationMaxAvgLenMillis} ms")

        stream.forEach {
            val startNanoTime = System.nanoTime()
            action(it)
            iterationLenFactMillis = (maxOf(System.nanoTime() - startNanoTime, 0)) / 1000000
            println("iteration took ${iterationLenFactMillis} ms")

            var sleepInterval: Long = 0L
            val difference = iterationMaxAvgLenMillis - iterationLenFactMillis
            if (difference < 0) {
                sleepInterval = 0L
                accumulatingOverdue += Math.abs(difference)
            } else {
                sleepInterval = maxOf(difference - accumulatingOverdue, 0)
                accumulatingOverdue = maxOf(accumulatingOverdue - difference, 0)
            }

            Thread.sleep(sleepInterval)
            println("slept for ${sleepInterval} ms, total overdue: ${accumulatingOverdue} ms")
            println("\n")
        }
    }
}