[Combine] Subscriber 사용을 위해 미리 정의된 것들 - Combine 공부 5

안녕하세요 Pingu입니다.🐧

 

지난 글에서는 Combine을 사용하지 않던 코드에서 간단하게 Combine을 적용하고 싶을 때 사용하면 좋은 Subject를 알아봤습니다.

이번 글에서는 여기서 알아본 미리 정의된 Publisher와 같이 미리 정의된 Subscriber에는 뭐가 있는지 살펴보려고 합니다.

 

Subscriber란?

간단하게 Subscriber가 뭔지 다시 알아볼게요.

정의부터 살펴보면 다음과 같았습니다.

즉 간단하게 말해서 Publisher에게 값을 받기 위해 선언해둔 프로토콜이라고 할 수 있습니다.

 

그리고 구현은 다음과 같이 되어있었어요.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
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>
}

 

• Input
  • Publisher에게 받는 값의 타입입니다.
• Failure
  • Publisher에게 받는 Error 타입입니다.
  • 만약 Error를 수신하지 않고 싶다면 Never 타입으로 설정해주면 됩니다!
• receive(subscription:)
  • Publisher가 만들어서 주는 subscription을 받습니다.
• receive(input:)
  • Publisher가 주는 값을 받습니다.
  • Demand를 반환하는데 이는 값을 더 원하는지에 대한 여부입니다.
• receive(completion:)
  • Publisher가 주는 completion event를 받습니다.

 

여기서 중요한 것은 값을 받으려는 Publisher의 <Output, Failure>의 타입이 Subscriber의 <Input, Failure> 타입과 일치해야 한다는 점이었습니다.

 

그리고 여기 보면 receive(input:) 메서드의 반환 타입이 Subscribers.Demand였고, 여기 나오는 Subscribers에 있는 애들이 바로 이번 글에서 알아볼 미리 정의된 Subscriber들입니다.

 

그리고 그 종류는 다음과 같아요.

• Demand
• Completion
• Sink
• Assign

그럼 하나씩 자세히 알아보도록 하겠습니다.

 

Sink (class)

이전 글의 예제에서도 자주 보였던 Sink부터 알아보도록 하겠습니다.

정의를 먼저 보면 다음과 같아요.

정의를 보면 Sink는 횟수의 제한 없이 Subscription을 통해 값을 요청하는 간단한 Subscriber라고 합니다.

 

구현을 살펴보면 다음과 같습니다.

final public class Sink<Input, Failure> : Subscriber, Cancellable, CustomStringConvertible,
                                            CustomReflectable, CustomPlaygroundDisplayConvertible
where Failure : Error {
    final public var receiveValue: (Input) -> Void { get }
    final public var receiveCompletion: (Subscribers.Completion<Failure>) -> Void { get }
    
    final public var description: String { get }
    final public var customMirror: Mirror { get }
    final public var playgroundDescription: Any { get }
    
    public init(receiveCompletion: @escaping ((Subscribers.Completion<Failure>) -> Void), 
                receiveValue: @escaping ((Input) -> Void))
    
    final public func cancel()
    
    // Subscriber 프로토콜의 필수요소들
    final public func receive(subscription: Subscription)
    final public func receive(_ value: Input) -> Subscribers.Demand
    final public func receive(completion: Subscribers.Completion<Failure>)
}

• receiveValue
  • 값을 받았을 때 실행될 클로저입니다.
• receiveCompletion
  • Completion을 받았을 때 실행될 클로저입니다.
• description
• customMirror
  • 모든 타입의 인스턴스에 대한 하위 구조 및 표시되는 스타일을 나타내는 Mirror를 커스텀할 수 있습니다.
• cancel
  • 말 그대로 subscription을 취소합니다.
  • Cancellable 프로토콜을 채택하기 때문에 구현되어있습니다.

여기서 실제로 주로 사용하는 것은 receiveValue, receiveCompletion 정도입니다.

 

실제로 사용해보면 다음과 같이 사용할 수 있습니다.

let intArrayPublisher = [1,2,3,4,5].publisher
        
let sink = Subscribers.Sink<Int, Never>(receiveCompletion: { print("completion: \($0)") },
                                        receiveValue: { print("value: \($0)")})

intArrayPublisher.subscribe(sink)

근데 보통은 위와 같이 사용하기보다는 아래와 같이 사용합니다.

let intArrayPublisher = [1,2,3,4,5].publisher

intArrayPublisher
    .sink(receiveCompletion: { print("completion: \($0)") },
          receiveValue: { print("value: \($0)")})

