create的闭包什么时候执行?
-
create
源码中可以看出,执行create
方法返回了一个 AnonymousObservable
对象, 整个subscribe
闭包都传递该对象,继续往里面跟代码
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
AnonymousObservable(subscribe)
}
-
AnonymousObservable
继承自 Producer
, 同时重写了父类中的run
方法
-
run
方法里面有一个AnonymousObservableSink
对象来处理,所以要弄清楚create
中的闭包在什么执行,就要弄清楚run方法是在什么执行的
// 匿名可观察序列
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self.subscribeHandler = subscribeHandler
}
// 重写了父类的run方法
// 弄清楚run方法在什么时候执行
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
/// 匿名可观察槽(Sink: 水槽)
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
/// 订阅是由sink执行run方法产生的
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
- 继续跟进
AnonymousObservableSink
,发现里面中的run
方法直接就执行了这个subscribeHandler
, 同时传入了AnyObserver对象
final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
typealias Element = Observer.Element
typealias Parent = AnonymousObservable<Element>
// state
private let isStopped = AtomicInt(0)
override init(observer: Observer, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
func on(_ event: Event<Element>) {
#if DEBUG
self.synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self.synchronizationTracker.unregister() }
#endif
switch event {
case .next:
if load(self.isStopped) == 1 {
return
}
self.forwardOn(event)
case .error, .completed:
if fetchOr(self.isStopped, 1) == 0 {
self.forwardOn(event)
self.dispose()
}
}
}
func run(_ parent: Parent) -> Disposable {
/// 执行AnonymousObservable的subscribeHandler闭包
/// 传入的参数是一个AnyObserver,AnyObserver是AnonymousObservableSink的包装
parent.subscribeHandler(AnyObserver(self))
}
}
- 接下来就去查找
AnonymousObservable
中run
方法是在哪里执行的,去父类Producer
中阅读发现, run
是在subscribe
方法执行的,并且规定run
方法必须被子类重写(抽象方法)
class Producer<Element>: Observable<Element> {
override init() {
super.init()
}
/// Observable执行subscribe时调用了run方法
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
if !CurrentThreadScheduler.isScheduleRequired {
// The returned disposable needs to release all references once it was disposed.
let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
else {
return CurrentThreadScheduler.instance.schedule(()) { _ in
let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
}
}
// run方法必须被子类重写
func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
rxAbstractMethod()
}
}
- 既然知道了run方法是在subscribe中的执行的,那么我们就去找subscribe是在哪里被调用的
- 找一圈发现,subscribe方法是在
subscribe(onNext:
方法中被执行的, subscribe(onNext:
方法是被外部主动调用的, 改方法会创建一个AnonymousObserver
对象, 同时将这个对象传递 subscribe(observer)
public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
let disposable: Disposable
if let disposed = onDisposed {
disposable = Disposables.create(with: disposed)
}
else {
disposable = Disposables.create()
}
#if DEBUG
let synchronizationTracker = SynchronizationTracker()
#endif
let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
let observer = AnonymousObserver<Element> { event in
#if DEBUG
synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { synchronizationTracker.unregister() }
#endif
switch event {
case .next(let value):
onNext?(value)
case .error(let error):
if let onError = onError {
onError(error)
}
else {
Hooks.defaultErrorHandler(callStack, error)
}
disposable.dispose()
case .completed:
onCompleted?()
disposable.dispose()
}
}
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
}
- 所以综上,create的闭包的闭包是在外部调用
subscribe(onNext,
其内部执行subscribe(observer)
,,执行其子类AnonymousObservable
中的run
方法,AnonymousObservable
的run
方法通过AnonymousObservableSink
对象执行run方法,最终执行了subscribeHandler
闭包
subscribe(onNext 中的next闭包什么时候执行
- 当执行
subscribe(onNext
方法时,内部创建了一个AnonymousObserver
对象给 subscribe
方法
public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
let disposable: Disposable
if let disposed = onDisposed {
disposable = Disposables.create(with: disposed)
}
else {
disposable = Disposables.create()
}
let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
// 创建匿名观察者
let observer = AnonymousObserver<Element> { event in
switch event {
case .next(let value):
onNext?(value)
case .error(let error):
if let onError = onError {
onError(error)
}
else {
Hooks.defaultErrorHandler(callStack, error)
}
disposable.dispose()
case .completed:
onCompleted?()
disposable.dispose()
}
}
// 执行 subscribe 方法,传入创建的匿名观察者
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
}
- 在create方法的闭包执行时(_subscribeHandler),会将AnonymousObservableSink包装成一个AnyObserver对象, AnonymousObservableSink对象中Observer就是AnonymousObserver
final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
typealias Element = Observer.Element
typealias Parent = AnonymousObservable<Element>
// state
private let _isStopped = AtomicInt(0)
#if DEBUG
private let _synchronizationTracker = SynchronizationTracker()
#endif
override init(observer: Observer, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
func on(_ event: Event<Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
switch event {
case .next:
if load(self._isStopped) == 1 {
return
}
// 执行父类的forwardOn方法
self.forwardOn(event)
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.forwardOn(event)
self.dispose()
}
}
}
func run(_ parent: Parent) -> Disposable {
// crate方法中的闭包参数是AnyObserver对象
// 执行observer.on(.next(1)) 等价于调用了AnyObserver中的next方法
return parent._subscribeHandler(AnyObserver(self))
}
}
- 执行observer.on(.event(1))方法,相当于是调用了AnyObserver对象中的on方法
public struct AnyObserver<Element> : ObserverType {
/// Anonymous event handler type.
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
public init(eventHandler: @escaping EventHandler) { //
self.observer = eventHandler
}
// observer == AnonymousObservableSink
public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
// observer闭包指向AnonymousObservableSink中的on方法
self.observer = observer.on
}
/// - parameter event: Event instance.
public func on(_ event: Event<Element>) {
// 执行EventHandler闭包, 意味着执行的是AnonymousObservableSink中的on方法
return self.observer(event)
}
}
- 在AnyObserver创建时,传入Observer是AnonymousObservableSink对象,同时将AnonymousObservableSink对象的on方法赋值给AnyObserver对象的observer闭包,执行EventHandler闭包, 意味着执行的是AnonymousObservableSink中的on方法
// AnonymousObservableSink
func on(_ event: Event<Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
switch event {
case .next:
if load(self._isStopped) == 1 {
return
}
// 执行父类的forwardOn方法
self.forwardOn(event)
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.forwardOn(event)
self.dispose()
}
}
}
- 执行的是AnonymousObservableSink中的on方法, 执行父类的forwardOn方法
// Sink
final func forwardOn(_ event: Event<Observer.Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
if isFlagSet(self._disposed, 1) {
return
}
// 执行传入的Observer的on方法, 即AnonymousObserver中的on方法
self._observer.on(event)
}
- 执行传入的Observer的on方法, 即AnonymousObserver中的on方法,
- AnonymousObserver中的on方法,在父类ObserverBase中,on方法的内部则执行的是onCore方法,该方法则在子类AnonymousObserver中实现
- onCore方法则执行了其创建时的EventHandler闭包
// ObserverBase
func on(_ event: Event<Element>) {
switch event {
case .next:
if load(self._isStopped) == 0 {
// 执行onCore方法,onCore方法必须被子类重写
self.onCore(event)
}
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.onCore(event)
}
}
}
// AnonymousObserver
// oncore方法执行的是传入的EventHandler闭包,EventHandler闭包根据枚举类型执行对应的代码
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
observer.on(.next(1))的执行流程