[Combine] Republishing Elements by Subscribing to New Publishers - Operator 공부 11

안녕하세요 Pingu입니다.🐧

 

지난 글에서는 Combine의 Combining Elements from Multiple Publishers로 분류된 것들 중 Collecting and Republishing the Oldest Unconsumed Elements from Multiple Publishers 역할을 하는 Zip 시리즈에 대해 알아봤습니다. 여러 개의 Publisher에서 가장 오래 사용되지 않은 값들을 모아서 처리하는 역할을 했었습니다.

 

이번 글에서는 Republishing Elements by Subscribing to New Publishers로 분류된 것들에 대해 알아보도록 하겠습니다.

 

Republishing Elements by Subscribing to New Publishers

이번글에서 공부할 것들을 분류한 이름을 보면 새로운 Publisher를 subscribe 해서 값을 다시 내보내는 역할을 한다고 되어있네요.

 

일단 어떤 Publisher들이 있는지 살펴보겠습니다.

• FlatMap
• SwitchToLatest

위와 같이 2개의 Publisher로 만든 Operator는 아래와 같습니다.

• flatMap(maxPublishers:,_:)
• switchToLatest()

그럼 바로 하나씩 자세히 알아보도록 할게요.

 

FlatMap

이번 글에서 알아볼 첫 번째 Publisher는 FlatMap 입니다.

정의를 보면 Upstream Publisher의 값을 새로운 Publisher로 변환하는 역할을 하는 Publisher라고 되어있네요.

실제로 이를 활용해서 만든 Operator를 보고 역할을 이해해보겠습니다.

flatMap(maxPublishers:,_:)

FlatMap Publisher를 활용해서 만든 Operator는 flatMap(maxPublishers:_:)입니다. 정의를 보면 Upstream의 값을 지정한 최대 횟수까지 내보내는 새로운 Publisher를 반환하는 Publisher입니다.

 

매개변수로는 maxPublishers가 있고 transform이 있습니다.

maxPublishers의 타입은 Subscribers.Demand인걸 볼 수 있습니다. 이는 처리할 Publisher의 수라고 볼 수 있어요. 즉 몇 개의 Publisher를 처리할 것인지 정하는 것인데요, 따로 설정하지 않으면 무한 개의 Publisher를 처리할 수 있습니다.

transform은 Upstream의 값을 사용해서 해당 타입을 내보내는 Publisher를 반환하는 클로저입니다.

 

그럼 간단하게 한 번 사용해보며 이해해보겠습니다.

let publisher1 = PassthroughSubject<String, Never>()
let publisher2 = PassthroughSubject<String, Never>()

let publishers = PassthroughSubject<PassthroughSubject<String, Never>, Never>()

publishers
    .flatMap(maxPublishers: .max(1)) { $0 }
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })

publishers.send(publisher1)
publishers.send(publisher2) // 무시됨

publisher1.send("Hello")
publisher1.send("Pingu")

publisher2.send("Good Bye") // 무시됨
publisher2.send("Pinga") // 무시됨

publisher1.send(completion: .finished)
publishers.send(completion: .finished)

// flatMap(maxPublishers:_:) 예제 코드
Hello
Pingu
finished

위 코드를 보면 flatMap의 maxPublishers 값에 max(1)이 들어간 것을 볼 수 있는데요, 즉 한 개의 Publisher만 처리하겠다는 의미가 됩니다.

그렇기 때문에 위와 같이 publisher1만 처리되고 publisher2는 무시되게 됩니다.

이렇게 flatMap을 사용하면 여러 개의 Publisher를 하나의 Publisher처럼 사용할 수 있게 됩니다.

 

그럼 이번엔 flatMap을 좀 더 실용적으로 사용해보겠습니다.

func decodeOnlyAlphabet(_ codes: [Int]) -> AnyPublisher<String, Never> {
    Just(
        codes
            .compactMap { code in
                guard (65...90).contains(code) || (97...122).contains(code) else { return nil }
                return String(UnicodeScalar(code) ?? " ")
            }
            .joined()
    )
    .eraseToAnyPublisher()
}

let intArrayPublisher = PassthroughSubject<[Int], Never>()
intArrayPublisher
    .flatMap(decodeOnlyAlphabet)
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })

intArrayPublisher.send([1, 80, 105, 110, 103, 117])
intArrayPublisher.send([1, 80, 105, 110, 103, 97])
intArrayPublisher.send(completion: .finished)

// flatMap(maxPublishers:_:) 예제 코드
Pingu
Pinga
finished

위 코드를 보면 decodeOnlyAlphabet(_:)이라는 함수가 있습니다. 해당 함수는 정수 배열을 받아서 각각의 값 중 아스키코드의 알파벳으로 바꿀 수 있는 값만 String으로 바꿔서 합친 뒤 Downstream으로 내려보내는 Publisher를 반환합니다.  그리고 메서드는 flatMap의 transform 매개변수에 사용됩니다.

 

collect()로 Upstream에서 받은 값을 모두 모아서 flatMap에 전달해주면 decodeOnlyAlphabet의 매개변수로 전달하면 flatMap은 Downstream으로 값을 전달하는 게 아닌 Publisher가 전달됩니다.

 

그래서 결과는 위와 같이 나오게 됩니다.

 

이렇게 flatMap을 사용하면 Upstream에서 받은 Publisher와는 다른 타입의 Publisher를 Downstream으로 전달할 수 있습니다.

