Swift Combine

简介

Combine是Apple在2019年WWDC上推出的一个新框架。该框架提供了一个声明性的Swift API，用于随时间处理值。这些值可以表示多种异步事件。

Publisher协议声明了一种可以随时间传递一系列值的类型。Operators根据从upstream publishers接受到的值采取行动，并重新发布这些值。

在publishers链的末尾，Subscriber在接收元素时对其进行操作。Publisher仅在Subscriber明确请求时才会发出值。

通过采用Combine，通过集中事件处理代码并消除嵌套闭包和基于约定的回调等麻烦的技术，使代码更易于阅读和维护。

Combine 是基于泛型实现的，是类型安全的。它可以无缝地接入已有的工程，用来处理现有的 Target/Action、Notification、KVO、callback/closure 以及各种异步网络请求。

在 Combine 中，有几个重要的组成部分：

发布者：Publiser
订阅者：Subscriber
操作符：Operator

overview.png

Publisher

在 Combine 中，Publisher 相当于RxSwift中的 Observable，并且可以通过组合变换（Operator）重新生成新的 Publisher。

public protocol Publisher {

    /// The kind of values published by this publisher.
    associatedtype Output

    /// The kind of errors this publisher might publish.
    ///
    /// Use `Never` if this `Publisher` does not publish errors.
    associatedtype Failure : Error

    /// This function is called to attach the specified `Subscriber` to this `Publisher` by `subscribe(_:)`
    ///
    /// - SeeAlso: `subscribe(_:)`
    /// - Parameters:
    ///     - subscriber: The subscriber to attach to this `Publisher`.
    ///                   once attached it can begin to receive values.
    func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

在 Publisher 的定义中，Output 代表数据流中输出的值，值的更新可能是同步，也可能是异步，Failure 代表可能产生的错误，也就是说 Pubslier 最核心的是定义了值与可能的错误。Publisher 通过 receive(subscriber:) 用来接受订阅，并且要求 Subscriber 的值和错误类型要一致来保证类型安全。

看一个例子：

let justPubliser = Just("Hello")

justPubliser 会给每个订阅者发送一个 "Hello" 消息，然后立即结束（这个数据流只包含一个值）。

Combine提供了一个 enum Publishers，包括：

struct Empty : 一个从不发布任何值的publisher，并且可以选择立即完成。
struct Fail : 立即使用指定错误终止的publisher。
struct Once: 只有一次向每个订阅者发布输出然后完成的publisher，或者在没有生成任何元素的情况下立即失败的publisher。
struct Optional : 如果可选值具有值，则publisher仅向每个订阅者发布一次可选值。
struct Sequence : 发布给定元素序列的publisher。
struct Deferred : 在运行提供的闭包之前等待订阅的发布者，以便为新订阅者创建发布者。
...

Subscriber

Subscriber相当于RxSwift中的Observer。

public protocol Subscriber : CustomCombineIdentifierConvertible {

    /// The kind of values this subscriber receives.
    associatedtype Input

    /// The kind of errors this subscriber might receive.
    ///
    /// Use `Never` if this `Subscriber` cannot receive errors.
    associatedtype Failure : Error

    /// Tells the subscriber that it has successfully subscribed to the publisher and may request items.
    ///
    /// Use the received `Subscription` to request items from the publisher.
    /// - Parameter subscription: A subscription that represents the connection between publisher and subscriber.
    func receive(subscription: Subscription)

    /// Tells the subscriber that the publisher has produced an element.
    ///
    /// - Parameter input: The published element.
    /// - Returns: A `Demand` instance indicating how many more elements the subcriber expects to receive.
    func receive(_ input: Self.Input) -> Subscribers.Demand

    /// Tells the subscriber that the publisher has completed publishing, either normally or with an error.
    ///
    /// - Parameter completion: A `Completion` case indicating whether publishing completed normally or with an error.
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

可以看出,Publisher 在自身状态改变时，调用 Subscriber 的三个不同方法（receive(subscription), receive(_:Input), receive(completion:)）来通知 Subscriber。

image.png

这里也可以看出，Publisher 发出的通知有三种类型：

Subscription：Subscriber 成功订阅的消息，只会发送一次，取消订阅会调用它的 Cancel 方法来释放资源
Value（Subscriber 的 Input，Publisher 中的 Output）：真正的数据，可能发送 0 次或多次
Completion：数据流终止的消息，包含两种类型：.finished 和 .failure(Error)，最多发送一次，一旦发送了终止消息，这个数据流就断开了，当然有的数据流可能永远没有终止
大部分场景下我们主要关心的是后两种消息，即数据流的更新和终止。

Combine 内置的 Subscriber 有三种：

• Sink
• Assign
• Subject

Sink 是非常通用的 Subscriber，我们可以自由的处理数据流的状态。

let once: Publishers.Once<Int, Never> = Publishers.Once(100)
let observer: Subscribers.Sink<Int,Never> = Subscribers.Sink(receiveCompletion: {
    print("completed: \($0)")
}, receiveValue: {
    print("received value: \($0)")
})
once.subscribe(observer)

Assign 可以很方便地将接收到的值通过 KeyPath 设置到指定的 Class 上（不支持 Struct）

class Student {
    let name: String
    var score: Int
    
