Controlling Publishing with Connectable Publishers

Coordinate when publishers start sending elements to subscribers.

퍼블리셔가 구독자에게 요소를 보낼 때 이를 조율합니다.

Overview

간혹 퍼블리셔가 동작에 영향을 미칠 때처럼 퍼블리셔가 요소 제공을 시작하기 전에 이 퍼블리셔를 조작하길 원할 수 있습니다. 하지만 흔히 사용되는 sink(receiveValue:)와 같은 구독자는 즉시 제한없는 요소를 요구하며, 이 경우 퍼블리셔가 원하는대로 동작하는 것을 막을 수 있습니다. 구독자가 준비되기 전에 값을 제공하는 퍼블리셔는 구독자가 둘 혹은 그보다 많을 때 문제가 생길 수 있습니다. 여러 구독자가 있는 시나리오는 레이스 조건을 생성합니다. 두 번쨰 구독자가 있다고 하더라도 퍼블리셔는 첫 번째 구독자에게 요소를 전달할 수 있습니다.

아래 이미지에 해당하는 시나리오를 생각해볼 수 있습니다. URLSession.DataTaskPublisher를 생성하고, URL 데이터를 받는 sink 구독자(구독자 1)가 데이터를 불러오게 됩니다. 이후 두 번째 구독자(구독자 2)를 붙일 수도 있습니다. 만약 두 번째 구독자가 붙기 전에 데이터 테스크가 다운로드를 완료되면 두 번째 구독자는 데이터를 놓치고 오직 컴플리션만 볼 수 있습니다.

Figure 1

Hold Publishing by Using a Connectable Publisher

퍼블리셔가 조건이 준비되기 전에 요소 전송하는 것을 방지하기 위해 컴바인은 ConnectablePublisher 프로토콜을 제공합니다. 연결 가능한 퍼블리셔는 connect() 메소드를 호출하기 전까지 요소를 제공하지 않습니다. 요소 제공이 준비가 되어 있고 충족되지 않은 요구가 있다고 하더라도 연결 가능한 퍼블리셔는 구독자가 명시적으로 connect()을 호출하기 전까지 어떤 요소도 전달하지 않습니다.

아래 그림은 위에서 설명한 URLSession.DataTaskPublisher 시나리오를 보여주면서도 구독자 앞에 ConnectablePublisher가 있다는 차이를 보여줍니다. 두 구독자가 connect()를 호출하기 전까지 기다리기만 하고, 데이터 테스크는 다운로드를 시작하지 않습니다. 이 경우 레이스 조건을 제거하고, 모든 구독자가 데이터를 받을 수 있음을 보장합니다.

Figure 2

ConnectablePublisher를 사용하려면 Publishers.MakeConnectable 인스턴스로 기존 퍼블리셔를 감싸기 위해 makeConnectable() 오퍼레이터를 사용하면 됩니다. 아래 코드는 위에서 설명한 것처럼 데이터 테스크 퍼블리셔의 레이스 조건을 수정하고 있으며, makeConnectable()이 그 역할을 하고 있습니다. 보통 AnyCancellable(cancellable 1)을 반환하는 sink를 붙이면 데이터 테스크가 즉시 시작됩니다. 이 시나리오의 경우 두 번째 sink, 즉 AnyCancellable(cancellable 2)를 반환하는 구독은 1초 후에도 붙지 않고, 데이터 테스크 퍼블리셔는 두 번째 sink가 붙기 전에 완료될 것입니다. 대신 명시적으로 ConnectablePublisher를 사용하면 데이터 테스크가 오직 앱이 2초 딜레이와 함께 connect()를 호출한 이후에만 시작될 것입니다.

let url = URL(string: "https://example.com/")!
let connectable = URLSession.shared
    .dataTaskPublisher(for: url)
    .map() { $0.data }
    .catch() { _ in Just(Data() )}
    .share()
    .makeConnectable()


cancellable1 = connectable
    .sink(receiveCompletion: { print("Received completion 1: \($0).") },
          receiveValue: { print("Received data 1: \($0.count) bytes.") })


DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
    self.cancellable2 = connectable
        .sink(receiveCompletion: { print("Received completion 2: \($0).") },
              receiveValue: { print("Received data 2: \($0.count) bytes.") })
}


DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    self.connection = connectable.connect()
}

Important
connect()는 유지시켜줘야 하는 Cancellable 인스턴스를 반환합니다. 퍼블리싱을 취소하기 위해 이 인스턴스를 사용할 수 있으며, 명시적으로 cancel() 메소드를 호출할 수 있고 인스턴스 해제를 할 수도 있습니다.

Use the Autoconnect Operator If You Don’t Need to Explicitly Connect

Publishers.Multicast,Timer.TimerPublisher와 같은 몇 가지 컴바인 퍼블리셔는 이미 ConnectablePublisher를 구현하고 있습니다. 이와 같은 퍼블리셔를 사용하는 것은 반대의 문제를 발생시킬 수 있습니다. 퍼블리셔를 설정할 필요가 없거나 여러 구독자를 붙이는 경우 명시적으로 connect()를 호출하면 부담이 될 것입니다.

이와 같은 경우를 위해 ConnectablePublisher는 autoconnect() 오퍼레이터를 제공합니다. 이 오퍼레이터는 구독자가 subscribe(_:)로 퍼블리셔에 붙을 때 즉시 connect()를 호출합니다.

아래 예시는 autoconnect()를 사용함으로써 1초마다 한 번 발생(once-a-second)하는 Timer.TimerPublisher로부터 즉시 요소를 수신합니다. autoconnect()가 없는 예시의 경우 특정 시점에 connect()를 호출해서 타이머 퍼블리셔를 시작해야 할 필요가 있습니다.

let cancellable = Timer.publish(every: 1, on: .main, in: .default)
    .autoconnect()
    .sink() { date in
        print ("Date now: \(date)")
     }