• upstream publisher가 방출한 값을 downstream과 다른 operator에게 보낼 수 있다

Filtering Elements

filter(_:)

func filter(_ isIncluded: @escaping (Self.Output) -> Bool) -> Publishers.Filter<Self>

클로저에 맞는 항목만 다시 publish 하는 operator

• isIncluded: 한 항목을 파라미터로 받고 항목을 다시 publish 할지에 대한 Boolean 값을 리턴하는 클로저

let numbers: [Int] = [1, 2, 3, 4, 5]
cancellable = numbers.publisher
    .filter { $0 % 2 == 0 }
    .sink { print("\($0)", terminator: " ") }

// Prints: "2 4"

tryFilter(_:)

func tryFilter(_ isIncluded: @escaping (Self.Output) throws -> Bool) -> Publishers.TryFilter<Self>

isIncluded 클로저가 항목을 다시 publish 할지 여부를 리턴하거나 에러를 던짐.

클로저가 에러를 던지면 publisher는 에러와 함께 실패한다.

struct ZeroError: Error {}

let numbers: [Int] = [1, 2, 3, 4, 0, 5]
cancellable = numbers.publisher
    .tryFilter{
        if $0 == 0 {
            throw ZeroError()
        } else {
            return $0 % 2 == 0
        }
    }
    .sink(
        receiveCompletion: { print ("\($0)") },
        receiveValue: { print ("\($0)", terminator: " ") }
     )

// Prints: "2 4 failure(DivisionByZeroError())".

removeDuplicates(by:)

func removeDuplicates(by predicate: @escaping (Self.Output, Self.Output) -> Bool) -> Publishers.RemoveDuplicates<Self>

이전 항목과 다른 항목만 publish 하는 operator

• predicate : 두 항목이 같은지 비교하는 클로저
  • 필터링 하기 위함
  • 두 번째 항목이 첫 번째 항목과 같으면 true 를 반환함
• 중복된 항목을 publish 하는 게 아니라 consume 하는 publisher를 반환한다.

언제 쓰면 좋을까?

사용자가 연속으로 같은 문자를 입력했을 때, 중복된 문자를 무시하고 싶을 때 유용할 듯

• upstream publisher에서 중복되는 항목을 지우고 싶을 때 사용
  • 직전에 publish된 항목과 현재 publish된 항목을 비교함
• Equatable 을 구현하지 않은 타입들을 비교하거나 Equatable 을 구현한 것과 다르게 값을 비교하고 싶을 때 사용

struct Point {
    let x: Int
    let y: Int
}

let points = [Point(x: 0, y: 0), Point(x: 0, y: 1),
              Point(x: 1, y: 1), Point(x: 2, y: 1)]
cancellable = points.publisher
    .removeDuplicates { prev, current in
        // Considers points to be duplicate if the x coordinate
        // is equal, and ignores the y coordinate
        prev.x == current.x
    }
    .sink { print("\($0)", terminator: " ") }

// Prints: Point(x: 0, y: 0) Point(x: 1, y: 1) Point(x: 2, y: 1)

Publishers.RemoveDuplicates

struct RemoveDuplicates<Upstream> where Upstream : Publisher

이전 항목과 일치하지 않는 항목들만 publish하는 publisher

tryRemoveDuplicates(by:)

func tryRemoveDuplicates(by predicate: @escaping (Self.Output, Self.Output) throws -> Bool) -> Publishers.TryRemoveDuplicates<Self>

• predicate 클로저가 Boolean 값을 반환하거나 에러를 던짐.
  • 클로저가 에러를 던지면 publisher는 그 에러와 함께 종료함

