Time-based Operators

1. Interval (특정 주기마다 정수를 방출하는 연산자)

let i = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)

let subscription1 = i.subscribe { print("1 >> \($0)") }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    subscription1.dispose()
}

=================================================================================
결과 
1 >> next(0)
1 >> next(1)
1 >> next(2)
1 >> next(3)
1 >> next(4)

위 코드에선 asyncAfter를 통해 5초 뒤에 dispose() 되도록 구현하여 구독이 중지된다.

interval 연산자가 생성하는 Observable은 내부에 Timer를 갖고 있다. 이는 생성 시점이 아닌 구독자가 구독을 시작하는 시점에 동작한다. 아래 코드를 보자~!

let i = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)

let subscription1 = i.subscribe { print("1 >> \($0)") }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    subscription1.dispose()
}

var subsription2: Disposable?

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    subsription2 = i.subscribe { print("2 >> \($0)") }
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    subsription2?.dispose()
}

=================================================================================
결과 
1 >> next(0)
1 >> next(1)
1 >> next(2)
2 >> next(0)
1 >> next(3)
2 >> next(1)
1 >> next(4)
2 >> next(2)
2 >> next(3)

위에서 보았던 코드에 추가적으로 다른 구독자를 추가하였다. subsription2를 2초 뒤부터 두번구독을 시작하도록 하여 다시 0부터 방출하는 것을 볼 수 있다.

interval 연산자의 핵심은 새로운 구독이 추가되면 내부 타이머가 시작된다는 점이다!

2. Timer (지연 시간과 반복 주기를 지정해서 정수를 방출하는 연산자)

let bag = DisposeBag()

let subsription1 = Observable<Int>.timer(.seconds(1), period: .milliseconds(500),scheduler: MainScheduler.instance)
    .subscribe { print($0) }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    subsription1.dispose()
}

==============================================================
결과
next(0)
next(1)
next(2)
next(3)
next(4)
next(5)
next(6)
next(7)
next(8)

timer 연산자에서 첫 번재 파라미터는 반복 주기가 아닌 첫번째 요소가 구독 후 구독자에게 전달되는 시간이다. 두번째 파라미터가 반복주기이다.

3. Timeout (지정된 시간 이내에 요소를 방출하지 않으면 에러 이벤트를 전달)

let bag = DisposeBag()

let subject = PublishSubject<Int>()

subject.timeout(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print($0) }
    .disposed(by: bag)


Observable<Int>.timer(.seconds(2), period: .seconds(5), scheduler: MainScheduler.instance)
    .subscribe(onNext: { subject.onNext($0) })
    .disposed(by: bag)

================================================================================================    
결과
next(0)
error(Sequence timeout.)

위 코드에서 timeout 연산자의 첫번째 파라미터는 지정된 시간이다. 즉 저 시간안에 요소를 방출하지 않으면 에러 이벤트가 전달되는 것이다.

앞서 보았던 timer를 통해 첫번째 요소가 구독 후 구독자에게 방출되는 시간이 2초이기 때문에 타임아웃에 걸리지 않지만 주기가 5초이기 때문에 첫 번째 요소가 방출된 후 에러 이벤트가 전달된다.

timeout 연산자는 또 다른 파라미터를 받을 수 있다. 아래 코드를 보자

let bag = DisposeBag()

let subject = PublishSubject<Int>()

subject.timeout(.seconds(3), other: Observable.just(0), scheduler: MainScheduler.instance)
    .subscribe { print($0) }
    .disposed(by: bag)

Observable<Int>.timer(.seconds(2), period: .seconds(5), scheduler: MainScheduler.instance)
    .subscribe(onNext: { subject.onNext($0) })
    .disposed(by: bag)

================================================================================================ 
결과
next(0)
next(0)
completed

위 코드에서 보면 other: 라는 파라미터가 있다. 이는 timeout이 발생하는 경우 other 파라미터에 들어간 값으로 구독 대상이 변경된다. 즉 에러 이벤트가 전달되는 것이 아닌 Observable.just(0)이 전달되고 구독이 종료된다.

4. Delay (Next 이벤트가 구독자로 전달되는 시점과 구독이 시작되는 시점을 지연시키는 연산자)

let bag = DisposeBag()

func currentTimeString() -> String {
   let f = DateFormatter()
   f.dateFormat = "yyyy-MM-dd HH:mm:ss.SSS"
   return f.string(from: Date())
}

Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
    .take(10)
    .debug()
    .delay(.seconds(5), scheduler: MainScheduler.instance)
    .subscribe { print( currentTimeString(), $0) }
    .disposed(by: bag)
    
