[Combine] 네트워킹

Processing URL Session Data Task Results with Combine  에 나오는 내용입니다. 

 

[ 목표 ]

Combine으로 네트워크 쪽 코드를 작성하는 것은 크게 두가지 절차를 거칩니다.

 

1. Data Task Publisher 를 만든다. 

2. 비동기 연산자 (asynchronous operator) 들을 체이닝하여 data를 받고 처리한다. 

 

이를 알아봅니다. 

 

또한 

네트워크 쪽 코드작성할 때 필요한 

에러 핸들링, 스케쥴링, share 도 알아봅니다! 

 

 

[1] Data Task Publisher 만들기 

URLSession  은 Combine Publisher 인 URLSession.DataTaskPublisher 를 제공합니다. 

func dataTaskPublisher(for: URLRequest) -> URLSession.DataTaskPublisher
func dataTaskPublisher(for: URL) -> URLSession.DataTaskPublisher

DataTaskPublisher는  URL  또는  URLRequest 에서 가져온 data의 결과를 전달해 주는 publisher 입니다. 

 

Ouput은 (Data, URLResponse) 타입,

Failure 는 URLError 타입입니다. 

    public struct DataTaskPublisher : Publisher, Sendable {

        public typealias Output = (data: Data, response: URLResponse)

        public typealias Failure = URLError
        
        ...
  }

 

task가 성공으로 끝나면 Output을 publish 하고

실패로 끝나면 Failure를 publish 합니다. 

let url = URL(string: "https://example.com/endpoint")!
urlSession.dataTaskPublisher(for: url)

 

 

[2] 데이터 받고 처리 

 

2.1 데이터 받기

 

우선 raw Data 를 받아봅시다.  

 

map(_:) operator 를 쓸 수 도 있지만, response가 성공이 아니라면 error를 던지기 위해 tryMap(_:) 을 써줍니다. 

guard let url = URL(string: "https://example.com/endpoint") else { return }
URLSession.shared.dataTaskPublisher(for: url)
    .tryMap { output -> Data in
        guard let httpResponse = output.response as? HTTPURLResponse,
              httpResponse.statusCode == 200 else {
            throw URLError(.badServerResponse)
        }
        return output.data
    }

 

2.2  데이터를 원하는 타입(Decodable을 따르는) 으로 컨버팅 

 

decode(type:decoder:) 를 사용해줍니다. 

struct User: Codable {
    let name: String
    let userID: String
}

guard let url = URL(string: "https://example.com/endpoint") else { return }
URLSession.shared.dataTaskPublisher(for: url)
    .tryMap { output -> Data in
        guard let httpResponse = output.response as? HTTPURLResponse,
              httpResponse.statusCode == 200 else {
            throw URLError(.badServerResponse)
        }
        return output.data
    }
    .decode(type: User.self, decoder: JSONDecoder())

 

 

2.3 구독하기 

 

체이닝을 해준 publisher에 subscriber를 붙여서 구독해줍니다. 

sink(receiveCompletion:receiveValue:) 를 사용해줍니다. 

guard let url = URL(string: "https://example.com/endpoint") else { return }
URLSession.shared.dataTaskPublisher(for: url)
    .tryMap { output -> Data in
        guard let httpResponse = output.response as? HTTPURLResponse,
              httpResponse.statusCode == 200 else {
            throw URLError(.badServerResponse)
        }
        return output.data
    }
    .decode(type: User.self, decoder: JSONDecoder())
    .sink(receiveCompletion: { print ("Received completion: \($0).") },
          receiveValue: { user in print ("Received user: \(user).")})

 

참고로 receiveCompletion 의 Subscribers.Completion enum을 활용해서 이런 식으로 해줄 수 도 있습니다! 

    .sink(receiveCompletion: { completion in
        switch completion {
        case .failure(let error):
            print("Error: \(error.localizedDescription).")
        case .finished
            break
        }
    },
    receiveValue: { user in print ("Received user: \(user).")})

 

 

[3] 에러 핸들링

 

3.1 재시도 

retry(_:) operator 를 사용해서 에러를 핸들링할 수 있습니다. 

upstream publisher에 대해 넘겨준 횟수만큼 구독을 다시 만듭니다. 

 

 

3.2  error를 대체 

catch(_:)  를 이용해서 subscriber에게 가기 전에  에러를 다른 publisher로 대체할 수 있습니다.

또는

replaceError(with:)  를 사용해서 에러를 element 로 대체할 수 있습니다. 

 

 

아래의 예제는 재시도(3.1)와 에러 대체(3.2)를 동시에 보여줍니다. 

