创建Publisher 以及对其进行订阅

新建Playground，添加工具方法用于输出示例代码的运行情况

public func example(of description: String, action: () -> Void) {
    print("\n——— Example of:", description, "———")
    action()
}

首先在swift标准库中我们可以使用如下方式发送和监听一个通知

/// 原始使用Notification的示例
example(of: "Notification") {
    let myNotification = Notification.Name(rawValue: "MyNotification")
    let observer = NotificationCenter.default.addObserver(forName: myNotification, object: nil, queue: nil) { _ in
        print("receive notification")
    }
    NotificationCenter.default.post(name: myNotification, object: nil, userInfo: nil)
    NotificationCenter.default.removeObserver(observer)
}

/*输出：
——— Example of: Notification ———
receive notifacation
*/

同样我们使用Combine后也可以对通知进行[图片上传中...(截屏2020-03-03下午2.51.48.png-66e51d-1583218332269-0)]
订阅

/// 使用Combine管理Notification的例子
import Combine
example(of: "Combine") {
    let myNotification = Notification.Name(rawValue: "MyNotification")
    //创建一个publisher
    let publisher = NotificationCenter.default.publisher(for: myNotification, object: nil)
    //创建一个订阅
    let subscription = publisher.sink { _ in
        print("receive notification")
    }
    NotificationCenter.default.post(name: myNotification, object: nil, userInfo: nil)
    //结束订阅
    subscription.cancel()
}

/*输出：
——— Example of: Combine ———
receive notifacation
*/

sink

注意以上代码中的skin方法，与其本意下沉似乎没有什么关系，我们在Xcode中查看其源代码声明：

extension Publisher where Self.Failure == Never {

    /// 用闭包添加一个subscriber 
    ///
    /// 此方法会创建一个 subscriber 并且立即订阅 and immediately requests an unlimited number of values, prior to returning the subscriber.
    /// 返回一个可取消的实例；用于在结束对接收值的使用时删除订阅流。
        public func sink(receiveValue: @escaping ((Self.Output) -> Void)) -> AnyCancellable
}

以上我们可以利用skin创建一个subscriber了，同时我们也发现skin还有另一个定义

Just

example(of: "Just") {
    let just = Just("Hello world!")
    _ = just.sink(
        receiveCompletion: {
            print("Received completion", $0)
        },
        receiveValue: {
            print("Received value", $0)
    })
}

/*输出
——— Example of: Just ———
Received value Hello world!
Received completion finished  
*/

skin的这种订阅可以分别订阅到完成事件以及发出的值事件

Just和RXSwift中一样会创建一个只发出发出一个元素和一个完成事件的publisher。

assign(to:on:)

除了skin之外，内置的assign（to:on:）操作符还允许您将接收到的值分配给对象的KVO属性。

example(of: "assign(to:on:)") {    
    class SomeObject {
    // 使用具有打印新值的didSet属性观察者的属性定义类。
        var value: String = "" {
            didSet { print(value)}
        }
    }
    // 创建该类的实例。
    let object = SomeObject()
    // 从字符串数组创建发布服务器。
    let publisher = ["Hello", "world!"].publisher
     // publisher将接收到的每个值分配给对象的value属性。
    _ = publisher.assign(to: \.value, on: object)
}
/*输出
——— Example of: assign(to:on:) ———
Hello
world!
*/

Cancellable

当订阅publisher完成并且不再希望从publisher接收值时，最好取消订阅以释放资源并停止任何相应的活动，例如网络调用。
订阅返回anyCancelable的实例作为“取消令牌”，这样在完成订阅后就可以取消订阅。
anyCancelable符合Cancelable协议，因此可以使用cancel()方法取消订阅。
例如在 example(of: "Combine") 例子中的subscription.cancel()就取消了对消息中心的订阅和removeObserver(observer)起到同样的效果。如果不调用cancel()那么直到publisher发出完成事假，订阅才会取消

注意：也可以忽略订阅的返回值（例如，_=just.sink…）。但是有一个警告：如果您不在整个项目中存储订阅，则一旦程序流退出其创建的范围(方法结束后订阅对象销毁，订阅结束)，该订阅将被取消！

Publisher

我们在Xcode中点击跳转到Publisher的定义

