Combine 内置的 Subscriber

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)