첫 번째 코드는 sink를 직접 만들고 이를 Publisher의 subscribe로 전달했다면, 두 번째 코드는 Publisher의 extension에 구현된 sink Operator를 사용한 예입니다.

 

두 가지 예 모두 결과는 아래와 같아요.

value: 1
value: 2
value: 3
value: 4
value: 5
completion: finished

 

두 번째 예에서 사용한 sink Operator를 사용하면 첫 번째 예에서 만든 sink를 알아서 만들고 바로 subscribe 하게 됩니다.

직접 만들 필요가 없어서 사용이 편리하죠! 그리고 값도 알아서 바로 request 합니다.

 

description, playgroundDescription, customMirror는 어떻게 출력되는지 궁금해서 한 번 출력해봤는데, 결과는 다음과 같아요.

description: Sink
playgroundDescription: Sink
customMirror: Mirror for Sink<Int, Never>

 

이번에는 Sink의 cancel()을 사용해보겠습니다.

let subject = PassthroughSubject<Int, Never>()
let firstSink = Subscribers.Sink<Int, Never>(receiveCompletion: { print("first sink completion: \($0)") },
                                             receiveValue: { print("first sink value: \($0)")})
let secondSink = Subscribers.Sink<Int, Never>(receiveCompletion: { print("second sink completion: \($0)") },
                                              receiveValue: { print("second sink value: \($0)")})

subject.subscribe(firstSink)
subject.subscribe(secondSink)
subject.send(1)

// 첫 번째 sink 취소
firstSink.cancel()

subject.send(2)
subject.send(completion: .finished)

이렇게 하면 2는 firstSink에 전달되지 않을 것 같은데, 실제로도 그런지 확인해보면..

first sink value: 1
second sink value: 1
second sink value: 2
second sink completion: finished

값은 물론 completion도 전달되지 않은 것을 볼 수 있습니다.

Assign (class)

다음으로 Assign을 알아볼게요.

정의부터 보면 다음과 같습니다.

정의를 보면 Assign은 key path로 표시된 프로퍼티에 수신된 값을 할당하는 간단한 Subscriber라고 하네요.

 

구현을 간단히 살펴보면 다음과 같습니다.

final public class Assign<Root, Input> : Subscriber, Cancellable, CustomStringConvertible, 
CustomReflectable, CustomPlaygroundDisplayConvertible {
    public typealias Failure = Never
    final public var object: Root? { get }
    final public let keyPath: ReferenceWritableKeyPath<Root, Input>
    
    public init(object: Root, keyPath: ReferenceWritableKeyPath<Root, Input>)

    final public func receive(subscription: Subscription)
    final public func receive(_ value: Input) -> Subscribers.Demand
    final public func receive(completion: Subscribers.Completion<Never>)
    
    final public func cancel()
}

눈에 띄는 것은 두 개네요.

• object
  • 프로퍼티를 포함하는 객체라고 합니다. Subscriber는 새로운 값을 받을 때마다 여기에 할당한다고 하네요.
  • Subscriber는 upstream publisher가 Subscriber의 receive(completion:)을 호출할 때까지 object에 대한 강한 참조를 유지하고, 호출된 이후에야 nil로 설정된다고 합니다.
• keyPath
  • 할당할 프로퍼티를 나타내는 key-path입니다.

간단하게 말해서 어떤 값을 받아서 어떤 곳에 저장하는 Subscirber입니다.

 

직접 사용해보면 이해가 빠를거같아요.

class SampleObject {
    var intValue: Int {
        didSet {
            print("intValue Changed: \(intValue)")
        }
    }
    
    init(intValue: Int) {
        self.intValue = intValue
    }
    
    deinit {
        print("sample object deinit")
    }
}

let myObject = SampleObject(intValue: 5)

let assign = Subscribers.Assign<SampleObject, Int>(object: myObject, keyPath: \.intValue)

let intArrayPublisher = [6,7,8,9].publisher
intArrayPublisher.subscribe(assign)
print(myObject.intValue)

위와 같이 SampleObject라는 간단한 클래스를 하나 만들고 그 안에 intValue라는 프로퍼티를 만들어줍니다. 그리고 해당 프로퍼티에 프로퍼티 옵저버를 사용해서 값이 변할 때마다 현재 값이 print 되도록 만들어줍니다.

 

그런 뒤에 assign을 만들고 [6, 7, 8, 9] 배열의 값을 내보내는 publisher를 만든 후에 assign으로 subscribe 하면 다음과 같은 결과를 볼 수 있습니다.

intValue Changed: 6
intValue Changed: 7
intValue Changed: 8
intValue Changed: 9
9
sample object deinit

