Combine | (III) Combine Operator：时间操作｜序列

时刻操作操作符(Time Manipulation Operators)

反应式编后的中心思维是跟着时刻的推移处理异步事情流。Combine 供给了一系列答应咱们和时刻相关的 Operator：跟着时刻的推移，序列对值做出处理。Combine 办理序列的时刻维度简略直接，这是 Combine 这的优势。

本节演示运用的 Playground 根底代码：

import Combine
import Foundation
import PlaygroundSupport
PlaygroundPage.current.needsIndefiniteExecution = true
func example(_ desc: String, _ action:() -> Void) {
    print("--- \(desc) ---")
    action()
}
var subscriptions = Set<AnyCancellable>()
Timer.publish(every: 1.0, on: .main, in: .common)
    .autoconnect()
    .scan(-1, { last, _ in return last + 1 })
    .sink { print("\($0) second has passed...") }
    .store(in: &subscriptions)

咱们运用 needsIndefiniteExecution 使 Playground 无限的执行。example 是现已熟知的帮咱们封装每个 example 的办法。subscriptions 帮咱们处理 Subscription。最终的 Timer 帮咱们在每 1 秒曩昔后打印信息。

Timer 是 Foundation Timer 类的 Combine 扩展。 它需求一个 RunLoop 和 RunLoop.Mode，Timer 是可衔接的(connectable) Publisher 类的一部分，需求在开始宣布值之前被衔接。咱们运用 autoconnect() 在第一次订阅时当即衔接。咱们将在后文了解更多 Timer 的信息。

时移值

delay(for:tolerance:scheduler:options)

func delay<S>(
    for interval: S.SchedulerTimeType.Stride,
    tolerance: S.SchedulerTimeType.Stride? = nil,
    scheduler: S,
    options: S.SchedulerOptions? = nil
) -> Publishers.Delay<Self, S> where S : Scheduler

最基本的时刻操作 Operator 是推迟来自 Publisher 的值，使其比实际呈现的时刻晚。delay(for:tolerance:scheduler:options) Operator 对整个值序列进行时移：每次上游 Publisher 宣布一个值时，该 Operator 都会将其保存一段时刻，然后在要求的推迟后，在指定的 Scheduler 上宣布值。

在 Playground 上增加代码：

example("delay") {
    let valuesPerSecond = 1.0
    let delayInSeconds = 1.2
    let sourcePublisher = PassthroughSubject<Date, Never>()
    let delayedPublisher = sourcePublisher.delay(for: .seconds(delayInSeconds), scheduler: DispatchQueue.main)
    let subscription = Timer
        .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
        .autoconnect()
        .subscribe(sourcePublisher)
        .store(in: &subscriptions)
    sourcePublisher
        .sink {  print("sourcePublisher: \($0)") }
        .store(in: &subscriptions)
    delayedPublisher
        .sink {  print("delayedPublisher: \($0)") }
        .store(in: &subscriptions)
}

咱们界说了两个常量，别离表明每秒宣布值的数量、推迟宣布值的秒。

咱们界说了 sourcePublisher，咱们将供给 Timer 宣布的 Date 类型的值。 值实际的类型其实并不重要，咱们只关心 Publisher 在宣布值被推迟。

delayPublisher 将推迟来自 sourcePublisher 的值，并将它们发送到 DispatchQueue.main。咱们后续将了解所有关于 Scheduler 的更多内容。

创立一个在主线程上每秒发送一个值的 Timer。 运用 autoconnect() 当即发动它，并经过 sourcePublisher 接纳它宣布的值。

最终，咱们再别离订阅 sourcePublisher 和 delayPublisher，增加一些 print 了解正在有值被宣布。

咱们用弹珠图描绘上述进程，sourcePublisher 宣布的值，将被推迟 1.2 秒后，被 delayPublisher 宣布：

运转 Playground，咱们将看到：

--- delay ---
0 second has passed...
sourcePublisher: 2022-12-10 07:46:26 +0000
1 second has passed...
sourcePublisher: 2022-12-10 07:46:27 +0000
delayedPublisher: 2022-12-10 07:46:26 +0000
2 second has passed...
sourcePublisher: 2022-12-10 07:46:28 +0000
delayedPublisher: 2022-12-10 07:46:27 +0000
3 second has passed...
sourcePublisher: 2022-12-10 07:46:29 +0000
delayedPublisher: 2022-12-10 07:46:28 +0000
4 second has passed...
sourcePublisher: 2022-12-10 07:46:30 +0000
delayedPublisher: 2022-12-10 07:46:29 +0000
...