SwitchToLatest

다음으로 알아볼 Publisher는 SwitchToLatest입니다.

정의를 보면 중첩된 Publisher를 합치는 Publisher라고 되어있네요.

 

좀 더 자세히 알아보기 위해 SwitchToLatest Publisher의 실제 구현을 보면 아래와 같습니다.

public struct SwitchToLatest<P, Upstream> : Publisher where P : Publisher, P == Upstream.Output, Upstream : Publisher, P.Failure == Upstream.Failure {
	public typealias Output = P.Output
	public typealias Failure = P.Failure
	public let upstream: Upstream
    
	public init(upstream: Upstream)
    
	public func receive<S>(subscriber: S) where S : Subscriber, P.Output == S.Input, Upstream.Failure == S.Failure
}

생성자를 보면 Upstream, 즉 Publisher가 필요하다는 것을 알 수 있습니다. 즉 switchToLatest는 Publisher를 내보내는 Publisher에서만 사용할 수 있고 Upstream에서 Publisher를 전달받는다는 것을 알 수 있습니다.

 

Publisher들을 어떻게 합치는 녀석인지는 실제로 이를 활용해서 구현된 Operator를 보면 쉽게 이해할 수 있습니다.

switchToLatest()

SwitchToLatest Publisher를 활용해서 만들어진 Operator는 switchToLatest()입니다.

정의를 보면 가장 최근에 받은 Publisher의 값을 다시 내보낸다고 되어있습니다.

이렇게 보면 저는 잘 이해가 안 됐었는데요, 간단히 말해서 SwitchToLatest는 Upstream에서 Publisher를 받으며 가장 최근에 받은 Publisher가 내보내는 값만 Downstream으로 전달합니다.

 

간단하게 한 번 사용해볼게요.

let publisher1 = PassthroughSubject<Int, Never>()
let publisher2 = PassthroughSubject<Int, Never>()
let publisher3 = PassthroughSubject<Int, Never>()

let publishers = PassthroughSubject<PassthroughSubject<Int, Never>, Never>()

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

publishers.send(publisher1)
publisher1.send(1)

publishers.send(publisher2)
publisher1.send(2)  // downstream으로 전달이 안된다
publisher2.send(11)

publishers.send(publisher3)
publisher1.send(3)  // downstream으로 전달이 안된다
publisher2.send(12) // downstream으로 전달이 안된다
publisher3.send(111)

publisher3.send(completion: .finished)
publishers.send(completion: .finished)

// switchToLatest() 예제 코드
1
11
111
finished

위 코드를 보면 Int 값을 내보내는 publisher1, 2, 3가 있고, Publisher를 내보내는 publishers가 있습니다.

코드를 보면 publishers는 가장 최근에 내보낸 publisher의 값만 Downstream으로 전달되는 것을 볼 수 있습니다.

 

SwitchToLatest는 네트워크 요청 작업에서 유용하게 사용할 수 있는데요, 어떤 버튼을 누르면 네트워크 요청을 한다고 가정해볼게요.

사용자가 버튼을 여러 번 탭 하는 상황이 발생하면 가장 마지막 탭에서 발생한 네트워크 요청만 처리하고 싶을 수 있습니다.

이럴 때 switchToLatest를 사용하면 유용합니다.

 

간단하게 해당 상황을 실제로 구현해볼게요.

var subscriptions = Set<AnyCancellable>()

let url = URL(string: "https://source.unsplash.com/random")!
    
func getImage() -> AnyPublisher<UIImage?, Never> {
    URLSession.shared
        .dataTaskPublisher(for: url)
        .map { data, _ in UIImage(data: data) }
        .replaceError(with: nil)
        .eraseToAnyPublisher()
}

let userTap = PassthroughSubject<Void, Never>()

userTap
    .map { _ in getImage() }
    .switchToLatest()
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)

userTap.send()

DispatchQueue.main.asyncAfter(deadline: .now() + 2.0) {
    userTap.send()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 2.01) {
    userTap.send()
}

// switchToLatest with network request 예제 코드
Optional(<UIImage:0x6000027e8d80 anonymous {1080, 1919} renderingMode=automatic>)
Optional(<UIImage:0x6000027e8cf0 anonymous {1080, 720} renderingMode=automatic>)

위 코드를 보면 userTap이 사용자가 버튼을 탭하는 행위라고 보면 됩니다.

userTap이 3번 send() 했는데 실제로 Downstream에서 받은 이미지의 수는 2개인 것을 볼 수 있죠.

이는 두 번째 탭과 세 번째 탭 사이의 시간이 0.01초라서 두 번째 탭의 네트워크 요청이 완료되기 전에 세번째 탭의 네트워크 요청이 발생하여 세번째 탭의 네트워크 요청만 처리된 것입니다.

물론 첫 번째 탭의 네트워크 결과가 2초 안에 처리되지 않았다면 Downstream이 받은 이미지는 1개일 수도 있겠죠?

 

이렇게 가장 최근에 받은 Publisher가 내보낸 값만 Downstream으로 전달하는 역할을 하는 Operator가 switchToLatest()입니다.

 

이렇게 Combine의 Publisher와 Operator 중에서 Republishing Elements by Subscribing to New Publishers로 분류된 것들에 대해 알아봤습니다.

 

다음 글에서는 Handling Error로 분류된 녀석들을 알아보도록 하겠습니다.

 

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

 

감사합니다!