[Combine] 미리 정의된 Publisher들 - Combine 공부 3

안녕하세요 Pingu입니다.🐧

 

지난 글에서는 Publisher, Subscriber 프로토콜 그 자체에 대해 알아봤었는데요, 이번 글에서는 Publisher 프로토콜로 Apple에서 미리 구현한 Publisher들을 알아보려고 합니다.

 

간단히 Publisher가 뭔지 짚어보면, Subscription을 만들고 Subscriber에게 값과 completion event를 내보내는 타입을 위한 프로토콜이었습니다.

 

먼저 Apple에서 미리 구현한 Publisher들은 아래와 같습니다.

• Just
• Future
• Deferred
• Empty
• Fail
• Record
• AnyPublisher

그럼 하나씩 차례대로 알아보겠습니다~

 

Just (Struct)

가장 간단한 Publisher로 자신을 subscribe 하는 Subscriber들에게 한 번에 값을 내보낸 뒤 finish 이벤트를 보내는 Publisher입니다.

사용법도 간단합니다.

let just = Just("This is Output")
just
    .sink(
        receiveCompletion: { completion in
            print("received completion: \(completion)")
        },
        receiveValue: { value in
            print("received value: \(value)")
        })

이렇게 하면 결과는 아래와 같이 나옵니다.

received value: This is Output
received completion: finished

정의된 대로 자신을 subscribe 하는 Subscriber에게 값을 내보낸 뒤에 finish 이벤트를 보내네요.

 

값을 한 번에 보내고 finish 이벤트를 보내는 것이 Just의 특징이라고 할 수 있습니다.

그런데 Publisher 프로토콜은 Output, Failure 프로퍼티가 필요하다고 했었는데 Just에는 왜 따로 Failure타입을 설정하지 않았을까요?

이건 Just의 구현을 보면 쉽게 이해가 됩니다.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public struct Just<Output> : Publisher {
    public typealias Failure = Never
    public let output: Output
    
    public init(_ output: Output)
    public func receive<S>(subscriber: S) where Output == S.Input, S : Subscriber, S.Failure == Just<Output>.Failure
}

이렇게 위와 같이 Failure 타입이 Never로 설정되어있기 때문입니다.

 

그럼 Never는 뭐죠?

정의는 "정상적으로 리턴하지 않는 함수의 리턴 타입, 값이 없는 타입"이라고 합니다.

Combine에서는 Publisher가 오류를 생성하지 않는 경우 Never 타입으로 설정합니다.

이렇게 Failure타입이 Never일 때만 사용할 수 있는 Operator도 있는데 이건 나중에 정리해보도록 하겠습니다.

 

Future (Class)

다음으로 알아볼 것은 Future입니다. 정의부터 봐야겠죠?

Future는 하나의 결과를 비동기로 생성한 뒤 completion event를 보냅니다.

여기서 중요한 것은 하나의 결과를 비동기로 생성한 뒤에 subscribe되는 

Future의 구현을 보면 비동기로 처리된다는 게 느껴집니다.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
final public class Future<Output, Failure> : Publisher where Failure : Error {
    public typealias Promise = (Result<Output, Failure>) -> Void
	
    public init(_ attemptToFulfill: @escaping (@escaping Future<Output, Failure>.Promise) -> Void)
    
    final public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
}

• Promise는 Future가 값을 내보낼 때 호출되는 클로저입니다.
• init에서 Promise 클로저를 매개변수로 받습니다.

즉 Future는 생성할 때 값을 내보낼 때 호출할 클로저를 매개변수로 받아서 값을 한 번 내보내면 해당 값을 계속 내보내는 Publisher입니다.

 

그리고 구현이 Class인 것도 확인할 수 있습니다. 이번 글에서 배울 다른 Publisher들이 모두 Struct인데 얘만 Class인 이유가 궁금해서 좀 찾아보니 비동기로 작동할 때 상태 저장 동작을 가능하게 하기 위해 Class로 구현되었다고 하네요.

 

그럼 간단히 사용해볼게요.

var subscriptions = Set<AnyCancellable>()
var emitValue: Int = 0
var delay: TimeInterval = 3

func createFuture() -> Future<Int, Never> {
    return Future<Int, Never> { promise in
        delay -= 1
        DispatchQueue.main.asyncAfter(deadline: .now() + delay) {
            emitValue += 1
            promise(.success(emitValue))
        }
    }
}