================================================================================================ 
결과
2024-05-20 17:23:56.601: delay.playground:41 (__lldb_expr_3) -> subscribed
2024-05-20 17:23:57.668: delay.playground:41 (__lldb_expr_3) -> Event next(0)
2024-05-20 17:23:58.667: delay.playground:41 (__lldb_expr_3) -> Event next(1)
2024-05-20 17:23:59.667: delay.playground:41 (__lldb_expr_3) -> Event next(2)
2024-05-20 17:24:00.667: delay.playground:41 (__lldb_expr_3) -> Event next(3)
2024-05-20 17:24:01.667: delay.playground:41 (__lldb_expr_3) -> Event next(4)
2024-05-20 17:24:02.667: delay.playground:41 (__lldb_expr_3) -> Event next(5)
2024-05-20 17:24:02.671 next(0)
2024-05-20 17:24:03.668: delay.playground:41 (__lldb_expr_3) -> Event next(6)
2024-05-20 17:24:03.675 next(1)
2024-05-20 17:24:04.667: delay.playground:41 (__lldb_expr_3) -> Event next(7)
2024-05-20 17:24:04.677 next(2)
2024-05-20 17:24:05.667: delay.playground:41 (__lldb_expr_3) -> Event next(8)
2024-05-20 17:24:05.679 next(3)
2024-05-20 17:24:06.667: delay.playground:41 (__lldb_expr_3) -> Event next(9)
2024-05-20 17:24:06.667: delay.playground:41 (__lldb_expr_3) -> Event completed
2024-05-20 17:24:06.667: delay.playground:41 (__lldb_expr_3) -> isDisposed
2024-05-20 17:24:06.681 next(4)
2024-05-20 17:24:07.683 next(5)
2024-05-20 17:24:08.685 next(6)
2024-05-20 17:24:09.686 next(7)
2024-05-20 17:24:10.688 next(8)
2024-05-20 17:24:11.690 next(9)
2024-05-20 17:24:11.690 completed
    

위 코드를 보면 delay 연산자를 통해 5초 뒤에 next 이벤트가 전달되는 것을 볼 수 있다.
원본 Observable이 방출한 next 이벤트가 5초 뒤에 구독자에게 전달된 것이다.

만약 구독 시점을 지연 시키기 위해선 delaySubsription 연산자를 사용하면 된다.

let bag = DisposeBag()

func currentTimeString() -> String {
   let f = DateFormatter()
   f.dateFormat = "yyyy-MM-dd HH:mm:ss.SSS"
   return f.string(from: Date())
}

Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
    .take(10)
    .debug()
    .delaySubscription(.seconds(7), scheduler: MainScheduler.instance)
    .subscribe { print(currentTimeString(), $0) }
    .disposed(by: bag)

================================================================================================ 
결과
2024-05-20 17:30:25.048: delaySubscription.playground:41 (__lldb_expr_7) -> subscribed
2024-05-20 17:30:26.052: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(0)
2024-05-20 17:30:26.053 next(0)
2024-05-20 17:30:27.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(1)
2024-05-20 17:30:27.051 next(1)
2024-05-20 17:30:28.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(2)
2024-05-20 17:30:28.051 next(2)
2024-05-20 17:30:29.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(3)
2024-05-20 17:30:29.051 next(3)
2024-05-20 17:30:30.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(4)
2024-05-20 17:30:30.051 next(4)
2024-05-20 17:30:31.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(5)
2024-05-20 17:30:31.052 next(5)
2024-05-20 17:30:32.052: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(6)
2024-05-20 17:30:32.053 next(6)
2024-05-20 17:30:33.051: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(7)
2024-05-20 17:30:33.052 next(7)
2024-05-20 17:30:34.052: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(8)
2024-05-20 17:30:34.053 next(8)
2024-05-20 17:30:35.052: delaySubscription.playground:41 (__lldb_expr_7) -> Event next(9)
2024-05-20 17:30:35.052 next(9)
2024-05-20 17:30:35.053: delaySubscription.playground:41 (__lldb_expr_7) -> Event completed
2024-05-20 17:30:35.053 completed
2024-05-20 17:30:35.054: delaySubscription.playground:41 (__lldb_expr_7) -> isDisposed

delaySubscription 연산자는 구독 시점을 지연시킬 뿐 next 이벤트가 전달하는 시점은 지연 시키지 않는다.

KXCoding 강의
Marble Diagram
ReactiveX 사이트