搜集值

collect(_:options:)

func collect<S>(
    _ strategy: Publishers.TimeGroupingStrategy<S>,
    options: S.SchedulerOptions? = nil
) -> Publishers.CollectByTime<Self, S> where S : Scheduler

在某些情况下，咱们或许需求以指定的时刻距离从 Publisher 那里搜集值，这是一种缓冲方式。 例如，咱们想要搜集某时刻段内的一组值并核算平均值并输出。

将之前 Playground 的 example 代码注释或删除，增加以下代码：

example("collect") {
    let valuesPerSecond = 1.0
    let collectTimeStride = 3.0
    let sourcePublisher = PassthroughSubject<Date, Never>()
    let collectedPublisher = sourcePublisher
        .collect(.byTime(DispatchQueue.main, .seconds(collectTimeStride)))
    let subscription = Timer
        .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
        .autoconnect()
        .subscribe(sourcePublisher)
        .store(in: &subscriptions)
    sourcePublisher
        .sink {  print("sourcePublisher: \($0)") }
        .store(in: &subscriptions)
    collectedPublisher
        .sink {  print("collectedPublisher:\($0)") }
        .store(in: &subscriptions)
}

咱们界说了两个常量，别离表明每秒宣布值的数量、期望搜集一次值的时刻距离。

咱们界说了 sourcePublisher，将接纳 Timer 宣布的 Date 类型的值并再宣布。

collectedPublisher 将搜集来自 sourcePublisher 的值，并将它们发送到 DispatchQueue.main。

最终，咱们再别离订阅 sourcePublisher 和 collectedPublisher。

咱们用弹珠图描绘上述进程，sourcePublisher 宣布的值，将被 collectedPublisher 搜集宣布：

运转 Playground，咱们将看到：

--- collect ---
0 second has passed...
sourcePublisher: 2022-12-10 08:16:19 +0000
1 second has passed...
sourcePublisher: 2022-12-10 08:16:20 +0000
2 second has passed...
sourcePublisher: 2022-12-10 08:16:21 +0000
collectedPublisher:[2022-12-10 08:16:19 +0000, 2022-12-10 08:16:20 +0000, 2022-12-10 08:16:21 +0000]
3 second has passed...
sourcePublisher: 2022-12-10 08:16:22 +0000
4 second has passed...
sourcePublisher: 2022-12-10 08:16:23 +0000
5 second has passed...
sourcePublisher: 2022-12-10 08:16:24 +0000
collectedPublisher:[2022-12-10 08:16:22 +0000, 2022-12-10 08:16:23 +0000, 2022-12-10 08:16:24 +0000]
6 second has passed...
sourcePublisher: 2022-12-10 08:16:25 +0000
7 second has passed...
sourcePublisher: 2022-12-10 08:16:26 +0000

推迟事情

咱们或许期望在用户查找时，请求回来输入联想的展现。当然，不能在用户输入一个字母时都发送请求，需求有某种机制来操控请求的时机：能够仅在用户完结一段时刻的输入后。

Combine 供给了两个能够在这里为咱们供给帮助的 Operator：防抖(Debounce)和节省(Throttle)。

debounce(for:scheduler:options:)

func debounce<S>(
    for dueTime: S.SchedulerTimeType.Stride,
    scheduler: S,
    options: S.SchedulerOptions? = nil
) -> Publishers.Debounce<Self, S> where S : Scheduler

debounce Operator 经过指定的时刻距离后才会发布值。用来对从上游 Publisher 传递值，和传递值之间的时刻做操控。此 Operator 可用于处理突发事情流或大量事情流，咱们需求将传递到下流的值的数量削减到指定的速率。

注释调之前的代码，包含咱们记时用的 Timer，增加以下代码：

example("debounce") {
    let subject = PassthroughSubject<Int, Never>()
    let debounced = subject.debounce(for: .seconds(0.5), scheduler: DispatchQueue.main)
    subject
        .sink {  print("subject: \($0)") }
        .store(in: &subscriptions)
    debounced
        .sink {  print("debounced: \($0)") }
        .store(in: &subscriptions)
    let bounces:[(Int,TimeInterval)] = [
        (0, 0),
        (1, 0.3),   // 0.3s interval since last index
        (2, 1),     // 0.7s interval since last index
        (3, 1.3),   // 0.3s interval since last index
        (4, 1.5),   // 0.2s interval since last index
        (5, 2.1)    // 0.6s interval since last index
    ]
    for bounce in bounces {
        DispatchQueue.main.asyncAfter(deadline: .now() + bounce.1) {
            subject.send(bounce.0)
        }
    }
}