let firstFuture = createFuture()
let secondFuture = createFuture()
let thirdFuture = createFuture()

firstFuture
    .sink(receiveCompletion: { print("첫번째 Future Completion: \($0)") },
          receiveValue: { print("첫번째 Future value: \($0)") })
    .store(in: &subscriptions)

secondFuture
    .sink(receiveCompletion: { print("두번째 Future completion: \($0)") },
          receiveValue: { print("두번째 Future value: \($0)") })
    .store(in: &subscriptions)

thirdFuture
    .sink(receiveCompletion: { print("세번째 Future completion: \($0)") },
          receiveValue: { print("세번째 Future value: \($0)") })
    .store(in: &subscriptions)

thirdFuture
    .sink(receiveCompletion: { print("세번째 Future completion2: \($0)") },
          receiveValue: { print("세번째 Future value2: \($0)") })
    .store(in: &subscriptions)

위와 같이 코드를 만들면 결과는 아래와 같아요.

세번째 Future value2: 1
세번째 Future completion2: finished
세번째 Future value: 1
세번째 Future completion: finished
두번째 Future value: 2
두번째 Future completion: finished
첫번째 Future value: 3
첫번째 Future Completion: finished

세 번째 Future가 가장 먼저 완료되고, 두 번째, 첫 번째 순으로 완료되는 것을 볼 수 있습니다.

이렇게 결과가 나오는 이유는 sink로 만든 각각의 Future에서 delay를 1초씩 감소시켰기 때문입니다.

 

첫번째 Future는 3초 대기를 하다가 Promise 클로저가 호출되지만

두 번째 Future는 1초 줄어든 2초 대기를 하다가 Promise 클로저가 호출되고,

세 번째 Future는 1초가 더 줄어든 1초 대기를 하다가 Promise 클로저가 호출되어 위와 같은 결과가 발생한 거죠!

 

그리고 세 번째 Future에는 두 번의 subscribe를 했는데요, 결과가 똑같이 나오는 것을 볼 수 있습니다. 즉 한 번 방출된 값을 계속해서 사용하는 것도 알 수 있습니다.

 

이렇게 비동기로 결과를 처리할 수 있는 게 Future입니다.

 

Deferred (Struct)

Deferred는 새로운 Subscriber의 Publisher를 만들기 위해 제공된 클로저를 실행하기 전에 Subscription을 기다리는 Publisher라고 하네요.

 

Deferred가 번역하면 "지연된"이라는 의미인데

 

이번에도 Deferred의 구현을 보면서 알아보겠습니다.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public struct Deferred<DeferredPublisher> : Publisher where DeferredPublisher : Publisher {

    public typealias Output = DeferredPublisher.Output
    public typealias Failure = DeferredPublisher.Failure
    
    public let createPublisher: () -> DeferredPublisher
    public init(createPublisher: @escaping () -> DeferredPublisher)
    
    public func receive<S>(subscriber: S) where S : Subscriber, DeferredPublisher.Failure == S.Failure, DeferredPublisher.Output == S.Input
}

뭐 다른 것들은 모두 동일한데 createPublisher 클로저와 생성자가 좀 다르네요.

• createPublisher는 Publisher가 subscribe 됐을 때 실행할 클로저입니다.
• init에서 받은 클로저는 subscribe(_:)가 호출될 때 실행됩니다.

실제로 사용해보겠습니다.

struct PinguPublisher: Publisher {
    typealias Output = String
    typealias Failure = Never
    
    func receive<S>(subscriber: S) where S : Subscriber, Never == S.Failure, String == S.Input {
        subscriber.receive("안녕 나는 pinguPublisher")
        subscriber.receive(completion: .finished)
    }
}

print("deferred publisher가 만들어짐")
let deferred = Deferred { () -> PinguPublisher in
    print("pinguPublisher가 만들어짐\n")
    return PinguPublisher()
}

deferred
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })

실행해보면 결과는 아래와 같아요.

deferred publisher가 만들어짐
pinguPublisher가 만들어짐

안녕 나는 pinguPublisher
finished

결과를 보면 Deferred가 만들어졌을 때 PinguPublisher는 만들어지지 않았고, 이후에 sink를 통해 subscribe 했을 때 Deferred를 생성할 때 구현한 클로저에서 만들어지는 것을 볼 수 있습니다.

 

