delay.kt

import io.reactivex.rxjava3.core.Flowable
import io.reactivex.rxjava3.core.Scheduler
import io.reactivex.rxjava3.schedulers.TestScheduler
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.CsvSource
import java.util.concurrent.TimeUnit

class DelayEmits {
    @ParameterizedTest
    @CsvSource("45,225,11", "45,225,5", "225,45,5")
    fun emitEvery(from: Int, to: Int, step: Int) {
        val testScheduler = TestScheduler()

        val list = when (from < to) {
            true -> genSeq(from, to, step).toList()
            false -> genSeq(to, from, step).toList().reversed()
        }

        val test = list.toDelayedFlowable(step, testScheduler).test()

        testScheduler.advanceTimeBy(10, TimeUnit.HOURS)

        test.assertValueCount(((kotlin.math.abs(from - to)) / step) + 2)
        // last value is 225, and not 9*20
        // .assertValueAt(16, 225)
    }
}

internal fun <T> Iterable<T>.toDelayedFlowable(step: Int, scheduler: Scheduler): Flowable<T> {
    return Flowable.fromIterable(this)
            .flatMap({ Flowable.just(it).delay(step.toLong(), TimeUnit.SECONDS, scheduler) }, 1)
}

internal fun genSeq(from: Int, to: Int, step: Int): Sequence<Int> {
    return generateSequence(from, {
        val inc = it + step
        when {
            it == to -> null
            inc > to -> to
            else -> inc
        }
    })
}