[Combine] Processing Published Elements with Subscribers - developer 문서로 시작한 Combine 공부(4)

이전 포스팅에서 이어집니다

https://teunteun2.tistory.com/37

[Combine] 정의 및 구조 - developer 문서로 시작한 Combine 공부(1)

 

 

https://developer.apple.com/documentation/combine/processing-published-elements-with-subscribers

Processing Published Elements with Subscribers | Apple Developer Documentation

해당 Article을 번역하고 정리하였습니다.

 

 

 

Processing Published Elements with Subscribers

Publisher가 element를 생성하는 시기를 정확하게 제어하기 위해 배압(back pressure)을 가하는 방법

 

 

OverView

Combine에서 Publisher는 element를 생성하고 Subscriber은 수신된 element에 대해 작업을 수행합니다.

그러나 Publisher은 Subscriber가 연결되고 Publisher에 요청할 때까지 element를 내보낼 수 없습니다.

 

Subscriber은 Subscribers.Demand 타입을 사용하여 자신이 수신할 수 있는 element 수를 나타내는 방식으로

Publisher가 element를 전달하는 비율을 제어할 수 있습니다.

 

Subscriber은 요구사항을 두가지 방법으로 나타낼 수 있습니다 :

 

1. Subscriber가 처음 구독되었을 때 Publisher가 제공한
   구독 인스턴스에 대한 request(_:) 메서드를 호출하는 방식으로
2. Publisher가 element를 전달하기 위해 Subscriber의 receive(_:)  메서드를 호출할 때
   새 요구사항을 리턴하는 방식으로

요구사항은 추가적인 사항입니다.

 

Subscriber가 요구한 두개의 element가 있고, 그 다음 또 Subscribers.Dement(.max(3))을 요청했다면,

Publisher의 만족되지 않은 요구사항은 다섯개의 element 입니다.

그런 다음 Publisher가 element를 전달하면, 이제 만족되지 않은 요구사항은 4개로 줄어듭니다.

element를 Publishing 하는 것은 만족되지 않은 요구를 줄이는 유일한 방법입니다.

Subscriber은 negative demand를 요청할 수 없습니다.

 

(여기에서 negative demand가 무슨뜻인지 모르겠네요?

구글링 해봤는데, 마케팅 관련 용어에 부정적 수요라는 이름으로 있긴 한데, 이 맥락에서 이 뜻인지 잘 모르겠네요)

 

대부분의 앱은 편리한 Subscriber 타입인 Subscriber.Sink, Subscriber.Assign을 생성하기 위해

각각 sink(recieveValue:)와 assign(to:on:) 연산자를 사용합니다.

이 두 Subscriber은 Publisher에 처음 연결할 때 무제한 element를 요구합니다.

Publisher가 무제한 요구를 받게되면, Subscriber와 Publisher 간의 요구 협상은 불가능하게 됩니다.

 

 

 

Consume Elements as the Publisher Produces Them

위와 같이 Publisher가 많은 element를 요구받거나, 무제한 element를 요구받을 때,

Subscriber가 element를 처리하는 속도보다 Publisher가 element를 전달하는 속도가 더 빠를 수 있습니다.

아래의 시나리오는 처리 대기중에 element가 버퍼를 채우면서

element가 누락되거나 메모리 압력이 급격히 증가하는 결과를 초래할 수 있습니다.

 

이 시나리오는 편리한 Subscriber를 사용하였을 때 무제한 element를 요구하기 때문에 발생할 수 있습니다.

sink(receiveValue:)에 제공하는 클로저와 assign(to: on:)의 부작용이 다음 특징을 따르는지 확인해봅시다:

 

• Publisher을 차단하지 마세요
• element를 버퍼링하여 메모리를 과도하게 소모하지 마세요
• element를 처리하는 데 실패하지 마세요

 

다행히, UI element와 관련된 Publisher와 같이 일반적으로 많이 사용되는 Publisher는

관리가 가능할 속도로 element를 전달합니다.

다른 일반 Publisher은 URL 로딩시스템의 URLSession.DataTaskPublisher 처럼

오직 하나의 element만 생성합니다.

이 Publisher들에 Subscriber을 sink 하거나 assign 하는 것은 완벽히 안전합니다.

 

 

 

Apply Back Pressure with a Custom Subscriber

Publisher가 Subscriber에게 element를 보내는 속도를 제어하려면

Subscriber 프로토콜의 사용자 정의 구현을 만들어야합니다.

구현을 통해 Subscriber가 충족할 수 있는 요구사항을 지정합니다.

