Swift: Combine(2) - Publisher, Subscriber

Publisher

Publisher: Declares that a type can transmit a sequence of values over time.(유형이 시간이 지남에 따라 일련의 값을 전송할 수 있다고 선언합니다.)

 

protocol Publisher<Output, Failure>

protocol Publisher {
    associatedtype Output
    associatedtype Failure : Error
    func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

 

Publisher는 하나 이상의 Subcriber 인스턴스에 요소(값)를 전달합니다.

Subscriber의 Input, Failure 타입은 Publisher가 선언한 Output, Failure 타입과 일치해야 하고, Publisher는 Subscriber를 수락하기 위해 receive(subcriber:) 메서드를 구현해야 합니다.

 

Publisher는 Subcriber의 메소드들을 호출할 수 있어요.

• receive(subscription:): subcribe 요청을 수락하고, Subscription 인스턴스를 반환합니다. Subscriber는 Subscription을 사용하여 Publisher의 요소를 요구하고, Publishing을 취소합니다.
• receive(_:): Publisher로부터 Subscriber에게 하나의 요소를 전달합니다.
• receive(completion:): Subcriber에게 Publishing이 정상 또는 오류와 함께 종료되었음을 알립니다.

 

모든 Publisher는 다운스트림(downstream) Subscriber가 올바르게 작동하기 위해 위 방법을 준수해야 합니다.

Publisher의 Extension을 통해 정교한 이벤트 처리 체인을 만들기 위한 운영자(operator)를 정의할 수 있고, 각 운영자는 Publisher 프로토콜을 구현하는 타입을 반환합니다.

이러한 타입의 대부분은 Publishers 열거형의 Extension으로 존재하는데 예를 들어, map(_:) 연산자는 Publishers.Map의 인스턴스를 반환합니다.

 

 

Publisher 구현

Publisher 프로토콜은 직접 구현하는 대신, Combine 프레임워크에서 제공하는 타입 중 하나를 사용하기를 권장합니다.

• Subject를 채택한 클래스에서 send(_:) 메소드를 호출
  • CurrentValueSubject: 현재 값을 감싸고 있다가 값이 바뀌면 Publish 하는 Subject
  • PassthroughSubject: downstream Subscriber에게 broadcast 하는 Subject
• Convenience Publishers
  • class Future: 단일 값을 생산한 뒤 완료 또는 실패하는 Publisher
  • struct Just: 각 Subscriber에게 단 한 번 출력한 뒤 완료하는 Publisher
  • struct Deferred: 새로운 Subscriber를 위한 Publisher를 생성하기 위해서 subscription을 기다리는 Publisher(= subscription이 이루어질 때 Publisher를 생성)
  • struct Empty: 어떤 값도 publish하지 않고 선택적으로 즉시 종료되는 Publisher
  • struct Fail: 지정된 오류로 즉시 종료되는 Publisher
  • struct Record: 각 Subscriber에게 나중에 재생(playback)할 수 있도록 일련의 입력과 완료를 녹음(recording)할 수 있는 Publisher
• @Published 어노테이션을 사용해서 값이 바뀔 때 마다 이벤트를 방출(emit)하는 Publisher 구현

 

 

Subject

let passthroughSubject = PassthroughSubject<Int, Never>()
print("PassthroughSubject Publish")

passthroughSubject.sink {
    print("PassthroughSubject \\($0)")
} receiveValue: {
    print($0)
}
passthroughSubject.send(2)
passthroughSubject.send(3)
passthroughSubject.send(completion: .finished)

// PassthroughSubject Publish
// 2
// 3
// PassthroughSubject finished

let currentValueSubject = CurrentValueSubject<Int, Never>(1)
print("CurrentValueSubject Publish")

currentValueSubject.sink {
    print("CurrentValueSubject \\($0)")
} receiveValue: {
    print($0)
}
currentValueSubject.send(2)
currentValueSubject.send(3)
currentValueSubject.send(completion: .finished)

// CurrentValueSubject Publish
// 1
// 2
// 3
// CurrentValueSubject finished

 

 

@Published

class Person {
    @Published var age = 25
    
    func increaseAge() {
        age += 1
    }
}

let person = Person()

_ = person.$age.sink {
    print("age: \\($0)")
}

person.increaseAge()
person.increaseAge()
person.increaseAge()

// age: 25
// age: 26
// age: 27
// age: 28

 

Subscriber

Subscriber: A protocol that declares a type that can receive input from a publisher.(게시자로부터 입력을 받을 수 있는 유형을 선언하는 프로토콜.)

protocol Subscriber<Input, Failure> : CustomCombineIdentifierConvertible

public protocol Subscriber : CustomCombineIdentifierConvertible {
    associatedtype Input
    associatedtype Failure : Error

    func receive(subscription: Subscription)
    func receive(_ input: Self.Input) -> Subscribers.Demand
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

 

 

Subscriber 인스턴스는 Publisher로부터 요소를 전달받으며, Publisher의 <Output, Failure>는 Subscriber의 <Input, Failure>와 일치해야 합니다.

 

 

Subscriber는 요소를 요구하기 위해 Publisher의 subscribe(:) 메서드를 호출해서 연결하고,

이후 Subscriber의 receive(subscription:) 메소드를 통해 정상적으로 연결되었음을 선언한 뒤,

request(_:) 메소드로 필요한 요소에 대해 요구하면,

receive(_:) 메소드로 요구에 맞춰 요소를 전달받고,

전달이 모두 끝나면 receive(completion:) 메서드를 통해 전달에 대한 완료 또는 오류 여부를 전달합니다.

 

 

Combine은 Publisher 타입의 operator로 sink(receiveCompletion:receiveValue:), assign(to:on:)과 같은 Subscriber를 제공합니다.

 

• sink(receiveCompletion:receiveValue:): 완료 신호를 받고, 새로운 요소를 받을 때마다 임의의 클로저를 실행
• assign(to:on:): 주어진 인스턴스의 키 경로로 식별된 속성에 새로 받은 각 값을 기록

 

사용 예

let myPub = [1, 2, 3].publisher

class MySubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never
    
    func receive(subscription: Subscription) {
        subscription.request(.max(1)) // 1
    }
    
    func receive(_ input: Int) -> Subscribers.Demand {
        print(input)
        return Subscribers.Demand.max(1) // 2
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print ("Publish 완료")
    }
}

let mySub = MySubscriber()
print("Publish 시작")
myPub.subscribe(mySub)

// Publish 시작
// 1
// 2
// 3
// Publish 완료

• #1 - 요소 1개 요청
• #2 - 요소 1개를 더 요청

 

 

receive(_ input: Int)의 결과 값을 통해 요소에 대한 제어를 할 수 있습니다.

let myPub = [1, 2, 3].publisher

class MySubscriber: Subscriber {
    // ...
    
    func receive(_ input: Int) -> Subscribers.Demand {
        print(input)
        return input == 2 ? Subscribers.Demand.none : Subscribers.Demand.max(1)
    }
    
    // ...
}

let mySub = MySubscriber()
print("Publish 시작")
myPub.subscribe(mySub)

// Publish 시작
// 1
// 2

 

 

또는 sink를 통해 간단하게 사용할 수 있습니다.

let myPub = [1, 2, 3].publisher

print("Publish 시작")
myPub.sink { _ in
    print("Publish 완료")
} receiveValue: {
    print($0)
}
// Publish 시작
// 1
// 2
// 3
// Publish 완료

 

 

---

참고

Swift Combine, 시작하기

[Swift] Combine 입문가하기1 — 구성요소

개린이의 관점에서 Combine 살펴보기(2) - Publisher

Processing Published Elements with Subscribers