[Swift] Reactive Programming Combine - 6: Time manipulation

Time Manipulation Operators

delay

1초마다 값을 방출하는 publisher에서 값을 지연하여 방출할 수 있습니다.

let valuesPerSecond = 1.0
let delayInSeconds = 1.5

let sourcePublisher = PassthroughSubject<Date, Never>()
let delayedPublisher = sourcePublisher.delay(for: .seconds(delayInSeconds), scheduler: DispatchQueue.main)

let subscription = Timer
  .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
  .autoconnect()
  .subscribe(sourcePublisher)

delay

 

collect

지정된 간격으로 publisher가 내보내는 값을 수집해야 할 수 있습니다. 버퍼링의 한 형태로 유용할 수 있습니다.

collect를 이용하여 시간을 기준으로 그룹화하는 전략을 사용합니다.

 

또 다른 방법으로는 특정 시간 또는 버퍼의 최대 크기에 도달할 때까지 값을 수집하고 게시합니다.

let valuesPerSecond = 1.0
let collectTimeStride = 4
let collectMaxCount = 2

let sourcePublisher = PassthroughSubject<Date, Never>()

let collectedPublisher = sourcePublisher
    .collect(
        .byTime(
            DispatchQueue.main,
            .seconds(collectTimeStride)
        )
    )
    .flatMap { dates in dates.publisher }

let collectedPublisher2 = sourcePublisher
    .collect(
        .byTimeOrCount(
            DispatchQueue.main,
            .seconds(collectTimeStride),
            collectMaxCount
        )
    )
  .flatMap { dates in dates.publisher }

let subscription = Timer
    .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
    .autoconnect()
    .subscribe(sourcePublisher)

 

flatMap은 개별 값으로 분해하지만 시퀀스의 모든 값을 동시에 모두 방출합니다.

collected values

 

 

debounce

이벤트 사이에 지정된 시간 간격이 경과한 후에만 값을 게시합니다.

downstream으로 전달되는 값의 수를 지정된 속도로 줄여야 하는 폭주하거나 대용량 트래픽을 처리하는 데 유용합니다.

var cancellable: AnyCancellable
let bounces:[(Int,TimeInterval)] = [
    (0, 0),
    (1, 0.25),  // 0.25s interval since last index
    (2, 1),     // 0.75s interval since last index
    (3, 1.25),  // 0.25s interval since last index
    (4, 1.5),   // 0.25s interval since last index
    (5, 2)      // 0.5s interval since last index
]

let subject = PassthroughSubject<Int, Never>()
cancellable = subject
    .debounce(for: .seconds(0.5), scheduler: RunLoop.main)
    .sink { index in
        print ("Received index \(index)")
    }

for bounce in bounces {
    DispatchQueue.main.asyncAfter(deadline: .now() + bounce.1) {
        subject.send(bounce.0)
    }
}

//  Time 0: Send index 0.
//  Time 0.25: Send index 1. Index 0 was waiting and is discarded.
//  Time 0.75: Debounce period ends, publish index 1.
//  Time 1: Send index 2.
//  Time 1.25: Send index 3. Index 2 was waiting and is discarded.
//  Time 1.5: Send index 4. Index 3 was waiting and is discarded.
//  Time 2: Debounce period ends, publish index 4. Also, send index 5.
//  Time 2.5: Debounce period ends, publish index 5.

 

trottle

지정된 시간 간격 내에 upstream publisher가 게시한 가장 최근 요소 또는 첫 번째 요소를 내보냅니다.

latest 옵션이 true라면 시간 간격 동안 받은 마지막 요소를 false라면 첫 번째 요소를 반환합니다.

cancellable = Timer.publish(every: 3.0, on: .main, in: .default)
    .autoconnect()
    .print("\(Date().description)")
    .throttle(for: 10.0, scheduler: RunLoop.main, latest: true)
    .sink(
        receiveCompletion: { print ("Completion: \($0).") },
        receiveValue: { print("Received Timestamp \($0).") }
     )


