今天介绍可连接序列和连接相关操作符
可连接序列 Connection Observable, 不同于一般的序列，有订阅时不会立刻开始发送事件消息，只有当调用 connect()之后才会开始发送值。
connect & publish 操作符
我们看一个例子：
let observable = Observable<Int>.create({ anyObserver in
print("subscrition")
anyObserver.onNext(1)
anyObserver.onNext(2)
anyObserver.onNext(3)
return Disposables.create()
})
.publish()
observable
.subscribe {
print("订阅1", $0)
}
.disposed(by: bag)
DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
_ = observable.connect()
}
observable.subscribe {
print("订阅2", $0)
}
subscrition
订阅1 next(1)
订阅2 next(1)
订阅1 next(2)
订阅2 next(2)
订阅1 next(3)
订阅2 next(3)
姑且把创建序列传入的闭包称之为subscribe handler
3秒后才会打印信息，因为3秒后才连接。而且我们发现subscrition只打印了一次，说明subscribe handler只执行了一遍，如果去掉publish和connect，则会执行两遍，相当于每次订阅都会重新执行一遍；如果只去掉connect，则不会发送事件。
对于上面的结果我们分析一下源码：
对于publish操作符，不妨看看：
public func publish() -> ConnectableObservable<Element> {
self.multicast { PublishSubject() }
}
publish本质上调用了multicast操作符：
public func multicast<Subject: SubjectType>(makeSubject: @escaping () -> Subject)
-> ConnectableObservable<Subject.Element> where Subject.Observer.Element == Element {
ConnectableObservableAdapter(source: self.asObservable(), makeSubject: makeSubject)
}
返回一个ConnectableObservableAdapter对象，它也是个Observable，保存了原始的Observable和 makeSubject闭包。
- ConnectableObservableAdapter继承了ConnectableObservable，ConnectableObservable表示一个可连接的序列，ConnectableObservable遵循ConnectableObservableType协议声明一个connect()协议，表示连接操作。
- SubjectType表示这是一个Subject，在上一篇文章已经讲述过，那么连接操作实际上将一个普通的Observable连接到一个Subject，而publish操作明显是接到PublishSubject，它是没有buffer的Subject。
然后再看看subscribe代码在哪：
public func subscribe(_ on: @escaping (Event<Element>) -> Void) -> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
然后调用熟悉的subscribe(observer), 这里会进入ConnectableObservableAdapter的subscribe：
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Subject.Element {
self.lazySubject.subscribe(observer)
}
这里的lazySubject是通过makeSubject闭包创建的PublishSubject()；通过它继续调用subscribe，这里相当于做了一次变形，而我们在上一篇已经介绍过PublishSubject执行subscribe是不会发送事件消息的（先不考虑error和complete事件），它只是将观察者插入到observers中保存起来，这块的代码上一篇已经详细讲解过，这里不再讲解。所以就解释了为什么connect之前调用subscribe不会发送事件消息；那么connect又干了什么呢？我们继续看connect的代码：
// at `ConnectableObservableAdapter` class
override func connect() -> Disposable {
return self.lock.performLocked {
if let connection = self.connection {
return connection
}
let singleAssignmentDisposable = SingleAssignmentDisposable()
let connection = Connection(parent: self, subjectObserver: self.lazySubject.asObserver(), lock: self.lock, subscription: singleAssignmentDisposable)
self.connection = connection
let subscription = self.source.subscribe(connection)
singleAssignmentDisposable.setDisposable(subscription)
return connection
}
}
这里创建了一个Connection，并且执行了原始序列的subscribe(connection), 以connection作为参数，原始序列是最初的AnonymousObservable, AnonymousObservable的subscribe方法实现已经比较熟悉了，不用再看。它最终会执行最初创建序列的闭包。但是因为传入的observer是connection，而且Connection类重写了on方法，所以在调用onNext时，会调用到这个on方法：
func on(_ event: Event<Subject.Observer.Element>) {
if isFlagSet(self.disposed, 1) {
return
}
if event.isStopEvent {
self.dispose()
}
self.subjectObserver.on(event)
}
然后执行PulishSubject的on方法：这里会向所有观察者发送消息，例子中的观察者有两个。意思是例子每次调用onNext会通知所有的观察者。这也解释了为什么“subscrition”只打印了一遍，onNext每个打印两遍。如果先connect再subscribe，则只会打印“subscrition”，因为connect时还没有观察者。
share
share操作符可接受两个参数，replay 和 scope, replay表示缓存的个数或者叫重放的次数，这个好理解，而scope可接受两个枚举值.whileConnected 和 .forever，这两个什么区别：
-
.whileConnected只有在连接时才支持缓存和重放，意味着如果连接断开则不能重放，下次再订阅时只能重新执行subscribe handler，一般什么操作会断开连接？当发送error或complete事件或者回收资源时会断开连接。 -
.forever永远支持缓存和重放，无论是否断开连接。
我们看看官方给的例子：
func testScope(){
let xs = Observable.deferred { () -> Observable<TimeInterval> in
print("Performing work ...")
return Observable.just(Date().timeIntervalSince1970)
}
.share(replay: 1, scope: .forever)
_ = xs.subscribe(onNext: { print("next \($0)") }, onCompleted: { print("completed\n") })
_ = xs.subscribe(onNext: { print("next \($0)") }, onCompleted: { print("completed\n") })
_ = xs.subscribe(onNext: { print("next \($0)") }, onCompleted: { print("completed\n") })
}
这里的just方法会发送一个单一的元素：当前时间戳，还会在发一个complete事件，scope设置的是forever
打印结果：
Performing work ...
next 1622777422.7355828
completed
next 1622777422.7355828
completed
next 1622777422.7355828
completed
第一次订阅时，执行了subscribe handler，打印了Performing work ... next和complete，之后两次订阅时是对第一次的重放，打印的时间戳也一样。我们设置重放的个数为1，所以会重放一次next，同时也会重放complete；第一次执行complete其实以及断开连接了，但是依然能重放，说明就算连接断开了依然支持重放，这就是forever的意思。
如果改成.willConnected
则无法支持重放，打印结果：
Performing work ...
next 1622778439.6327991
completed
Performing work ...
next 1622778439.6332831
completed
Performing work ...
next 1622778439.633359
completed
这里相当于执行了三遍subscre handler，所以Performing work ...打印的三次，而且3次next打印的时间戳不一样；为什么呢，因为第一订阅之后就执行了complete事件了（just操作符会执行一次onNext+一次onComplete），这时候连接断开，再次订阅时需要重新建立连接，然又会重新执行subscribe handler。
我们再看看share的源码：
public func share(replay: Int = 0, scope: SubjectLifetimeScope = .whileConnected)
-> Observable<Element> {
switch scope {
case .forever:
switch replay {
case 0: return self.multicast(PublishSubject()).refCount()
default: return self.multicast(ReplaySubject.create(bufferSize: replay)).refCount()
}
case .whileConnected:
switch replay {
case 0: return ShareWhileConnected(source: self.asObservable())
case 1: return ShareReplay1WhileConnected(source: self.asObservable())
default: return self.multicast(makeSubject: { ReplaySubject.create(bufferSize: replay) }).refCount()
}
}
}
当scope == forever时，调用的multicast， replay == 0 连接的subject是PublishSubject，这个跟publish的操作符是一样的，不一样的是后面接了一个refCount操作符。
refCount返回一个RefCount对象，并将当前对象作为source
public func refCount() -> Observable<Element> {
RefCount(source: self)
}
RefCount是一个与源保持连接的可观察序列，只要该可观察序列至少有一次订阅。RefCount有一个对应的RefCountSink类，外面执行订阅时会调用RefCountSink的run方法：
func run() -> Disposable {
let subscription = self.parent.source.subscribe(self)
self.parent.lock.lock(); defer { self.parent.lock.unlock() }
self.connectionIdSnapshot = self.parent.connectionId
if self.isDisposed {
return Disposables.create()
}
if self.parent.count == 0 {
self.parent.count = 1
self.parent.connectableSubscription = self.parent.source.connect()
}
else {
self.parent.count += 1
}
/// 省略。。。
第一句执行self.parent.source.subscribe(self)，这里的parent是RefCount对象，而它的source是ConnectableObservableAdapter对象，它的subscribe代码：
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Subject.Element {
self.lazySubject.subscribe(observer)
}
lazySubject是要连接的Subject，之后调用subject的subscribe，这个subscribe会执行重放和插入观察者操作，回到RefCountSink的run：
后面有一个关键代码：
if self.isDisposed {
return Disposables.create()
}
因为第一订阅时，会执行complete，所以执行当前dispose，第二次订阅时self.isDisposed == true，直接返回了。所以第二次第三次订阅时不会执行subscribe handler，只做重放。
如果share操作改成share(replay:2, .whileConnected), 执行的代码是return self.multicast(makeSubject: { ReplaySubject.create(bufferSize: replay) }).refCount()， 这个跟 forever分支内的 return self.multicast(ReplaySubject.create(bufferSize: replay)).refCount()代码很像，为什么一个是forever一个是whileConnected。关键在于return self.multicast(ReplaySubject.create(bufferSize: replay)).refCount()这里只会创建一个ReplaySubject对象，而 { ReplaySubject.create(bufferSize: replay) }是一个闭包，每次在执行闭包时都会创建一个新的subject。所以每次订阅时都会创建一个新的subject。新的subject不存在重放的buffer；上面的代码self.isDisposed 一直是false，这导致每次需要再次连接。而每次连接都会执行subscribe handler。
还剩下下面两个case的代码没有解读：
case 0: return ShareWhileConnected(source: self.asObservable())
case 1: return ShareReplay1WhileConnected(source: self.asObservable())
ShareWhileConnected 处理无buffer的情况，ShareReplay1WhileConnected处理一个buffer的情况。这两个类非常相似，这两个分支并没有连接Subject，而是用ShareReplay1WhileConnected和ShareWhileConnected代替了Subject的功能。在ShareReplay1WhileConnected和ShareWhileConnected中，保存了所有的观察者；在ShareReplay1WhileConnected中保存了最后一个元素，这是为了实现重放。
它们都持有一个Connection，分别是ShareWhileConnectedConnection和ShareReplay1WhileConnectedConnection，这两个类实现了连接：
final func connect() {
self.subscription.setDisposable(self.parent.source.subscribe(self))
}
这里会去调用原始序列的subscribe，最终调用subscribe handler。
同时实现了订阅和通知观察者的功能，代码类似于Subject。
fileprivate var observers = Observers()
private var element: Element?
那这两个类是怎么实现whileConnected效果的？ 看看ShareReplay1WhileConnected的subscribe的实现：
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
self.lock.lock()
let connection = self.synchronized_subscribe(observer)
let count = connection.observers.count
let disposable = connection.synchronized_subscribe(observer)
self.lock.unlock()
if count == 0 {
connection.connect()
}
return disposable
}
@inline(__always)
private func synchronized_subscribe<Observer: ObserverType>(_ observer: Observer) -> Connection where Observer.Element == Element {
let connection: Connection
if let existingConnection = self.connection {
connection = existingConnection
}
else {
connection = ShareReplay1WhileConnectedConnection<Element>(
parent: self,
lock: self.lock)
self.connection = connection
}
return connection
}
当count == 0 执行连接，如果发送complete事件断开连接后，此时再次订阅，则count依然是==0 需要重新连接，这样就会再执行一遍subscribe handler，如果没有发送complete时间，则只会连接一次，后面的订阅只会重放buffer。
replay
replay操作符实际上是通过self.multicast { ReplaySubject.create(bufferSize: bufferSize) }实现，效果类似于publish，只不过publish不带buffer，replay可以指定buffer个数。
总结
普通可观察序列都可以通过以上操作符转变成可连接的序列，可连接序列只有在连接之后才能发送事件元素，而whileConnected和forever区别在于：
- whileConnected：每个连接将有它自己Subject实例来存储重放事件。
- forever：一个Suject将存储到source的所有连接的重放事件。
