URLSession.DataTaskPublisher

DataTaskPublisher

URLSession 은 combine publisher인 DataTaskPublisher를 제공한다.
dataTaskPublisher(for:) 메쏘드로 publisher를 생성 할 수 있으며, 결과는 듀플( Data: , URLResponse)로 전달된다.

public struct DataTaskPublisher: Publisher {
   typealias Output = (data: Data, response: URLResponse)
   typealias Failure = URLError
   
   let request: URLRequest
   let session: URLSession
   
   func receive<S>(subscriber: S) where S: Subscriber, S.Failure == URLSession.DataTaskPublisher.Failure, S.Input == URLSession.DataTaskPublisher.Output
}

 

응답 및 json 디코딩

dataTaskPublisher를 통해 http 응답을 받고, codable 객체를 통해 json을 객체로 변환한다.
DataTaskPublisher의 output은 (Data, URLResponse) 튜플인데, decode 메쏘드는 Data만 필요하므로, tryMap으로 data 만 별도로 추출한다.
DataTaskPublisher의 Failure 타입이 URLError 이므로, 별도 정의된 에러를 사용하기 위해 mapError 사용.

publisher

enum HttpError: LocalizedError {
   case unknown
   case httpStatusError(Int, String)
}

class Http {
   static func request<T: Codable>(type: T.Type) -> AnyPublisher<T, HttpError> {
      let url = URL(string: "http://test/url")!

      return urlSession
         .dataTaskPublisher(for: url)
         .tryMap() { data, response in
            guard let httpResponse = response as? HTTPURLResponse else {
               print("응답 오류")
               throw HttpError.unknown
            }
      
            guard 200..<300 ~= httpResponse.statusCode else {
               throw HttpError.httpStatusError(httpResponse.statusCode, httpResponse.description)
            }
      
            guard !data.isEmpty else {
               throw HttpError.unknown
            }
      
            return data
         }
         .decode(type: type, decoder: JSONDecoder())
         .mapError { error in
            if let error = error as? HttpError {
               return error
            } else {
               return HttpError.unknown
            }
         }
         .eraseToAnyPublisher()
    }
 }

 

test

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

class HttpTest: XCTestCase {
   var subscription: Cancellable?
   
   override func tearDownWithError() throws {
      if let subscription = subscription {
         subscription.cancel()
      }
      subscription = nil
   }
   
   
   func httpResponseTest() {
      let expectation = XCTestExpectation()
      
      subscription = Http.request(type: User.self)
         .sink(receiveCompletion: { completion in
            switch completion {
            case .finished:
               print("finished")
            
            case .failure(let err):
               print("error: \(err)")
            }
            expectation.fulfill()
         }, receiveValue: { data in
         
         })
      
      wait(for: [expectation], timeout: 2.0)
   }
}

 

 

재시도 및 실패시 다른 publisher 처리

retry 로 재시도 오퍼레이터를 사용하고, catch 를 통해 에러 발생시 다른 url 의 publisher를 리턴~

let publisher = urlSession
   .dataTaskPublisher(for: url)
   .retry(1)
   .catch() { _ in
      self.fallbackUrlSession.dataTaskPublisher(for: fallbackURL)
   }
   
let cancellable = publisher
   .sink(receiveCompletion: {
   
   }, receiveValue: {
   
   })

 

dataTaskPublisher(for:) 이후 ui 처리등을 위해 메인 스레드를 사용하는 경우 receive(on: DispatchQueue.main) 등으로 처리할 쓰레드 변경.

let publisher = urlSession
   .dataTaskPublisher(for: url)
   .retry(1)
   .catch() { _ in
      self.fallbackUrlSession.dataTaskPublisher(for: fallbackURL)
   }
   
let cancellable = publisher
   .receive(on: DispatchQueue.main)
   .sink(receiveCompletion: {
   
   }, receiveValue: {
   
   })

 

만약 특정 시간 이후에 재시도를 하려고 하면 catch -> retry 순서로 변경한다.
catch를 통해 에러를 먼저 잡은 뒤 딜레이를 두어 에러를 방출하면 이후 retry 구문에서 재시도를 시도하는 형태로 구현할 수 있다.

 

 

 

스트림을 여러 subscriber 에게 전달

다중 다운스트림 구독의 경우 ConnectablePublisher 를 사용해 여러 구독자에게 스트림을 전달할 수 있는데, publisher에는 ConnectablePublisher와 같은 기능을 하는 share(), multicast(_:) 메쏘드들을 제공한다.
상황에 따라 share()를 사용하거나 multicast(_:)를 사용하게 되는데, share()는 내부적으로 구독이 발생하면 connect() 메쏘드가 자동으로 발생하고, multicast(_:)는 명시적으로 connect()를 호출해야 한다는 차이가 있다.
즉, share()의 경우 첫 구독자가 연결되고 요청이 완료된 이후에 다른 구독자가 연결되면 나중에 등록된 구독자는 스트림을 받지 못하게 되거나 완료 이벤트만 전달받게 된다.
특정 시점에 publisher가 동작하도록 하려면, share().makeConnectable() 이나 multicast(_:)의 반환값인 ConnectablePublisher를 사용해야 한다. 

URLSession.dataTaskPublisher 관련 다중 구독 예)

publisher

// share
let sharedPublisher = urlSession
   .dataTaskPublisher(for: url)
   .share()  
   
   
// share, makeConnectable   
let sharedPublisher = urlSession
   .dataTaskPublisher(for: url)
   .share()
   .makeConnectable()
   
// 구독 추가
   .
   .
sharedPublisher.connect()


// connectable publisher
let sharedPublisher = urlSession
   .dataTaskPublisher(for: url)
   .multicast { PassthroughSubject<(data: Data, response: URLResponse), , URLError>() }
   
// 구독 추가
   .
   .

// 구독 추가 완료 후 publisher 시작
let cancellable = sharedPublisher.connect()

 

subscriber

데이터와 http 응답을 별도의 subscriber에서 처리하고자 하는 경우 하나의 스트림에 대해 여러개의 구독으로 처리하게 된다. 

let cancellable1 = sharedPublisher
   .tryMap {
      guard $0.data.count > 0 else { throw URLError(.zeroByteResource) }
      return $0.data
   }
   .decode(type: MyData.self, decoder: JSONDecoder())
   .receive(on: DispatchQueue.main)
   .sink(receiveCompletion: {
   
   }, receiveValue: {
   
   })
   
   
let cancellable2 = sharedPublisher
   .map {
      $0.response
   }
   .sink(receiveCompletion: {
   
   }, receiveValue: { response in
   
   })