//  Publish at: 2020-03-19 18:26:54 +0000: receive value: (2020-03-19 18:26:57 +0000)
//  Received Timestamp 2020-03-19 18:26:57 +0000.
//  Publish at: 2020-03-19 18:26:54 +0000: receive value: (2020-03-19 18:27:00 +0000)
//  Publish at: 2020-03-19 18:26:54 +0000: receive value: (2020-03-19 18:27:03 +0000)
//  Publish at: 2020-03-19 18:26:54 +0000: receive value: (2020-03-19 18:27:06 +0000)
//  Publish at: 2020-03-19 18:26:54 +0000: receive value: (2020-03-19 18:27:09 +0000)
//  Received Timestamp 2020-03-19 18:27:09 +0000.

 

Debounce 와 Trottle 출력의 차이

Debounce는 이벤트를 받다가 일정 기간이 지나도록 이벤트가 없다면 마지막 값을 방출합니다.

Trottle은 일정 기간 동안의 첫 번째 혹은 마지막 값을 방출합니다.

즉, Trottle은 일정 간격으로 값을 방출하지만, Debounce는 일정 간격동안 이벤트가 없어야만 값을 방출합니다.

timeout

timeout의 주 목적은 실제 타이머와 timeout 조건을 의미론적으로 구분하는 것입니다.

upstream publisher가 요소를 생성하지 않고 시간 간격을 초과하는 경우 종료됩니다.

 

Timeout이 실행되면 publisher를 완료하거나 지정된 오류를 발생시킵니다. 두 경우 모두 종료됩니다.

let subject = PassthroughSubject<String, Never>()
let cancellable = subject
    .timeout(.seconds(5), scheduler: DispatchQueue.main, options: nil, customError: nil)
    .sink(
          receiveCompletion: { print ("completion: \($0) at \(Date())") },
          receiveValue: { print ("value: \($0) at \(Date())") }
     )

DispatchQueue.main.asyncAfter(
    deadline: .now() + .seconds(2),
    execute: {
        subject.send("Some data - sent after a delay of 2 seconds")
    })

DispatchQueue.main.asyncAfter(
    deadline: .now() + .seconds(8),
    execute: {
        subject.send("Some data - sent after a delay of 8 seconds")
    })
    
// *--- RESULT ---*
// value: Some data - sent after a delay of 2 seconds at 2024-05-16 01:53:10 +0000
// completion: finished at 2024-05-16 01:53:15 +0000

 

5초 동안 publisher가 요소를 보내지 않아 실패로 완료합니다.

enum TimeoutError: Error {
    case timeout
}

let subject = PassthroughSubject<String, TimeoutError>()
let cancellable = subject
    .timeout(.seconds(5), scheduler: DispatchQueue.main, customError: { .timeout })
    .sink(
          receiveCompletion: { print ("completion: \($0) at \(Date())") },
          receiveValue: { print ("value: \($0) at \(Date())") }
     )
     
// *--- RESULT ---*
// completion: failure(__lldb_expr_225.TimeoutError.timeout) at 2024-05-16 02:42:27 +0000

 

measureInterval

measureInterval(using:) 연산자는 publisher가 내보냉 두 개의 연속된 값 사이엥 경과한 시간을 측정하여 전송합니다.

cancellable = Timer.publish(every: 1, on: .main, in: .default)
    .autoconnect()
    .measureInterval(using: RunLoop.main)
    .sink { print("\($0)", terminator: "\n") }


// *--- RESULT ---*
// Stride(magnitude: 1.0013610124588013)
// Stride(magnitude: 0.9992760419845581)

 

---

Key points

• 댐과 같이 시간 경과에 따른 값의 흐름을 관리하고 청크 단위로 값을 해제할 수 있습니다.
• 시간 이동을 제공하지 않지만 불연속적인 이벤트를 처리하는 대신 장기간에 걸친 작업을 추상화할 수 있습니다.