17. RxJava - delay, timeInterval

delay

timer, interval, defer처럼 delay도 시간을 다루는 함수입니다.

timer, interval, defer가 Observable을 생성하는 역할이라면, delay는 연산자로서 보조 역할을 합니다.

 

 

입력

import common.CommonUtils
import common.Log
import io.reactivex.Observable
import java.util.concurrent.TimeUnit

class DelayExample {
    fun marbleDiagram() {
        val data = arrayOf("1", "7", "2", "3", "4")

        val source = Observable.fromArray(*data)
            .delay(100, TimeUnit.MILLISECONDS)
        source.subscribe { value -> Log.it(value) }
        CommonUtils.sleep(1000)
    }
}

fun main() {
    val demo = DelayExample()
    demo.marbleDiagram()
}

 

출력

RxComputationThreadPool-1 | value = 1
RxComputationThreadPool-1 | value = 7
RxComputationThreadPool-1 | value = 2
RxComputationThreadPool-1 | value = 3
RxComputationThreadPool-1 | value = 4

 

 

delay 함수는 인자로 delay 변수와 시간 단위를 받는다. 그리고 interval과 마찬가지로 계산 스케줄러에서 실행합니다.

 

 

 

 

timeInterval

어떤 값을 발행했을 때 이전 값을 발행한 이후 얼마나 시간이 지났는지 알려 줍니다.

 

 

입력

import common.CommonUtils
import common.Log
import io.reactivex.Observable
import kotlin.random.Random

class TimeIntervalExample {
    fun marbleDiagram() {
        val data = arrayOf("1", "3", "7")

        CommonUtils.start()

        val source = Observable.fromArray(*data)
            .delay { item ->
                Thread.sleep(Random.nextLong(100))
                Observable.just(item)
            }
            .timeInterval()

        source.subscribe { value -> Log.it(value) }
        CommonUtils.sleep(1000)
    }
}

fun main() {
    val demo = TimeIntervalExample()
    demo.marbleDiagram()
}

 

출력

main | 440 | value = Timed[time=94, unit=MILLISECONDS, value=1]
main | 512 | value = Timed[time=77, unit=MILLISECONDS, value=3]
main | 515 | value = Timed[time=3, unit=MILLISECONDS, value=7]