Swift의 lazy와 비슷하게 Publisher가 실제로 사용될 때 Publisher를 생성해서 사용해서 메모리를 효율적으로 사용할 수 있어 보입니다.

 

AnyPublisher

다음 Publisher들을 알아보려면 AnyPublisher를 알아야 될 거 같아서 지금 알아보도록 하겠습니다.

정의를 읽어보면 AnyPublisher는 자체적으로 뭐 중요한 건 없고 Upstream Publisher의 값들을 전달하는 Publisher입니다.

어떤 Publisher에서도 EraseToAnyPublisher()를 호출하면 AnyPublisher로 래핑 됩니다.

 

그럼 한 번 사용해보겠습니다.

let originalPublisher = [1, nil, 3].publisher

let anyPublisher = originalPublisher.eraseToAnyPublisher()
anyPublisher.sink { value in
    print(value)
}

위와 같이 하면 originalPublisher의 타입은 

이렇게 되는데, 이걸 eraseToAnyPublisher로 AnyPublisher로 래핑 한 anyPublisher의 타입은

이렇게 됩니다.

 

이러한 AnyPublisher는 Combine을 사용하다 보면 여러 Operator를 사용하면서 여러 Publisher 타입이 생성될 수 있는데 이걸 간단하게 처리하기 위해서 사용합니다.

어떻게 복잡해지는지는 Empty에서 알아보도록 하겠습니다.

 

Empty (Struct)

뭔가 Empty라는 단어에서 느껴지듯이 Empty는 아무런 값도 내보내지 않고 즉시 completion 이벤트를 보낼지 선택할 수 있는 Publisher입니다.

 

구현을 보면 정의가 좀 더 이해가 빠르게 됩니다.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public struct Empty<Output, Failure> : Publisher, Equatable where Failure : Error {
    public init(completeImmediately: Bool = true)
    public init(completeImmediately: Bool = true, outputType: Output.Type, failureType: Failure.Type)
    public let completeImmediately: Bool
    
    public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
}

init이 두 개니 차이점을 살펴보면

• init(completeImmediately: Bool = true)
  • completion 이벤트를 바로 보낼지만 결정합니다.
• init(completeImmediately: Bool = true, outputType: Output.Type, failureType: Failure.Type)
  • Empty를 Subscriber나 다른 Publisher에 연결할 때 사용합니다.
• completeImmediately는 Empty가 즉시 completion 되어야 하는지 여부를 나타냅니다.

이번에도 간단하게 사용해볼게요.

let empty = Empty<String, Never>()
empty
    .sink(receiveCompletion: { print("completion: \($0)") },
          receiveValue: { print("value: \($0)") }
    )

실행해보면 결과는 예상대로 그냥 completion만 출력됩니다.

completion: finished

 

근데 이렇게 말고 Subscriber나 다른 Publisher에 연결할 때 사용할 수도 있다고 했으니 그렇게도 한 번 사용해보겠습니다.

여기서 AnyPublisher도 사용합니다.

let anyPublisher = [1, nil, 3].publisher
    .flatMap { value -> AnyPublisher<Int, Never> in
        if let value = value {
            return Just(value).eraseToAnyPublisher()
        } else {
            return Empty().eraseToAnyPublisher()
        }
    }.eraseToAnyPublisher()


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

위와 같이 뭔가 조건에 맞게 Publisher를 처리하고 싶을 때 Empty를 사용할 수 있습니다.

실행 결과는 아래와 같아요.

value: 1
value: 3
AnyPublisher completion: finished

코드를 보면 nil을 거르고 싶은 코드라는 것을 알 수 있는데요, 여기서 값이 nil인 경우에는 Empty로 처리해서 DownStream에 보냅니다. 이렇게 하면 값을 빈 상태로 처리할 수 있어요. 따라서 결과에서도 nil은 안 나오고 1, 3만 나온 것을 볼 수 있습니다.

 

그리고 AnyPublisher를 공부할 때 Combine을 사용하다 보면 여러 타입의 Publisher를 처리하게 된다고 했는데 위에서도 Just, Empty의 두 가지 타입을 DownStream에 내려줘야 하는 상황이 생겼습니다. 이를 직접 처리하기보다는 AnyPublisher로 래핑 해서 DownStream으로 보내면 쉽게 처리할 수 있습니다.

 

Fail (Struct)

아까 Empty가 completion 이벤트를 즉시 보낼 수 있었다면 Fail은 Error를 즉시 보낼 수 있는 Publisher입니다.

 