Publisher가 내보내는 값을 계속해서 할당하는 것을 볼 수 있습니다.

 

근데 이를 쉽게 사용하기 위해 Publisher에는 assign이라는 Operator가 구현되어 있습니다.

이걸 사용해서 동일한 동작을 하는 코드를 만들어 보면 다음과 같습니다.

class SampleObject {
    var intValue: Int {
        didSet {
            print("intValue Changed: \(intValue)")
        }
    }
    
    init(intValue: Int) {
        self.intValue = intValue
    }
    
    deinit {
        print("sample object deinit")
    }
}

let myObject = SampleObject(intValue: 5)

let intArrayPublisher = [6,7,8,9].publisher

intArrayPublisher
    .assign(to: \.intValue, on: myObject)
    
print(myObject.intValue)

실행해보면 결과는 같습니다.

 

그런데 아까 object를 강한 참조로 가지고 있다고 했잖아요?

따라서 강한 참조 주기가 발생하는 경우가 발생하기도 하는데요, 이건 개인적으로 좀 더 본 뒤에 다음 글에서 알아보도록 할게요.

Demand (struct)

다음으로 알아볼 것은 Demand입니다.

정의를 보면 Subscriber가 subscription을 통해 Publisher에게 요청한 item의 수입니다.

즉 몇 번이나 값을 요청했느냐에 대한 값이에요.

 

구현을 간단히 살펴보면 다음과 같습니다.

@frozen public struct Demand : Equatable, Comparable, Hashable, Codable, 
CustomStringConvertible {
    public static let unlimited: Subscribers.Demand
    public static let none: Subscribers.Demand
    @inlinable public static func max(_ value: Int) -> Subscribers.Demand
}

Subscriber는 receive(input:)의 반환 값으로 Demand를 반환한다고 했는데요, 반환된 Demand 값에 따라 값을 더 요청할지 그만 할지를 결정합니다.

 

그리고 구현부에 정의된 것을 살펴보면..

• unlimited
  • 계속해서 값을 받겠다는 의미
• none
  
  • max(0)과 같은 의미입니다. (요청을 추가하지 않는다는 뜻)
• max(_ value: Int)
  • 매개변수로 받은 값만큼 추가로 요청합니다.

Subscriber가 값을 요청할 횟수를 subscription의 request(_:)를 처음 호출할 때 정할 수도 있지만 receive(input:)에서 추가할 수도 있습니다. 그리고 횟수를 감소시키는 것은 불가능합니다. 예를 들어 max(-3)을 반환하면 fatalError가 발생합니다. 그리고 Demand는 누적되는 값입니다.

 

Demand는 Subscriber 프로토콜을 채택한 애는 아니고 그냥 Subscriber를 사용할 때 필요한 애라고 할 수 있는데요, 그래서 사용법도 앞서 본 sink, assign과 다릅니다. Subscriber를 구현해서 receive(input:)이나 receive(subscription:)에서 사용해줘야 해요.

class DemandTestSubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never
    
    func receive(subscription: Subscription) {
        print("subscribe 시작!")
        // 여기서 Demand를 설정해줄 수도 있어요!
        // 현재 요청횟수는 1
        subscription.request(.max(1))
    }
    
    func receive(_ input: Int) -> Subscribers.Demand {
        print("receive input: \(input)")
        
        // input 값이 2일때만 요청횟수를 1 추가합니다.
        if input == 2 {
            return .max(1)
        } else {
            return .none
        }
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print("receive completion: \(completion)")
    }
}

let publisher = [2, 3, 4, 5].publisher
publisher
    .print()
    .subscribe(DemandTestSubscriber())

위에서 구현한 DemandTestSubscriber의 경우 처음 subscribe 할 때 요청 횟수를 1로 설정하고 이후에는 수신한 값이 2일 때만 요청횟수를 1 증가시키게 됩니다. publisher에서 사용한 print() Operator는 해당 subscription에서 발생하는 동작을 보여주는 Operator입니다.

 

위 코드를 실행하면 subscriber는 다음과 같이 2개의 값만 받을 수 있게 되죠.

receive subscription: ([2, 3, 4, 5])
subscribe 시작!
request max: (1)
receive value: (2)
receive input: 2
request max: (1) (synchronous)
receive value: (3)
receive input: 3

 

receive(subscription:)에서 처음 subscription을 받을 때 Demand를 unlimited로 해두고 receive(input:)에서는 항상 .none을 반환하면 어떻게 될까요?

class DemandTestSubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = Never
    
    func receive(subscription: Subscription) {
        print("subscribe 시작!")
        // 여기서 Demand를 설정해줄 수도 있어요!
        subscription.request(.unlimited)
    }
    
    func receive(_ input: Int) -> Subscribers.Demand {
        return .none
    }
    
    func receive(completion: Subscribers.Completion<Never>) {
        print("receive completion: \(completion)")
    }
}

let publisher = [2, 3, 4, 5].publisher
publisher
    .print()
    .subscribe(DemandTestSubscriber())

결과는 다음과 같습니다.

receive subscription: ([2, 3, 4, 5])
subscribe 시작!
request unlimited
receive value: (2)
receive value: (3)
receive value: (4)
receive value: (5)
receive finished
receive completion: finished

처음에 계속해서 값을 요청하겠다고 설정했으니 이후에 항상 Demand로 none을 반환해도 모든 값을 수신할 수 있는 거죠.

 

간단하죠?

Demand는 누적되는 값이다, 음수를 넣어서 감소시킬 수는 없다! 정도만 알면 사용할 때 큰 문제는 없겠어요.

Completion (enum)

다음으로 알아볼 것은 Completion입니다. 얘도 Demand와 동일하게 Subscriber는 아니고 Subscriber를 사용할 때 필요한 녀석인데요, 아까 DemandTestSubscriber를 직접 구현할 때 receive(completion:)에서 썼었습니다 ㅎㅎ

 

어쨌든 정의를 보면 다음과 같아요!

정의를 보니 정상적인 완료 혹은 에러로 인해 Publisher가 값을 더 이상 생성하지 않는다는 신호! 라네요.

 

바로 구현도 보겠습니다.

@frozen public enum Completion<Failure> where Failure : Error {
    case finished
    case failure(Failure)
}

간단하네요. finished는 정상적인 완료일 때, failure가 실패일 때 사용됩니다.

지금까지는 모두 Failure타입을 Never로 사용해서 에러가 없었는데, 에러가 있는 Subscriber를 한 번 만들어볼게요.

// custom Error를 만듭니다.
enum PinguError: Error {
    case pinguIsBaboo
}

class PinguSubscriber: Subscriber {
    typealias Input = Int
    typealias Failure = PinguError
    
    func receive(subscription: Subscription) {
        subscription.request(.unlimited)
    }
    
    func receive(_ input: Int) -> Subscribers.Demand {
        print("receive input: \(input)")
        return .none
    }
    
    func receive(completion: Subscribers.Completion<PinguError>) {
        // .pinguIsBaboo 수신시 실행
        if completion == .failure(.pinguIsBaboo) {
            print("Pingu는 바보입니다.")
        } else {
            print("finished!")
        }
    }
}

let subject = PassthroughSubject<Int, PinguError>()
let subscriber = PinguSubscriber()

subject.subscribe(subscriber)

subject.send(100)
subject.send(completion: .failure(.pinguIsBaboo))
subject.send(200)

위와 같이 Error를 하나 구현하고 Subscriber의 Failure 타입을 해당 에러로 설정하면 됩니다.

그러면 해당 Subscriber가 특정 에러를 받을 때 원하는 작업을 실행하도록 만들 수 있습니다.

위 코드를 실행해보면 아래와 같은 결과를 볼 수 있을 거예요.

receive input: 100
Pingu는 바보입니다.

Subscriber가 failure completion을 받았기 때문에 값이나 completion을 받지 않으므로 마지막에 전달한 200은 전달되지 않는 것을 볼 수 있어요.

 

이걸 잘 활용하면 Combine을 사용하면서 에러 처리를 잘할 수 있을 거 같네요.

AnySubscriber

마지막으로 AnySubscriber를 알아보겠습니다.

Publisher를 공부할 때도 AnyPublisher가 있었는데요, 얘도 비슷합니다.

정의는 위와 같고 말 그대로 어떤 subscriber의 타입을 간단하게 사용할 수 있도록 래핑 합니다.

 

사용은 아까 구현한 PinguSubscriber를 재활용해서 사용해볼게요.

let pinguSubscriber = PinguSubscriber()
let anySubscriber = AnySubscriber(pinguSubscriber)

이렇게 하면 pinguSubscriber의 타입은 아래와 같은데,

anySubscriber의 타입은 AnySubscriber로 래핑 된 것을 볼 수 있습니다.

 

이렇게 Apple에서 미리 만들어둔 Subscriber인 Sink, Assign과 Subscriber 사용을 위해 필요한 Demand, Completion에 대해 알아봤습니다. 

 

다음 글에서는 Subscription에 대해 알아보도록 하겠습니다.

 

전체 코드는 여기에서 볼 수 있습니다.

 

감사합니다.