let pub = urlSession
    .dataTaskPublisher(for: url)
    .retry(1)
    .catch() { _ in
        self.fallbackUrlSession.dataTaskPublisher(for: fallbackURL)
    }
cancellable = pub
    .sink(receiveCompletion: { print("Received completion: \($0).") },
          receiveValue: { print("Received data: \($0.data).") })

 

request가 실패하면 한번 더 시도하고 그후에는 fallback URL을 사용합니다. 

처음 요청, 재시도, fallback 요청 중 하나가 성공하면 receiveValue가 data를 받고

세가지가 모두 실패하면 receiveCompletion이 Subscribers.Completion.failure(_:) 을 받게 됩니다. 

 

 

[4] 스케쥴링 

 

Combine의 scheduling operator로  원하는 Dispatch Queue 로 이동할 수 있습니다. 

 

receive(on:options:)  를 사용해서 이 뒤에 나오는 operator 또는 subscriber 가 어디서 work를 실행할 지 명시할 수 있습니다.

receive는 Scheduler를 받는데,

 

DispatchQueue 와 RunLoop 는 Scheduler protocol을 채택하고 있어서 둘다 사용할 수 있습니다. 

 

아래의 예제는 sink가 프린트를 main dispatch queue에서 하는 것을 보장합니다.

cancellable = urlSession
    .dataTaskPublisher(for: url)
    .receive(on: DispatchQueue.main)
    .sink(receiveCompletion: { print ("Received completion: \($0).") },
          receiveValue: { print ("Received data: \($0.data).")})

 

 

subscribe와 비교

 

참고로 upstream에 영향을 미치는 subscribe(on:options:) 와 달리,

receive(on:options:) 은 downstream에 영향을 미친다는 것을 기억해주세요! 

 

 

In the following example, the subscribe(on:options:) operator causes jsonPublisher to receive requests on backgroundQueue, while the receive(on:options:) causes labelUpdater to receive elements and completion on RunLoop.main.

let jsonPublisher = MyJSONLoaderPublisher() // Some publisher.
let labelUpdater = MyLabelUpdateSubscriber() // Some subscriber that updates the UI.

jsonPublisher
    .subscribe(on: backgroundQueue)
    .receive(on: RunLoop.main)
    .subscribe(labelUpdater)

 

권장하는 패턴 

 

그리고 이거 보다

pub.sink {
    DispatchQueue.main.async {
        // Do something.
    }
}

 

이렇게 하는게 좋다고 가이드 하고 있습니다! 

pub.receive(on: DispatchQueue.main).sink {
    // Do something.
}

 

 

[5] Data Task Publisher의 결과를 여러개의 Subscriber들이 share 하기 

URL endpoint 로부터 온 데이터를 앱의 여러 부분에서 사용하고 싶을 수 있습니다.

네트워크 요청은 비용이 많이 들기 때문에 불필요하게 재발행하지 마십시오.

 

Combine은 하나의 URLSession.DataTaskPublisher 를 multiple subscribers 가 사용할 수 있게 해줍니다.

publisher가 한번의 request로 모든 subscribers에게 서비스할 수 있도록!

(영어 뉘앙스를 못살리겠네요;; "allowing the publisher to service all of them with a single request." )

 

share() operator 를 사용해주면 됩니다.

이 operator는  Publishers.Multicast and PassthroughSubject publishers 의 combination 처럼 동작합니다. 

let sharedPublisher = urlSession
    .dataTaskPublisher(for: url)
    .share()

cancellable1 = sharedPublisher
    .tryMap() {
        guard $0.data.count > 0 else { throw URLError(.zeroByteResource) }
        return $0.data
    }
    .decode(type: User.self, decoder: JSONDecoder())
    .receive(on: DispatchQueue.main)
    .sink(receiveCompletion: { print ("Received completion 1: \($0).") },
          receiveValue: { print ("Received id: \($0.userID).")})

cancellable2 = sharedPublisher
    .map() {
        $0.response
    }
    .sink(receiveCompletion: { print ("Received completion 2: \($0).") },
           receiveValue: { response in
            if let httpResponse = response as? HTTPURLResponse {
                print ("Received HTTP status: \(httpResponse.statusCode).")
            } else {
                print ("Response was not an HTTPURLResponse.")
            }
    }
)

 

이 예제는 URL session data task를 두가지 상관없는 목적으로 쓰고 있습니다.

첫번째 subscriber는 User type을 data로부터 파싱하고 main dispatch queue에 그것을 프린트 합니다. 

두번째 subscriber는 HTTP status code를 프린트 합니다. (어떤 queue를 쓰는지는 상관없이) 

 

share() 를 통해서 data task publisher는 단일 요청으로 두가지 subscribers에게 서비스를 제공할 수 있습니다.