구현을 보면..

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public struct Fail<Output, Failure> : Publisher where Failure : Error {
    public init(error: Failure)
    public init(outputType: Output.Type, failure: Failure)

    public let error: Failure
    public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
}

이렇게 즉시 보낼 error를 가지고 있는 것을 볼 수 있습니다.

 

그럼 한 번 사용해보겠습니다.

enum PinguError: Error {
    case itIsNil
}

let fail = Fail<String, PinguError>(error: .itIsNil)

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

간단하게 PinguError라는 에러 타입을 만들어서 Fail이 즉시 내보낼 에러로 설정합니다.

실행해보면..

receive completion: failure(__lldb_expr_92.PinguError.itIsNil)

별다른 값은 보내지 않고 에러만 받은 것을 볼 수 있습니다.

 

Fail도 아까 Empty와 비슷하게 활용할 수 있는데요! 

아까 예제와 비슷한 예제를 구현해보겠습니다.

let anyPublisher = [1, nil, 3].publisher
    .flatMap { value -> AnyPublisher<Int, PinguError> in
        if let value = value {
            let just = Just(value).setFailureType(to: PinguError.self)
            return just.eraseToAnyPublisher()
        } else {
            return Fail<Int, PinguError>(error: .itIsNil).eraseToAnyPublisher()
        }
    }
    .sink(receiveCompletion: { print("Completion: \($0)") },
          receiveValue: { print("value: \($0)") }
    )

아까와 다르게 nil이 발견되면 Fail을 반환해서 처리하도록 했습니다.

Fail은 Empty와 다르게 에러를 내보내기 때문에 결과도 아래와 같이 나오게 됩니다.

value: 1
Completion: failure(__lldb_expr_101.PinguError.itIsNil)

 

Record (Struct)

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

정의를 보면 각각의 Publisher가 나중에 내보낼 수 있도록 input과 completion을 저장해두는 Publisher라고 합니다.

 

input과 completion을 저장해둬야 하니까 구현에서도 그런 프로퍼티가 있을 거 같은데, 정말 그럴지 확인해보겠습니다.

@available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public struct Record<Output, Failure> : Publisher where Failure : Error {
    public let recording: Record<Output, Failure>.Recording
    public init(record: (inout Record<Output, Failure>.Recording) -> Void)
        
    public init(recording: Record<Output, Failure>.Recording)
    public init(output: [Output], completion: Subscribers.Completion<Failure>)
    public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
                    
	
    public struct Recording {
        public typealias Input = Output
        public var output: [Output] { get }
        public var completion: Subscribers.Completion<Failure> { get }
        public init()
        public init(output: [Output], completion: Subscribers.Completion<Failure> = .finished
        public mutating func receive(_ input: Record<Output, Failure>.Recording.Input)
        public mutating func receive(completion: Subscribers.Completion<Failure>)
    }

위 코드를 보면 값들을 저장하기 위해서 Recording이라는 구조체가 내부에 구현된 것을 볼 수 있네요.

생성자들도 저장할 값들을 받는 거 외에는 기존과 별 차이점이 없어 보입니다.

 

실제로 사용해보면..

let record = Record<String, Never>(output: ["Pingu", "Pinga", "Roby"], completion: .finished)

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

이렇게 간단하게 사용할 수 있습니다.

결과도 간단해요.

value: Pingu
value: Pinga
value: Roby
completion: finished

근데 여기서 이후에도 값을 넣을 수 있을지 궁금해서 좀 찾아봤더니 Record의 값들을 저장하는 Recording 구조체의 구현에 있는 receive() 메서드들에는 아래와 같은 특징이 있었습니다.

• receive(_ input: Record<Output, Failure>.Recording.Input)
  • completion이 추가된 이후에 값을 추가하려고 하면 Fatal Error 발생
• receive(completion: Subscribers.Completion<Failure>)
  • completion는 하나만 추가 가능, 추가로 호출할 시 Fatal Error 발생

넵.. 안된다는 걸로;

 

이렇게 Apple에서 미리 구현해둔 Publisher들을 알아봤습니다.

이것 외에도 직접 Publisher를 구현할 수도 있으니 필요한 건 직접 구현해서 사용하면 될 것 같아요.

 

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

 

전체 코드는 여기에서 볼 수 있어요.

 

감사합니다.