咱们让 subject 每隔 bounces 元组的第二个 TimeInterval 值宣布元组的第二个 Int 值。

咱们运用 debounced 限定了抵达速率为 0.5。即值在 0.5 秒内，没有下一个值被宣布，则该值可传递到下流。

只要值 1、值 4、值 5 后 0.5 秒内，没有值宣布。因而运转 Playground，将输出：

--- debounce ---
subject: 0
subject: 1
debounced: 1
subject: 2
subject: 3
subject: 4
debounced: 4
subject: 5
debounced: 5

上述进程，咱们运用弹珠图来描绘：

throttle(for:scheduler:latest:)

func throttle<S>(
    for interval: S.SchedulerTimeType.Stride,
    scheduler: S,
    latest: Bool
) -> Publishers.Throttle<Self, S> where S : Scheduler

throttle 在指定的时刻距离内，发布由上游 Publisher 发布的第一个(latest 为 false)或最终一个(latest 为 true)值。

以上一个例子为根底，咱们在 Playground 中增加代码：

example("throttle") {
    let subject = PassthroughSubject<Int, Never>()
    let throttled = subject.throttle(for: .seconds(1), scheduler: DispatchQueue.main, latest: false)
    subject
        .sink {  print("subject: \($0)") }
        .store(in: &subscriptions)
    throttled
        .sink {  print("throttled: \($0)") }
        .store(in: &subscriptions)
    let values:[(Int,TimeInterval)] = [
        (0, 0),
        (1, 0.1),
        (2, 0.5),
        (3, 3.5),
        (4, 3.9),
        (5, 4.1),
        (6, 4.4),
    ]
    for bounce in bounces {
        DispatchQueue.main.asyncAfter(deadline: .now() + bounce.1) {
            subject.send(bounce.0)
        }
    }
}

因为异步，时刻距离过短或许卡时刻点，都会有不准确的问题。咱们放大了示例代码中的时刻。

咱们指定的时刻为 1 秒。throttled 直接发布 0，在 1 秒时，从 (0, 1) 秒区间找到并发布 1。后续因为超越了 1 秒，收到 3 就直接发布，在 (3.5, 4.5) 秒区间找到并发布 1：

--- throttle ---
subject: 0
throttled: 0
subject: 1
subject: 2
throttled: 1
subject: 3
throttled: 3
subject: 4
subject: 5
subject: 6
throttled: 4

上述进程用弹珠图来描绘为：

咱们将 latest 改为 true：

 let throttled = subject.throttle(for: .seconds(1), scheduler: DispatchQueue.main, latest: true)

则会输出每个区间中最新的值：

--- throttle ---
subject: 0
throttled: 0
subject: 1
subject: 2
throttled: 2
subject: 3
throttled: 3
subject: 4
subject: 5
subject: 6
throttled: 6

现在咱们有了防抖动和节省的底子差异：

• 防抖等候它接纳到的值的事情暂停，在指定的时刻距离后宣布最新的值。
• 节省等候指定的时刻距离，然后宣布它在该时刻距离内收到的第一个或最新的值。

超时

timeout(_:scheduler:options:customError:)

func timeout<S>(
    _ interval: S.SchedulerTimeType.Stride,
    scheduler: S,
    options: S.SchedulerOptions? = nil,
    customError: (() -> Self.Failure)? = nil
) -> Publishers.Timeout<Self, S> where S : Scheduler

当超时触发时， Publisher 要不宣布 .finished ，要不宣布咱们指定的错误：

example("timeout") {
    enum TimeoutError: Error {
        case timedOut
    }
    let subject = PassthroughSubject<Void, TimeoutError>()
    let timedOutSubject1 = subject.timeout(.seconds(3), scheduler: DispatchQueue.main)
    let timedOutSubject2 = subject.timeout(.seconds(3), scheduler: DispatchQueue.main, customError: { .timedOut })
    timedOutSubject1
        .sink(
            receiveCompletion: { print("timedOutSubject1: \($0)") },
            receiveValue: {  print("timedOutSubject1: \($0)") }
        )
        .store(in: &subscriptions)
    timedOutSubject2
        .sink(
            receiveCompletion: { print("timedOutSubject2: \($0)") },
            receiveValue: {  print("timedOutSubject2: \($0)") }
        )
        .store(in: &subscriptions)
}

