Combine 框架，从0到1 —— 2.经过 ConnectablePublisher 控制什么时候发布

 

本文首发于 Ficow Shen's Blog，原文地址： Combine 框架，从0到1 —— 2.经过 ConnectablePublisher 控制什么时候发布。

 

内容概览

• 前言
• 使用 makeConnectable() 和 connect() 手动控制发布
• 使用 autoconnect() 操做符进行自动链接
• 总结

 

前言

 

使用 Connectable Publisher， 你能够决定发布者什么时候开始发送订阅元素给订阅者。那么，为何咱们须要这么作？

使用 sink(receiveValue:) 能够马上开始接收订阅元素，可是这可能不是你想要的结果。当多个订阅者订阅了同一个发布者时，有可能会出现其中一个订阅者收到订阅内容，而另一个订阅者收不到的状况。

好比，当你发起一个网络请求，并为这个请求建立了一个发布者以及链接了这个发布者的订阅者。

而后，这个订阅者的订阅操做触发了实际的网络请求。在某个时间点，你将第二个订阅者链接到了这个发布者。若是在链接第二个订阅者以前，网络请求已经完成，那么第二个订阅者将只会收到完成事件，收不到网络请求的响应结果。这时候，这个结果将不是你所指望。

在使用 Combine 的过程当中，咱们每每须要面对这些问题。如今就来弄清楚如何处理这一类问题吧~

 

使用 makeConnectable() 和 connect() 控制发布

 

ConnectablePublisher 是一个协议类型，它能够在你准备好以前阻止发布者发布元素。

/// 可链接的发布者，它提供了显式的链接、取消订阅的方式
///
/// 使用 `makeConnectable()` 来从任何一个失败类型是 `Never` 的发布者建立一个 `ConnectablePublisher`
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public protocol ConnectablePublisher : Publisher {

    /// 链接到发布者并返回一个用于取消发布的 `Cancellable` 实例
    ///
    /// - 返回值: 一个用于取消发布的 `Cancellable` 实例
    func connect() -> Cancellable
}

在你显式地调用 connect() 方法以前，一个 ConnectablePublisher 不会发送任何元素。

如今，就让咱们用 ConnectablePublisher 来解决上面提到的网络请求示例中的问题吧！

在两个订阅者都链接到发布者以后，调用 connect()，而后网络请求才被触发。这样就能够避免竞争(race condition)，保证两个订阅者都收到数据。

为了在你的 Combine 代码中使用 ConnectablePublisher，你能够使用 makeConnectable() 操做符将当前的发布者包装到一个 Publishers.MakeConnectable 结构体实例中。

以下方的代码所示：

class ConnectablePublisherDemo {
    
    private var cancellable1: AnyCancellable?
    private var cancellable2: AnyCancellable?
    private var connection: Cancellable?
    
    func run() {
        let url = URL(string: "https://ficow.cn")!
        let connectable = URLSession.shared
            .dataTaskPublisher(for: url)
            .map(\.data)
            .catch() { _ in Just(Data()) }
            .share()
            .makeConnectable() // 阻止发布者发布内容
        
        cancellable1 = connectable
            .sink(receiveCompletion: { print("Received completion 1: \($0).") },
                  receiveValue: { print("Received data 1: \($0.count) bytes.") })
        
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            self.cancellable2 = connectable.sink(receiveCompletion: { log("Received completion 2: \($0).") },
                                                 receiveValue: { log("Received data 2: \($0.count) bytes.") })
        }

        DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
			// 显式地启动发布。返回值须要被强引用，可用于取消发布（主动调用cancel方法或返回值被析构）
            self.connection = connectable.connect() 
        }
    }

}

请注意，在 makeConnectable() 操做符前面有一个 share() 操做符！请问，这个操做符有什么做用呢？

 

使用 autoconnect() 操做符进行自动链接

 

某些 Combine 发布者已经实现了 ConnectablePublisher 协议，如：Publishers.Multicast 和 Timer.TimerPublisher。使用这些发布者时，若是你不须要配置发布者或者不须要链接多个订阅者，你就须要显式地调用 connect() 方法。

对于这种状况，ConnectablePublisher 提供了 autoconnect() 操做符。当一个订阅者经过 subscribe(_:) 方法链接到发布者时，connect() 方法会被立刻调用。

let cancellable = Timer.publish(every: 1, on: .main, in: .default)
    .autoconnect()
    .sink() { date in
        print ("Date now: \(date)")
     }

上面的代码示例中使用了 autoconnect()，因此订阅者能够立刻接收到定时器发送的元素。若是没有 autoconnect()，咱们就须要在某个时刻手动地调用 connect() 方法。

 

总结

 

Combine 为咱们提供了很强大的异步编程功能，不过这也是有代价的，咱们须要深知使用 Combine 过程当中可能会遭遇的问题。若是不了解这些“坑”就开始上路，犯错的几率会很是高，犯错的成本也会很是高。

 

本文内容来源： Controlling Publishing with Connectable Publishers，转载请注明出处。