TakeCountSink
注意TakeCountSink是Disposable同时也是Observer
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
final private class TakeCountSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
typealias Element = Observer.Element
typealias Parent = TakeCount<Element>
private let _parent: Parent
private var _remaining: Int
init(parent: Parent, observer: Observer, cancel: Cancelable) {
self._parent = parent
self._remaining = parent._count
super.init(observer: observer, cancel: cancel)
}
func on(_ event: Event<Element>) {
switch event {
case .next(let value):
if self._remaining > 0 {
self._remaining -= 1
self.forwardOn(.next(value))
if self._remaining == 0 {
self.forwardOn(.completed)
self.dispose()
}
}
case .error:
self.forwardOn(event)
self.dispose()
case .completed:
self.forwardOn(event)
self.dispose()
}
}
}
TakeCount
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
final private class TakeCount<Element>: Producer<Element> {
private let _source: Observable<Element>
fileprivate let _count: Int
init(source: Observable<Element>, count: Int) {
if count < 0 {
rxFatalError("count can't be negative")
}
self._source = source
self._count = count
}
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
let sink = TakeCountSink(parent: self, observer: observer, cancel: cancel)
// 这里subscribe sink, 是sink作为observer的运用
let subscription = self._source.subscribe(sink)
return (sink: sink, subscription: subscription)
}
}
注：文中源码来自RxSwift