public protocol Publisher {
    /// 发送的值的类型
    associatedtype Output

    /// Publisher可能产生的错误类型；
    /// 如果保证Publisher不会产生错误，则使用`Never`。
    associatedtype Failure : Error

    /// 在调用 Publisher的`subscribe(_:)`方法时方法内部会调用此方法去附加`Subscriber`
    func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

还有关联的 Output和Failure一致是才能附加subscriber到publisher

Subscriber

public protocol Subscriber : CustomCombineIdentifierConvertible {

    /// 可以接收的值的类型
    associatedtype Input

    /// 可以接收的错误类型；如果不接收错误，则使用 `Never`
    associatedtype Failure : Error

    /// 调用此方法以提供订阅
    func receive(subscription: Subscription)

    /// 发送刚发布的新值
    func receive(_ input: Self.Input) -> Subscribers.Demand

    /// 发送错误或完成事件
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

Subscription

连接publisher 和 subscriber 是 subscription. 下面是 Subscription protocol的定义

public protocol Subscription : Cancellable, CustomCombineIdentifierConvertible {
    ///  告诉 publisher 可以发送多少个值到 subscriber.
    func request(_ demand: Subscribers.Demand)
}

subscriber说明其愿意接收多少值的概念称为backpressure。如果没有它或其他策略，subscriber可能会收到来自publisher的超出其处理能力的大量值，这可能会导致问题。在后面我们会详细介绍，你也可以浏览王巍 (@onevcat)的这篇文章关于 Backpressure 和 Combine 中的处理

注意Subscriber中的func receive(_ input: Self.Input) -> Subscribers.Demand方法，返回值是一个Subscribers.Demand，即使订阅最开始时设置了接受的最大值，我们在每次收到新的值都可以调整收到的最大值，这个值是累加的，并且传递负值时会直接返回 fatalError，所以这个值只会越来越大而不能减少它

自定义subscriber

example(of: "Custom Subscriber") {

    // 1 通过整数范围创建一个publisher
    let publisher = (1...6).publisher

    // 2 自定义一个整型的Subscriber实现协议
    final class IntSubscriber: Subscriber {
        // 3 指定接收值的类型和错误类型
        typealias Input = Int
        typealias Failure = Never

        // 4 实现协议方法 publisher会调用该方法
        func receive(subscription: Subscription) {
            //简单实现为接收订阅的值最多三个
            subscription.request(.max(3))
        }

        // 5 实现接受到值时的方法，返回接收值的最大个数变化
        func receive(_ input: Int) -> Subscribers.Demand {
            //这里简单实现为打印值，返回 .none,意思就是不改变最大接收数量 .none 等价于 .max(0).
            print("Received value", input)
            return .none
        }

        // 6 实现接收到完成事件的方法
        func receive(completion: Subscribers.Completion<Never>) {
            print("Received completion", completion)
        }
    }
    // 订阅publisher
    publisher.subscribe(IntSubscriber())
}

/*输出
——— Example of: Custom Subscriber ———
Received value 1
Received value 2
Received value 3
*/

这里在一开始指定了只接受3个值（.max(3)）并且在后续接受到新值时没有更改最大值(.none)，因此直接受到publisher发出的前三个值。如果我们将一开始的.none改成.unlimited或者接受到新值后我们每次加一最大值(.max(1))那么我们就会收到所有值，并且收到完成事件

输出结果应该会是：
——— Example of: Custom Subscriber ———
Received value 1
Received value 2
Received value 3
Received value 4
Received value 5
Received value 6
Received completion finished

Future

就像使用Just创建一个publisher一样，Future可以用于异步生成一个值，然后完成。
我们通过Xcode查看Future的定义

final public class Future<Output, Failure> : Publisher where Failure : Error {
    public typealias Promise = (Result<Output, Failure>) -> Void
    
    public init(_ attemptToFulfill: @escaping (@escaping Future<Output, Failure>.Promise) -> Void)
    final public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
}

Future是一个 publisher 会异步产生一个值然后完成或者失败，Promise是闭包的类型别名，它接收一个Reuslt包含输出的值或者失败

Result类型为Swift5新增内容，你可以查看Swift5新特性 & XCode 10.2更新获取更多信息