다른 값이나 시퀀스로 변형할 수 있는 operator

들어가기 전에

Operator는 Publisher다!

1주차 스터디에서 스터디원이 sink() 라는 메소드는 왜 있는 건가요?? 라고 던진 질문에 Publisher 안에 자체적으로 구현되어 있는 게 아닐까요?😅 라고만 답했었는데...

애초에 Publisher Operators 라는 메소드들이 공식적으로 정의가 되어 있었다!

받은 항목들을 가지고 downstream publisher나 subscriber를 만드는 메소드들을 Publisher Operators라고 한다.

Publisher Operators | Apple Developer Documentation

이 operator들은 publisher를 리턴한다.

이런 식으로..

• upstream value를 받는다 → 데이터를 조작한다 → 이 값을 downstream으로 보낸다

Collecting Values

collect()

func collect() -> Publishers.Collect<Self>

• 받은 항목들을 모아서 publisher가 종료될 때 하나의 배열로 방출함
• publisher가 에러를 방출하면 receiver에게 그대로 에러를 전달함
• publisher로부터 무한대로 항목을 받고, 받은 값들을 저장하기 위해 메모리를 무제한으로 사용함 → 메모리 관리에 주의해야 함
  

let numbers = (0...10)
cancellable = numbers.publisher
    .collect()
    .sink { print("\($0)") }

// Prints: "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"

특정 개수만큼 항목들을 묶어서 방출하고 싶으면 파라미터로 숫자를 전달한다.

func collect(_ count: Int) -> Publishers.CollectByCount<Self>

let numbers = (0...10)
cancellable = numbers.publisher
    .collect(5)
    .sink { print("\($0), terminator: " "") }

// Prints "[0, 1, 2, 3, 4] [5, 6, 7, 8, 9] [10] "

→ 파라미터는 최댓값이기 때문에 파라미터로 받은 숫자보다 더 적은 개수의 항목이 묶일 수도 있음

Publishers.Collect

struct Collect<Upstream> where Upstream : Publisher

Publishers.CollectByCount

struct CollectByCount<Upstream> where Upstream : Publisher

Mapping Values

map(_:)

func map<T>(_ transform: @escaping (Self.Output) -> T) -> Publishers.Map<Self, T>

publisher로부터 받은 값들을 변형하는 operator

let numbers = [5, 4, 3, 2, 1, 0]
let romanNumeralDict: [Int : String] =
   [1:"I", 2:"II", 3:"III", 4:"IV", 5:"V"]
cancellable = numbers.publisher
    .map { romanNumeralDict[$0] ?? "(unknown)" }
    .sink { print("\($0)", terminator: " ") }

// Prints: "V IV III II I (unknown)"

Publishers.Map

struct Map<Upstream, Output> where Upstream : Publisher

upstream publisher로부터 받은 항목들을 전달받은 클로저를 사용해서 변형하는 publisher

tryMap(_:)

func tryMap<T>(_ transform: @escaping (Self.Output) throws -> T) -> Publishers.TryMap<Self, T>

upstream publisher로부터 받은 항목들을 에러를 던지는 클로저를 이용해서 변형하는 operator

• transform : 한 항목을 파라미터로 받고, 새 항목을 리턴하는 클로저. 클로저가 에러를 던지면 publisher는 에러와 함께 실패한다.

struct ParseError: Error {}
func romanNumeral(from:Int) throws -> String {
    let romanNumeralDict: [Int : String] =
        [1:"I", 2:"II", 3:"III", 4:"IV", 5:"V"]
    guard let numeral = romanNumeralDict[from] else {
        throw ParseError()
    }
    return numeral
}
let numbers = [5, 4, 3, 2, 1, 0]
cancellable = numbers.publisher
    .tryMap { try romanNumeral(from: $0) }
    .sink(
        receiveCompletion: { print ("completion: \($0)") },
        receiveValue: { print ("\($0)", terminator: " ") }
     )

// Prints: "V IV III II I completion: failure(ParseError())"

에러 처리가 필요하면 tryMap(_:) 을, 아니면 map(_:) 을 사용하자

Publishers.TryMap

struct TryMap<Upstream, Output> where Upstream : Publisher

upstream publisher로부터 받은 항목들을 에러를 던지는 클로저를 이용해서 변형하는 pubilsher

Flattening Publishers

flatMap(maxPublishers:_:)

func flatMap<T, P>(
    maxPublishers: Subscribers.Demand = .unlimited,
    _ transform: @escaping (Self.Output) -> P
) -> Publishers.FlatMap<P, Self> where T == P.Output, P : Publisher, Self.Failure == P.Failure

Republishing Elements by Subscribing to New Publishers
새 publisher를 구독해서 항목을 새로 publish 하는 것

publisher로부터 받은 항목들을 새 publisher로 변형하는 operator

• maxPublishers : 동시에 존재하는 publisher subscription의 최댓값. 기본값은 unlimited
• transform : 항목을 파라미터로 받고, 같은 타입의 항목을 만드는 publisher를 반환하는 클로저

maxPublisher를 2로 설정했을 때 예시

• upstream publisher로부터 받은 값을 가지고 새 이벤트를 만들고 싶을 때 사용
• 클로저는 받은 값을 가지고 새 Publisher를 만든다
  • 이 Publisher는 하나 이상의 이벤트를 방출할 수 있다
  • 이 Publisher가 정상적으로 종료되어도 전체 스트림이 완료되지 않는다
  • 하지만 실패할 때는 전체 스트림도 같이 실패한다!
