Share Forever
每次订阅公用同一个connection, 没有订阅者，自动断开connection
1
2
3
4
5
6
7
8
9
10
11
12
13
extension ConnectableObservableType {
/**
Returns an observable sequence that stays connected to the source as long as there is at least one subscription to the observable sequence.
- seealso: [refCount operator on reactivex.io](http://reactivex.io/documentation/operators/refcount.html)
- returns: An observable sequence that stays connected to the source as long as there is at least one subscription to the observable sequence.
*/
public func refCount() -> Observable<Element> {
return RefCount(source: self)
}
}
final private class RefCount<ConnectableSource: ConnectableObservableType>: Producer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
fileprivate let _lock = RecursiveLock()
// state
fileprivate var _count = 0
fileprivate var _connectionId: Int64 = 0
fileprivate var _connectableSubscription = nil as Disposable?
fileprivate let _source: ConnectableSource
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable)
where Observer.Element == ConnectableSource.Element {
let sink = RefCountSink(parent: self, observer: observer, cancel: cancel)
let subscription = sink.run()
return (sink: sink, subscription: subscription)
}
final private class RefCountSink<ConnectableSource: ConnectableObservableType, Observer: ObserverType>: Sink, ObserverType where ConnectableSource.Element == Observer.Element
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
typealias Element = Observer.Element
typealias Parent = RefCount<ConnectableSource>
private let _parent: Parent
private var _connectionIdSnapshot: Int64 = -1
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.disposed {
return Disposables.create()
}
if self._parent._count == 0 {
self._parent._count = 1
self._parent._connectableSubscription = self._parent._source.connect()
}
else {
self._parent._count += 1
}
// }
return Disposables.create {
subscription.dispose()
self._parent._lock.lock(); defer { self._parent._lock.unlock() } // {
if self._parent._connectionId != self._connectionIdSnapshot {
return
}
if self._parent._count == 1 {
// 没有订阅者，自动结束connection
self._parent._count = 0
guard let connectableSubscription = self._parent._connectableSubscription else {
return
}
connectableSubscription.dispose()
self._parent._connectableSubscription = nil
}
else if self._parent._count > 1 {
self._parent._count -= 1
}
else {
rxFatalError("Something went wrong with RefCount disposing mechanism")
}
// }
}
}
func on(_ event: Event<Element>) {
switch event {
case .next:
self.forwardOn(event)
case .error, .completed:
// 只要收到（或者说只要source发出stop events, 就结束connection）
self._parent._lock.lock() // {
if self._parent._connectionId == self._connectionIdSnapshot {
let connection = self._parent._connectableSubscription
defer { connection?.dispose() }
self._parent._count = 0
self._parent._connectionId = self._parent._connectionId &+ 1
self._parent._connectableSubscription = nil
}
// }
self._parent._lock.unlock()
self.forwardOn(event)
self.dispose()
}
}