在上述代码中，timedOutSubject1 没有供给 customError 字段，而 timedOutSubject2 供给了。因而，当 subject 超越 3 秒未宣布事情，将触发超时：

--- timeout ---
timedOutSubject1: finished
timedOutSubject2: failure(Page_Contents.(unknown context at $10f5bfc44).(unknown context at $10f5bfc6c).(unknown context at $10f5bfcac).TimeoutError.timedOut)

时刻丈量

measureInterval(using:options:)

有时咱们需求找出 Publisher 宣布的两个连续值之间经过的时刻时，measureInterval Operator 是咱们的工具。

增加代码：

example("measureInterval") {
    let subject = PassthroughSubject<Int, Never>()
    let measureSubject = subject.measureInterval(using: DispatchQueue.main)
    subject.sink { print("emitted: \($0)") }
    .store(in: &subscriptions)
    measureSubject.sink { print("Measure emitted: \(Double($0.magnitude) / 1_000_000_000.0)") }
    .store(in: &subscriptions)
    let bounces:[(Int,TimeInterval)] = [
        (0, 0),
        (1, 0.1),
        (2, 0.5),
        (3, 3.5),
        (4, 3.9),
        (5, 4.1),
        (6, 4.4),
    ]
    for bounce in bounces {
        DispatchQueue.main.asyncAfter(deadline: .now() + bounce.1) {
            subject.send(bounce.0)
        }
    }
}

measureSubject 将宣布 subject 每次宣布的值距离上次宣布值的距离。

因为measureInterval 在 DispatchQueue 的情况下，TimeInterval 解释为：运用此类型的值创立的 DispatchTimeInterval，以纳秒为单位。因而进行了 Double($0.magnitude) / 1_000_000_000.0 转化：

--- measureInterval ---
Measure emitted: 0.010662459
emitted: 0
Measure emitted: 0.09147425
emitted: 1
Measure emitted: 0.421239375
emitted: 2
Measure emitted: 3.15006375
emitted: 3
Measure emitted: 0.4142835
emitted: 4
Measure emitted: 0.014388208
emitted: 5
Measure emitted: 0.517094042
emitted: 6

假如咱们进行以下更改，运用 RunLoop:

let measureSubject2 = subject.measureInterval(using: RunLoop.main)

则无需进行上述 Double($0.magnitude) / 1_000_000_000.0 的转化：

--- measureInterval ---
emitted: 0
Measure emitted: Stride(magnitude: 0.008463025093078613)
emitted: 1
Measure emitted: Stride(magnitude: 0.0957329273223877)
emitted: 2
Measure emitted: Stride(magnitude: 0.4204070568084717)
emitted: 3
Measure emitted: Stride(magnitude: 3.1296679973602295)
emitted: 4
Measure emitted: Stride(magnitude: 0.4410020112991333)
emitted: 5
Measure emitted: Stride(magnitude: 0.2083679437637329)
emitted: 6
Measure emitted: Stride(magnitude: 0.108254075050354)

序列操作符(Sequence Operators)

Publisher 自身便是序列。序列 Operator 与 Publisher 的值一同运用，就像 Array 或 Set 相同——当然，它们是有限序列。考虑到这一点，序列 Operator 主要将 Publisher 作为一个全体来处理，而不是像其他 Operator 那样处理单个值。此类中的许多 Operator 的名称和行为与 Swift 规范库中的对应办法几乎相同。

寻觅值

min() max()

func min() -> Publishers.Comparison<Self>
func max() -> Publishers.Comparison<Self>

min Operator 帮咱们找到 Publisher 宣布的最小值。 它是贪婪的，意味着必须等候 Publisher 发送一个完结事情后， Operator 的下流会宣布最小值：

上述弹珠图用代码表明为：

example("min") {
    let publisher = [1, -5, 10, 0].publisher
    publisher
        .print("publisher: ")
        .min()
        .sink(receiveCompletion: { print($0) }, 
              receiveValue: { print("Lowest value is \($0)") })
        .store(in: &subscriptions)
}

运转 Playground，将输出：

--- min ---
publisher: : receive subscription: ([1, -5, 10, 0])
publisher: : request unlimited
publisher: : receive value: (1)
publisher: : receive value: (-5)
publisher: : receive value: (10)
publisher: : receive value: (0)
publisher: : receive finished
Lowest value is -5
finished

将在 publisher 完结后，才会宣布最小值再完结。

Combine 知道这些数字中的哪一个是最小值，要归功于 Int 符合 Comparable 协议。咱们以在宣布 Comparable 类型的 Publisher 上直接运用 min() ，无需任何参数。