Subscriber가 element를 수신하면 receive(_:)을 통해 수신할 새 요구 값을 리턴해서 더 요청할 수도 있고,

구독의 request(_:) 메서드를 호출해서 더 요청할 수도 있습니다.

그러면 Subscriber은 Publisher가 언제든지 보낼 수 있는 element의 수를 미세 조정할 수 있습니다.

 

Subscriber가 element를 수신할 준비가 되어있음을 신호를 줌으로써 흐름을 제어하는

이러한 개념을 back pressure라고 합니다.

 

각 Publisher은 현재 만족하지 못한 요구를 추적합니다.

즉, Subscriber가 요청한 element의 수를 의미합니다.

Foundation의 Timer.TimerPublisher와 같이

보류중인 요구사항이 있을 때에만 element를 제공하는 자동화된 소스도 있습니다.

해당 예시 코드는 이 동작을 보여줍니다.

 

// Publisher: Uses a timer to emit the date once per second.
let timerPub = Timer.publish(every: 1, on: .main, in: .default)
    .autoconnect()


// Subscriber: Waits 5 seconds after subscription, then requests a
// maximum of 3 values.
class MySubscriber: Subscriber {
    typealias Input = Date
    typealias Failure = Never
    var subscription: Subscription?
    
    func receive(subscription: Subscription) {
        print("published                             received")
        self.subscription = subscription
        DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
            subscription.request(.max(3))
        }
    }
    
    func receive(_ input: Date) -> Subscribers.Demand {
        print("\(input)             \(Date())")
        return Subscribers.Demand.none
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print ("--done--")
    }
}


// Subscribe to timerPub.
let mySub = MySubscriber()
print ("Subscribing at \(Date())")
timerPub.subscribe(mySub)

 

Subscriber의 receive(subscription:)의 구현부분은

Publisher에게 element를 요청하기 전에 5초의 지연시간을 부여합니다.

이 기간 동안 Publisher은 존재하고, 또 유효한 Subscriber을 가지고 있지만

수요가 0이므로 element를 생성하지 않습니다.

다음 출력에서 볼 수 있듯이 시간 지연이 만료되고

Subscriber가 0이 아닌 요구를 제공한 후에만 element 전달을 시작합니다.

 

Subscribing at 2019-12-09 18:57:06 +0000
published                             received
2019-12-09 18:57:11 +0000             2019-12-09 18:57:11 +0000
2019-12-09 18:57:12 +0000             2019-12-09 18:57:12 +0000
2019-12-09 18:57:13 +0000             2019-12-09 18:57:13 +0000

 

이 예시에서는 5초 지연이 만료된 후 세개의 element만을 요구하는 수요를 방출합니다.

결과적으로 Publisher은 세 번째 이후에 추가 element를 더 보내지 않으며

Subscribers.Completion.finished 값을 보내지만 publishing은 끝내지 않습니다.

왜냐하면 Publisher은 element 수요를 더 기다리기 때문이죠.

element를 계속 수신하기 위해 Subscriber은 구독을 저장하고 주기적으로 추가 요소를 요청할 수 있습니다.

또한 receive(_:) 구현부분의 리턴값으로 새로운 수요를 지정할 수 있습니다.

 

 

 

Manage Unlimited Demand by Using Back-pressure Operators

커스텀 Subscriber를 작성하지 않아도, 우리는 Combine의 buffering 혹은 타이밍 연산자를 사용해서

back pressure를 적용할 수 있습니다:

 

• buffer(size: prefetch: whenFull:) :
  업스트림 Publisher로부터 아이템의 수를 고정합니다. 
  가득 차면 버퍼가 element를 삭제하거나 에러를 발생시킵니다.
• debounce(for: scheduler: options:) :
  업스트림 Publisher가 지정된 간격 동안 전달을 중지한 경우에만 게시합니다.
  (예를들어 searchBar에 텍스트를 입력할 때마다 api 요청을 하는 것이 아니라,
  사용자가 텍스트 입력을 멈췄을 때 요청할 때 쓰일 수 있겠네요)
• throttle(for: scheduler: latest:) :
  지정된 최대 속도로 element를 생성합니다.
  지정 간격 동안 여러개의 element를 수신하면 최신 또는 가장 오래된 element만 보냅니다.
• collect(_:), collect(_:options:) :
  지정된 카운트 또는 지정된 시간 간격을 초과할 때까지
  element들을 묶어두었다가, 우리에게 element 배열을 보냅니다.
  이 옵션은 Subscriber가 동시에 여러 element를 처리할 수 있는 경우에 유용합니다.