BehaviorSubject
- 线程安全
- With a initial value
- Recoding the latest value for replay it to observe on subscribe, so Observers can subscribe to the subject to receive the last (or initial) value and all subsequent notifications.
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
public final class BehaviorSubject<Element>
: Observable<Element>
, SubjectType
, ObserverType
, SynchronizedUnsubscribeType
, Cancelable {
public typealias SubjectObserverType = BehaviorSubject<Element>
typealias Observers = AnyObserver<Element>.s
typealias DisposeKey = Observers.KeyType
/// Indicates whether the subject has any observers
public var hasObservers: Bool {
self._lock.lock()
let value = self._observers.count > 0
self._lock.unlock()
return value
}
let _lock = RecursiveLock()
// state
private var _isDisposed = false
private var _element: Element
private var _observers = Observers()
private var _stoppedEvent: Event<Element>?
#if DEBUG
private let _synchronizationTracker = SynchronizationTracker()
#endif
/// Indicates whether the subject has been disposed.
public var isDisposed: Bool {
return self._isDisposed
}
/// Initializes a new instance of the subject that caches its last value and starts with the specified value.
///
/// - parameter value: Initial value sent to observers when no other value has been received by the subject yet.
public init(value: Element) {
self._element = value
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
}
/// Gets the current value or throws an error.
///
/// - returns: Latest value.
public func value() throws -> Element {
self._lock.lock(); defer { self._lock.unlock() } // {
if self._isDisposed {
throw RxError.disposed(object: self)
}
if let error = self._stoppedEvent?.error {
// intentionally throw exception
throw error
}
else {
return self._element
}
//}
}
/// Notifies all subscribed observers about next event.
///
/// - parameter event: Event to send to the observers.
public func on(_ event: Event<Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
dispatch(self._synchronized_on(event), event)
}
func _synchronized_on(_ event: Event<Element>) -> Observers {
self._lock.lock(); defer { self._lock.unlock() }
if self._stoppedEvent != nil || self._isDisposed {
return Observers()
}
switch event {
case .next(let element):
self._element = element
case .error, .completed:
self._stoppedEvent = event
}
return self._observers
}
/// Subscribes an observer to the subject.
///
/// - parameter observer: Observer to subscribe to the subject.
/// - returns: Disposable object that can be used to unsubscribe the observer from the subject.
public override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
self._lock.lock()
let subscription = self._synchronized_subscribe(observer)
self._lock.unlock()
return subscription
}
func _synchronized_subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
if self._isDisposed {
observer.on(.error(RxError.disposed(object: self)))
return Disposables.create()
}
if let stoppedEvent = self._stoppedEvent {
observer.on(stoppedEvent)
return Disposables.create()
}
let key = self._observers.insert(observer.on)
observer.on(.next(self._element))
return SubscriptionDisposable(owner: self, key: key)
}
func synchronizedUnsubscribe(_ disposeKey: DisposeKey) {
self._lock.lock()
self._synchronized_unsubscribe(disposeKey)
self._lock.unlock()
}
func _synchronized_unsubscribe(_ disposeKey: DisposeKey) {
if self._isDisposed {
return
}
_ = self._observers.removeKey(disposeKey)
}
/// Returns observer interface for subject.
public func asObserver() -> BehaviorSubject<Element> {
return self
}
/// Unsubscribe all observers and release resources.
public func dispose() {
self._lock.lock()
self._isDisposed = true
self._observers.removeAll()
self._stoppedEvent = nil
self._lock.unlock()
}
#if TRACE_RESOURCES
deinit {
_ = Resources.decrementTotal()
}
#endif
}
注：文中源码来自RxSwift