假如咱们的值不符合 Comparable ，咱们能够运用 min(by:) Operator 供给自己的比较闭包：

考虑以下示例，你的发布者宣布许多数据，而你期望找到最小的数据。

在以下代码示例中，咱们比较了 Data 的长度：

example("min non-Comparable") {
    let publisher =  ["12345",
                     "ab",
                     "!!@@##$$"]
        .map { Data($0.utf8) }
        .publisher
        .print("publisher")
        .min(by: { $0.count < $1.count })
        .sink(receiveCompletion: { print($0) }, 
              receiveValue: { data in
            let string = String(data: data, encoding: .utf8)!
            print("Smallest data is \(string), \(data.count) bytes")
        })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- min non-Comparable ---
publisher: receive subscription: ([5 bytes, 2 bytes, 8 bytes])
publisher: request unlimited
publisher: receive value: (5 bytes)
publisher: receive value: (2 bytes)
publisher: receive value: (8 bytes)
publisher: receive finished
Smallest data is ab, 2 bytes
finished

max Operator 同理，不在进行赘述。

first() last()

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

first 让第一个宣布的值经过然后完结。它是 lazy 的，这意味着它不会等候上游发布者完结，而是会在接纳到宣布的第一个值时撤销订阅。而 last 是贪婪的，需求在上游宣布完结事情后才会完结：

上述弹珠图用代码表明为：

example("first and last") {
    let publisher = [1, 2, 3, 4].publisher
    publisher
        .print("first: ")
        .first()
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print("First value is \($0)") })
        .store(in: &subscriptions)
    publisher
        .print("last: ")
        .last()
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print("Last value is \($0)") })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- first and last ---
first: : receive subscription: ([1, 2, 3, 4])
first: : request unlimited
first: : receive value: (1)
first: : receive cancel
First value is 1
finished
last: : receive subscription: ([1, 2, 3, 4])
last: : request unlimited
last: : receive value: (1)
last: : receive value: (2)
last: : receive value: (3)
last: : receive value: (4)
last: : receive finished
Last value is 4
finished

假如咱们需求更精细的操控，能够运用 first(where:)、last(where:)。 假如有的话，它将宣布与供给的条件匹配的第一个、最终一个值：

修正上述代码：

example("first and last") {
    let publisher = [1, 2, 3, 4].publisher
    publisher
        .print("first: ")
        .first(where: { $0 % 2 == 0 })
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print("First value is \($0)") })
        .store(in: &subscriptions)
    publisher
        .print("last: ")
        .last(where: { $0 % 3 == 0 })
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print("Last value is \($0)") })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- first and last ---
first: : receive subscription: ([1, 2, 3, 4])
first: : request unlimited
first: : receive value: (1)
first: : receive value: (2)
first: : receive cancel
First value is 2
finished
last: : receive subscription: ([1, 2, 3, 4])
last: : request unlimited
last: : receive value: (1)
last: : receive value: (2)
last: : receive value: (3)
last: : receive value: (4)
last: : receive finished
Last value is 3
finished

假如没有满意条件的值，下流也将直接纳到完结事情，不会有值宣布。

output(at:) output(in:)

output Operator 将查找上游发布者在指定索引处宣布的值：

上述弹珠图用代码表明为：