• flatMap은 각 output을 하나의 publisher로 변형하는데, 이 publisher들을 downstream으로 방출되는 publisher를 업데이트하기 위해 이 publisher들을 버퍼에 가지고 있는다. → 메모리 관리에 주의해야 한다!
  

public struct WeatherStation {
    public let stationID: String
}

var weatherPublisher = PassthroughSubject<WeatherStation, URLError>()

cancellable = weatherPublisher.flatMap { station -> URLSession.DataTaskPublisher in
    let url = URL(string:"https://weatherapi.example.com/stations/\(station.stationID)/observations/latest")!
    return URLSession.shared.dataTaskPublisher(for: url)
}
.sink(
    receiveCompletion: { completion in
        // Handle publisher completion (normal or error).
    },
    receiveValue: {
        // Process the received data.
    }
 )

weatherPublisher.send(WeatherStation(stationID: "KSFO")) // San Francisco, CA
weatherPublisher.send(WeatherStation(stationID: "EGLC")) // London, UK
weatherPublisher.send(WeatherStation(stationID: "ZBBB")) // Beijing, CN 

Publishers.FlatMap

struct FlatMap<NewPublisher, Upstream> where NewPublisher : Publisher, Upstream : Publisher, NewPublisher.Failure == Upstream.Failure

upstream publisher로부터 받은 항목을 새 publisher로 변형하는 publisher

🤯 map? flatMap?

[1, 2, 3].publisher.flatMap({ int in
  return (0..<int).publisher
  }).sink(receiveCompletion: { _ in }, receiveValue: { value in
    print("value: \(value)")
  })

/* Result:
value: 0
value: 0
value: 1
value: 0
value: 1
value: 2
*/

let tmp1 = [1, 2, 3].publisher.flatMap({ int in
  return (0..<int).publisher
  })

let tmp2 = [1, 2, 3].publisher.map { int in
    return (0..<int).publisher
}

/*
value: Sequence<Range<Int>, Never>(sequence: Range(0..<1))
value: Sequence<Range<Int>, Never>(sequence: Range(0..<2))
value: Sequence<Range<Int>, Never>(sequence: Range(0..<3))
*/

• Publishers.Map은 변형된 upstream publisher의 output을 output으로 가지는 publisher
• Publishers.FlatMap은 upstream publisher의 output을 요리조리 변형해서 새 publisher를 만드는 publisher
  • 그리고 애초에 flatMap은 publisher를 리턴한다.

그래서 map을 사용해서 publisher로 변형하면 output이 publisher인 publiser가 만들어지는 것!! (Publisher<NewPublisher, Error> 같은 느낌..)

하지만 flatMap을 사용하면 각 항목들을 변형한 하나의 publisher가 예쁘게 만들어진다.

Upstream Output 바꾸기

replaceNil(with:)

func replaceNil<T>(with output: T) -> Publishers.Map<Self, T> where Self.Output == T?

스트림에 있는 nil 항목을 특정 항목으로 교체하는 operator

let numbers: [Double?] = [1.0, 2.0, nil, 3.0]
numbers.publisher
    .replaceNil(with: 0.0)
    .sink { print("\($0)", terminator: " ") }

// Prints: "Optional(1.0) Optional(2.0) Optional(0.0) Optional(3.0)"

replaceEmpty(with:)

func replaceEmpty(with output: Self.Output) -> Publishers.ReplaceEmpty<Self>

빈 스트림을 전달된 항목으로 바꾸는 operator

• output : upstream publisher가 아무 값도 방출되지 않고 종료되면 방출되는 항목

let numbers: [Double] = []
cancellable = numbers.publisher
    .replaceEmpty(with: Double.nan)
    .sink { print("\($0)", terminator: " ") }

// Prints "(nan)".

Double.nan 을 방출하고 정상 종료됨

let otherNumbers: [Double] = [1.0, 2.0, 3.0]
cancellable2 = otherNumbers.publisher
    .replaceEmpty(with: Double.nan)
    .sink { print("\($0)", terminator: " ") }

// Prints: 1.0 2.0 3.0

방출되는 값이 있으면 replaceEmpty 로 전달한 값은 방출되지 않는다.

Publishers.ReplaceEmpty

struct ReplaceEmpty<Upstream> where Upstream : Publisher

Incrementally transforming output

scan(::)

func scan<T>(
    _ initialResult: T,
    _ nextPartialResult: @escaping (T, Self.Output) -> T
) -> Publishers.Scan<Self, T>

클로저한테 마지막으로 전달받은 값과 현재 항목을 가 직고 sptream publisher로부터 받은 항목들을 변형하는 operator

• initialResult : nextPartialResult 로부터 리턴받은 이전 결과
• nextPartialResult : 클로저로부터 리턴받은 이전 값과 upstream publisher로부터 방출받을 다음 항목을 인자로 받는 클로저

이전에 publish된 값들을 모아서 하나의 값으로 만들고 싶을 때 사용한다.
→ reduce() 랑 비슷한 느낌?

let range = (0...5)
cancellable = range.publisher
    .scan(0) { return $0 + $1 }
    .sink { print ("\($0)", terminator: " ") }
 // Prints: "0 1 3 6 10 15 ".

Publishers.Scan

struct Scan<Upstream, Output> where Upstream : Publisher