Interval Flowables

DynamicInterval.kt

private fun buildFlowableFromIntervals(
    intervals: List<Interval>,
): Flowable<ApiResponse> {
    val intervalFlowables = intervals.map(::buildFlowableForInterval)

    val completeFlowable = Flowable.concat(intervalFlowables)

    return completeFlowable
        .flatMap { 
            apiSingle().toFlowable()
        }
}

private fun buildFlowableForInterval(
    interval: Interval,
): Flowable<Long> {
    val intervalFlowable = Flowable
        .interval(
            interval.intervalSeconds.toLong(),
            TimeUnit.SECONDS,
        )
    
    // If duration seconds is not positive, emit indefinitely
    return if (interval.durationSeconds <= 0) {
        intervalFlowable
    } else {
        val totalIntervals = interval.durationSeconds / interval.intervalSeconds
        intervalFlowable.take(totalIntervals.toLong())
    }
}