example("output") {
    let publisher = [0, 1, 2, 3, 4].publisher
    publisher
        .print("output at: ")
        .output(at: 1)
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
    publisher
        .print("output in: ")
        .output(in: 1...3)
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- output ---
output at: : receive subscription: ([0, 1, 2, 3, 4])
output at: : request unlimited
output at: : receive value: (0)
output at: : request max: (1) (synchronous)
output at: : receive value: (1)
1
output at: : receive cancel
finished
output in: : receive subscription: ([0, 1, 2, 3, 4])
output in: : request unlimited
output in: : receive value: (0)
output in: : request max: (1) (synchronous)
output in: : receive value: (1)
1
output in: : receive value: (2)
2
output in: : receive value: (3)
3
output in: : receive cancel
finished

该 Operator 会在收到所供给范围内的所有值后当即撤销订阅。

查询值

count()

func count() -> Publishers.Count<Self>

count Operator 宣布单个值， 一旦上游 Publisher 发布完结事情，Operator 将宣布接纳到的值的数量：

上述弹珠图用代码表明为：

example("count") {
    let publisher = [1, 2, 3, 4].publisher
    publisher
        .print("publisher")
        .count()
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- count ---
publisher: receive subscription: ([1, 2, 3, 4])
publisher: request unlimited
publisher: receive value: (1)
publisher: receive value: (2)
publisher: receive value: (3)
publisher: receive value: (4)
publisher: receive finished
4
finished

正如预期的那样，只要在上游 Publisher 发送完结事情后，才会打印出值 4。

contains(_: contains(where:)

func contains(_ output: Self.Output) -> Publishers.Contains<Self>
func contains(where predicate: @escaping (Self.Output) -> Bool) -> Publishers.ContainsWhere<Self>

假如上游 Publisher 宣布指定的值，则 contains 操作符将宣布 true 并当即撤销订阅，假如宣布的值都不等于指定的值，则回来 false：

example("contains") {
    let publisher = [1, 2, 3, 4].publisher
    publisher
        .print("publisher")
        .contains(3)
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
    publisher
        .print("publisher")
        .contains(where: { $0 % 5 == 0 })
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- contains ---
publisher: receive subscription: ([1, 2, 3, 4])
publisher: request unlimited
publisher: receive value: (1)
publisher: receive value: (2)
publisher: receive value: (3)
publisher: receive cancel
true
finished
publisher: receive subscription: ([1, 2, 3, 4])
publisher: request unlimited
publisher: receive value: (1)
publisher: receive value: (2)
publisher: receive value: (3)
publisher: receive value: (4)
publisher: receive finished
false
finished

allSatisfy(_:)

allSatisfy 承受一个闭包，宣布一个布尔值，指示上游 Publisher 宣布的所有值是否与条件匹配。它是贪婪的，若每个值都满意，会比及上游 Publisher 宣布完结完结事情，否则撤销订阅：

example("allSatisfy") {
    let publisher = stride(from: 0, to: 5, by: 2).publisher
    publisher
        .print("publisher")
        .allSatisfy { $0 % 2 == 0 }
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { allEven in
            print(allEven ? "All numbers are even" : "Something is odd...")
        })
        .store(in: &subscriptions)
    publisher
        .print("publisher")
        .allSatisfy { $0 % 3 == 0 }
        .sink(receiveCompletion: { print($0) }, 
            receiveValue: { print($0) })
        .store(in: &subscriptions)
}

运转 Playground 将输出：

--- allSatisfy ---
publisher: receive subscription: (Sequence)
publisher: request unlimited
publisher: receive value: (0)
publisher: receive value: (2)
publisher: receive value: (4)
publisher: receive finished
All numbers are even
finished
publisher: receive subscription: (Sequence)
publisher: request unlimited
publisher: receive value: (0)
publisher: receive value: (2)
publisher: receive cancel
false
finished

reduce(_:_:)

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

reduce Operator 它不查找特定值或查询整个 Publisher。 它答应咱们依据上游 Publisher 的值，迭代累积一个新值：

咱们用代码描绘上述弹珠图：

example("reduce") {
    let publisher = ["He", "llo", " ", "Wo", "rld", "!"].publisher
    publisher
        .print("publisher")
        .reduce("") { accumulator, value in
            accumulator + value
        }
        .sink(receiveValue: { print("Reduced into: \($0)") })
        .store(in: &subscriptions)
}

在此代码中，咱们创立一个宣布六个字符串的 publisher。将 reduce 与空字符串一同运用，将宣布的值附加到它上面，创立最终的字符串结果：

--- reduce ---
publisher: receive subscription: (["He", "llo", " ", "Wo", "rld", "!"])
publisher: request unlimited
publisher: receive value: (He)
publisher: receive value: (llo)
publisher: receive value: ( )
publisher: receive value: (Wo)
publisher: receive value: (rld)
publisher: receive value: (!)
publisher: receive finished
Reduced into: Hello World!

reduce 的第二个参数是一个闭包，它承受两个某种类型的值并回来一个相同类型的值。在 Swift 中，+ 也是一个匹配该签名的函数，咱们完全能够改写代码：

example("reduce") {
    let publisher = ["He", "llo", " ", "Wo", "rld", "!"].publisher
    publisher
        .print("publisher")
        .reduce("", +)
        .sink(receiveValue: { print("Reduced into: \($0)") })
        .store(in: &subscriptions)
}

内容参考

• Combine | Apple Developer Documentation；
• 来自 Kodeco 的书本《Combine: Asynchronous Programming with Swift》；
• 对上述 Kodeco 书本的汉语自译版 《Combine: Asynchronous Programming with Swift》 的收拾、修正、弥补更新。