struct BadValuesError: Error {}
let numbers = [0, 0, 0, 0, 1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
cancellable = numbers.publisher
    .tryRemoveDuplicates { first, second -> Bool in
        if (first == 4 && second == 4) {
            throw BadValuesError()
        }
        return first == second
    }
    .sink(
        receiveCompletion: { print ("\($0)") },
        receiveValue: { print ("\($0)", terminator: " ") }
     )

 // Prints: "0 1 2 3 4 failure(BadValuesError()"

Reducing Elements

ignoreOutput()

func ignoreOutput() -> Publishers.IgnoreOutput<Self>

모든 upstream 항목들을 무시하지만 upstream publisher의 completion 상태(finished or failed)는 전달해주는 operator

• 모든 upstream 항목들을 무시하는 publisher를 반환함
• publisher가 정상적으로 완료하거나 실패하는지 결정할 때 사용

struct NoZeroValuesAllowedError: Error {}
let numbers = [1, 2, 3, 4, 5, 0, 6, 7, 8, 9]
cancellable = numbers.publisher
    .tryFilter({ anInt in
        guard anInt != 0 else { throw NoZeroValuesAllowedError() }
        return anInt < 20
    })
    .ignoreOutput()
    .sink(receiveCompletion: {print("completion: \($0)")},
          receiveValue: {print("value \($0)")})

// Prints: "completion: failure(NoZeroValuesAllowedError())"

위 예시에서 ignoreOutput() operator는 1~5까지는 정상적으로 전달받지만 downstream으로 publish되지 않는다.

0 이 publish 되었을 때는 NoZeroValuesAllowedError 와 함께 stream이 종료된다.

Publishers.IgnoreOutput<Self>

이 Publisher의 Output은 Never 이다.

절대 항목을 방출하지 않기 때문!

Selecting Specific Elements

first()

func first() -> Publishers.First<Self>

스트림의 첫 번째 항목을 publish하고 종료하는 operator

• downstream이 하나 이상의 항목을 요청하면 first() operator가 바로 upstream으로부터 unlimited 를 요청함
• first() 가 값을 받기 전에 upstream이 종료되면 어떤 값도 방출하지 않고 종료된다.

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

// Print: "-10"

first(where:)

func first(where predicate: @escaping (Self.Output) -> Bool) -> Publishers.FirstWhere<Self>

클로저 안의 내용을 만족하는 스트림의 첫 번째 항목을 publish하고 종료하는 operator

• predicate : 한 항목을 파라미터로 받고, 그 항목을 publish 할지 여부를 리턴하는 클로저
• 첫 번째 항목을 publish 한 후에는 다른 항목들은 모두 무시하고 정상 종료한다.

  • “모두 무시”라는 표현 때문에 이후 값도 다 받긴 하지만 그냥 무시하는 건가 싶었는데 첫 번째 값을 찾으면 바로 종료되는 것 같다.

    ——— Example of: first(where:) ———
    numbers: receive subscription: (1...9)
    numbers: request unlimited
    numbers: receive value: (1)
    numbers: receive value: (2)
    numbers: receive cancel
    2
    Completed with: finished

• 아무 항목도 받지 않았다면 아무 것도 publish 하지 않고 종료한다.

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

// Prints: "1"

tryFirst(where:)

func tryFirst(where predicate: @escaping (Self.Output) throws -> Bool) -> Publishers.TryFirstWhere<Self>

predicate 클로저가 Boolean 값을 반환하거나 에러를 던질 수 있다. 클로저가 에러를 던지면 publisher는 실패한다.

let numberRange: ClosedRange<Int> = (-1...50)
numberRange.publisher
    .tryFirst {
        guard $0 < 99 else {throw RangeError()}
        return true
    }
    .sink(
        receiveCompletion: { print ("completion: \($0)", terminator: " ") },
        receiveValue: { print ("\($0)", terminator: " ") }
     )

// Prints: "-1 completion: finished"
// If instead the number range were ClosedRange<Int> = (100...200), the tryFirst operator would terminate publishing with a RangeError.

last()

func last() -> Publishers.Last<Self>

스트림이 종료된 후에 스트림의 마지막 항목을 publish 한다.

• upstream publisher의 마지막 항목을 방출하고 싶을 때 사용한다

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

// Prints: "10"

Publishers.Last

struct Last<Upstream> where Upstream : Publisher

스트림이 종료될 때까지 기다렸다가 마지막 항목을 publish하는 publisher

→ 그래서 upstream publisher는 반드시 언젠가 종료되는 publisher여야 한다!

last(where:)

func last(where predicate: @escaping (Self.Output) -> Bool) -> Publishers.LastWhere<Self>

upstream이 종료된 후에 클로저 내의 조건을 만족하는 스트림의 마지막 항목을 publish하는 operator

• predicate: 한 항목을 파라미터로 받고 publish 할지 여부를 반환하는 클로저

let numbers = (-10...10)
cancellable = numbers.publisher
    .last { $0 < 6 }
    .sink { print("\($0)") }

// Prints: "5"

tryLast(where:)

func tryLast(where predicate: @escaping (Self.Output) throws -> Bool) -> Publishers.TryLastWhere<Self>

• predicate 클로저가 Boolean 값을 리턴하거나 에러를 던진다.
• 클로저가 에러를 던지면 publisher도 실패한다.

struct RangeError: Error {}

let numbers = [-62, 1, 6, 10, 9, 22, 41, -1, 5]
cancellable = numbers.publisher
    .tryLast {
        guard 0 != 0  else {throw RangeError()}
        return true
    }
    .sink(
        receiveCompletion: { print ("completion: \($0)", terminator: " ") },
        receiveValue: { print ("\($0)", terminator: " ") }
    )
// Prints: "5 completion: finished"
// If instead the numbers array had contained a `0`, the `tryLast` operator would terminate publishing with a RangeError."

Applying Sequence Operations to Elements

drop(untilOutputFrom:)

func drop<P>(untilOutputFrom publisher: P) -> Publishers.DropUntilOutput<Self, P> where P : Publisher, Self.Failure == P.Failure

파라미터로 받은 publisher로부터 항목을 받기 전까지는 upstream publisher의 항목을 무시하는 operator

operator를 호출한 publisher를 A, 파라미터로 받은 publisher를 B라고 하면..

• B가 항목을 생성하기 전까지는 upstream publisher(A)의 항목을 drop 하는 publisher를 반환한다.
• A는 B에게 하나의 값을 요청하고, B가 값을 생성하기 전까지는 upstream publisher(A)의 모든 항목들을 무시(drop)한다.
• B가 항목을 생성하면 이 operator는 B의 구독을 취소하고 A의 이벤트가 지나가도록 허용한다.
• operator가 A로부터 구독을 받으면 downstream에서 upstream publisher로 backpressure request(?)을 통과한다.
  • Backpressure: subscriber가 항목을 받을 준비가 되어 있다는 신호를 보내서 흐름을 제어하는 것 (공식 문서)
  • Backpressure in Combine
    • subscriber가 받는 항목들을 제한하는 것
    • 데이터를 consume하는 속도보다 publish하는 속도가 더 빠를 때 메모리 이슈 등이 발생할 수 있기 때문
• B가 항목을 생성하기 전 upstream publisher가 이 요청에 대해 동작하면, operator는 upstream publisher로부터 받은 항목을 drop한다.

After this publisher receives a subscription from the upstream publisher, it passes through backpressure requests from downstream to the upstream publisher. If the upstream publisher acts on those requests before the other publisher produces an item, this publisher drops the elements it receives from the upstream publisher.

let upstream = PassthroughSubject<Int,Never>()
let second = PassthroughSubject<String,Never>()
cancellable = upstream
    .drop(untilOutputFrom: second)
    .sink { print("\($0)", terminator: " ") }

upstream.send(1)
upstream.send(2)
second.send("A")
upstream.send(3)
upstream.send(4)
// Prints "3 4"

Publishers.DropUntilOutput

struct DropUntilOutput<Upstream, Other> where Upstream : Publisher, Other : Publisher, Upstream.Failure == Other.Failure

두 번째 publisher의 항목을 받기 전까지는 upstream publisher의 항목을 무시하는 publisher

dropFirst()

func dropFirst(_ count: Int = 1) -> Publishers.Drop<Self>

특정 개수의 항목들을 drop 한 후에 항목들을 publish 하는 operator

• count 의 기본값은 1
• count 로 전달받은 개수만큼의 항목은 publish하지 않는 publisher를 리턴한다.

let numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
cancellable = numbers.publisher
    .dropFirst(5)
    .sink { print("\($0)", terminator: " ") }

// Prints: "6 7 8 9 10 "

drop(while:)

func drop(while predicate: @escaping (Self.Output) -> Bool) -> Publishers.DropWhile<Self>

클로저가 false를 리턴할 때까지 upstream publisher의 값을 무시하고, 나머지 항목들을 publish하는 operator

• predicate : publisher로부터 받은 항목을 drop 할지 여부를 반환하는 클로저

let numbers = [-62, -1, 0, 10, 0, 22, 41, -1, 5]
cancellable = numbers.publisher
    .drop { $0 <= 0 }
    .sink { print("\($0)") }

// Prints: "10 0, 22 41 -1 5"

한 번 클로저가 false를 반환하고 나면 그 이후 모든 값들을 publish한다.

tryDrop(while:)

func tryDrop(while predicate: @escaping (Self.Output) throws -> Bool) -> Publishers.TryDropWhile<Self>

• 클로저가 Boolean 값을 리턴하거나 에러를 던진다.

• 클로저가 에러를 던지면 publisher는 에러와 함께 실패한다.

  • 이때는 아무 값도 방출되지 않음!
  • 에러를 던지기 전에 클로저가 false 를 반환하면 에러도 던지지 않고 무조건 publish 한다. → filter 는 조건에 만족하는 항목을 만난 후에도 계속 클로저의 조건을 확인하지만, drop(while:) 은 한 번 클로저의 조건을 만족하면 그 이후에는 조건을 다시 확인하지 않는다.

struct RangeError: Error {}
var numbers = [1, 2, 3, 4, 5, 6, -1, 7, 8, 9, 10]
let range: CountableClosedRange<Int> = (1...100)
cancellable = numbers.publisher
    .tryDrop {
        guard $0 != 0 else { throw RangeError() }
        return range.contains($0)
    }
    .sink(
        receiveCompletion: { print ("completion: \($0)") },
        receiveValue: { print ("value: \($0)") }
    )

// Prints: "-1 7 8 9 10 completion: finished"
// If instead numbers was [1, 2, 3, 4, 5, 6, 0, -1, 7, 8, 9, 10], tryDrop(while:) would fail with a RangeError.

prefix(_:)

func prefix(_ maxLength: Int) -> Publishers.Output<Self>

파라미터로 전달한 개수만큼의 항목을 publish한다.

⇒ upstream publisher가 방출할 수 있는 값의 개수를 제어

prefix라는 이름 그대로 먼저 항목을 publish 한 후에 완료한다. dropFirst와 반대되는 개념!

• maxLength : publish할 항목의 최대 개수
• downstream subscriber로 publish할 항목의 개수를 제한할 때 사용한다.
• 설정한 개수만큼 publish한 후에는 바로 종료된다.

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

// Prints: "0 1"

prefix(while:)

func prefix(while predicate: @escaping (Self.Output) -> Bool) -> Publishers.PrefixWhile<Self>

클로저의 조건을 만족하는 값을 방출할 때까지만 다시 publish하는 operator

• predicate : 하나의 항목을 파라미터로 받고 계속 publish 해야 하는지 여부를 반환하는 클로저
• upsteram publisher로부터 받은 항목이 조건을 만족하는 동안만 값을 방출할 때 사용한다.
• 클로저가 false 를 리턴하면 publisher는 종료된다.

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

// Prints: "0 1 2 3 4"

tryPrefix(while:)

func tryPrefix(while predicate: @escaping (Self.Output) throws -> Bool) -> Publishers.TryPrefixWhile<Self>

• 클로저가 Boolean 값을 반환하거나 에러를 던진다.
  • 클로저가 에러를 던지면 publisher는 에러와 함께 실패한다.

struct OutOfRangeError: Error {}

let numbers = (0...10).reversed()
cancellable = numbers.publisher
    .tryPrefix {
        guard $0 != 0 else {throw OutOfRangeError()}
        return $0 <= numbers.max()!
    }
    .sink(
        receiveCompletion: { print ("completion: \($0)", terminator: " ") },
        receiveValue: { print ("\($0)", terminator: " ") }
    )

// Prints: "10 9 8 7 6 5 4 3 2 1 completion: failure(OutOfRangeError()) "