    init(name: String, score: Int) {
        self.name = name
        self.score = score
    }
}

let student = Student(name: "Jack", score: 90)
print(student.score)
let observer = Subscribers.Assign(object: student, keyPath: \.score)
let publisher = PassthroughSubject<Int, Never>()
publisher.subscribe(observer)
publisher.send(91)
print(student.score)
publisher.send(100)
print(student.score)

一旦 publisher 的值发生改变，相应的，student 的 score 也会被更新。

PassthroughSubject 这里是 Combine 内置的一个 Publisher。

Subject

有些时候我们想随时在 Publisher 插入值来通知订阅者，在 Rx 中也提供了一个 Subject 类型来实现。Subject 通常是一个中间代理，即可以作为 Publisher，也可以作为 Subscriber。Subject 的定义如下：

public protocol Subject : AnyObject, Publisher {

    /// Sends a value to the subscriber.
    ///
    /// - Parameter value: The value to send.
    func send(_ value: Self.Output)

    /// Sends a completion signal to the subscriber.
    ///
    /// - Parameter completion: A `Completion` instance which indicates whether publishing has finished normally or failed with an error.
    func send(completion: Subscribers.Completion<Self.Failure>)
}

作为 Subscriber 的时候，可以通过 Publisher 的 subscribe(_:Subject) 方法订阅某个 Publisher。

作为 Publisher 的时候，可以主动通过 Subject 的两个 send 方法，我们可以在数据流中随时插入数据。目前在 Combine 中，有三个已经实现对 Subject: AnySubject，CurrentValueSubject 和 PassthroughSubject 。

CurrentValueSubject : 包含单个值并且当值改变时发布新元素的subject

let a = CurrentValueSubject<Int, NSError>(1)
a.sink(receiveCompletion: {
    print("11\($0)")
}, receiveValue: {
    print("22\($0)")
})

a.value = 2
a.value = 3
a.send(4)
a.send(completion: Subscribers.Completion<NSError>.finished)
// a.send(completion: Subscribers.Completion<NSError>.failure(NSError(domain: "domain", code: 500, userInfo: ["errorMsg":"error"])))
a.value = 5

当subject send completion后（不管是finished还是failure）,subject不再发出元素

PassthroughSubject与CurrentValueSubject类似，只是设置初始值，也不会保存任何值。

let a = PassthroughSubject<Int,NSError>()
a.sink(receiveCompletion: {
    print("11\($0)")
}, receiveValue: {
    print("22\($0)")
})

a.send(4)
a.send(completion: Subscribers.Completion<NSError>.finished)
// a.send(completion: Subscribers.Completion<NSError>.failure(NSError(domain: "domain", code: 500, userInfo: ["errorMsg":"error"])))
a.send(5)

AnyPublisher、AnySubscriber、AnySubject

通用类型,任意的 Publisher、Subscriber、Subject 都可以通过 eraseToAnyPublisher()、eraseToAnySubscriber()、eraceToAnySubject() 转化为对应的通用类型。

let name = Publishers.Sequence<[String], Never>(sequence: ["1","2"]).eraseToAnyPublisher()

Cancellable

可以取消活动或操作的协议。

public protocol Cancellable {

    /// Cancel the activity.
    /// Calling `cancel()` frees up any allocated resources. It also stops side effects such as timers, network access, or disk I/O.
    func cancel()
}

Operator

操作符是 Combine 中非常重要的一部分，通过各式各样的操作符，可以将原来各自不相关的逻辑变成一致的（unified）、声明式的（declarative）的数据流。

转换操作符：

• map/mapError
• flatMap
• replaceNil
• scan
• setFailureType

过滤操作符：

• filter
• compactMap
• removeDuplicates
• replaceEmpty/replaceError

reduce 操作符：

• collect
• ignoreOutput
• reduce

运算操作符：

• count
• min/max

匹配操作符：

• contains
• allSatisfy

序列操作符：

• drop/dropFirst
• append/prepend
• prefix/first/last/output

组合操作符：

• combineLatest
• merge
• zip

错误处理操作符：

• assertNoFailure
• catch
• retry

时间控制操作符：

• measureTimeInterval
• debounce
• delay
• throttle
• timeout

其他操作符：

• encode/decode
• switchToLatest
• share
• breakpoint/breakpointOnError
• handleEvents

未完待续

参考：
https://icodesign.me/posts/swift-combine/