RxSwift 事件
Event
next— 序列产生了一个新的元素error— 创建序列时产生了一个错误，导致序列终止completed— 序列的所有元素都已经成功产生，整个序列已经完成
-
序列的创建
//1、创建序列
let ob = Observable<Any>.create { (anyObserver) -> Disposable in
//3、发送信号
anyObserver.onNext("RxSwift")
anyObserver.onError(NSError(domain: "com.zm", code: 10087, userInfo: nil))
anyObserver.onCompleted()
return Disposables.create()
}
public class Observable<Element> : ObservableType{
//省略类的具体定义，我们主要看类的继承关系和遵守的协议
//类中的具体内容，可以在探索的过程中跳过去查看
}
-
Observable遵守协议ObservableType； -
Observable没有实现create，而是调用的是ObservableType协议可选方法create； -
subscribe闭包实现，闭包参数为AnyObserver类型的对象anyObserver，这里没有给这个对象命名为observer是为了与后面流程中以示区别。
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
return AnonymousObservable(subscribe)
}
-
subscribe闭包传值参数类型AnyObserver<Element>； - 返回私有的
AnonymousObservable匿名可观察序列，这时ob实际是一个AnonymousObservable类型的序列了； - 将
subscribe闭包保存在AnonymousObservable的属性_subscribeHandler中，完成序列的创建。 -
AnonymousObservable继承Producer.
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
-
发送订阅
//2、发送订阅
let _ = ob.subscribe(onNext: { (text) in
print("\(text)")
}, onError: { (error) in
print("\(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
-
subscribe发送订阅，注意这里的subscribe是方法，创建序列的时候的subscribe是闭包；
//发送订阅的方法
public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
//省略了销毁者相关代码，可以探索时在源码中查看
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()
}
}
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
}
- 创建了一个
AnonymousObserver匿名观察者对象observer； -
AnonymousObserver继承ObserverBase，ObserverBase遵守ObserverType协议； -
asOservable()确保是一个序列； -
subscribe在AnonymousObservable中没有实现，调用的是父类Producer中的subscribe方法，并将observer传入其中。
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
if !CurrentThreadScheduler.isScheduleRequired {
//省略部分与分析流程无关代码
}
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
}
}
}
-
demo不涉及开线程情况，走当前线程 - 调用
AnonymousObservable的run方法，传入observer
//AnonymousObservable中的run方法
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
//调用AnonymousObservableSink中run方法
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
- 此时又引入了一个新的类
AnonymousObservableSink，起到关联观察者Observer与可观察序列Observable的作用； -
AnonymousObservableSink继承Sink类，并遵守了ObserverType协议； - 创建
AnonymousObservableSink的对象sink，在初始化时又将run方法传递过来的observer，cancel交由父类Sink保存在_observer、_cancel中; - 调用
sink的run方法将AnonymousObservable自己传入；
typealias Parent = AnonymousObservable<Element>
//AnonymousObservableSink中的run方法
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self)
}
- 从源码中可以看出
Parent是AnonymousObservable的别名，parent是一个AnonymousObservable对象； - 在run方法的返回值中，执行了
parent的_subscribeHandler闭包回调，并将AnonymousObservableSink对象本身作为AnyObserver初始化方法的参数；
public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
self.observer = observer.on
}
- 在
AnyObserver的初始化方法中，将AnonymousObservableSink的on方法保存在了属性observer，注意这里有两个observer，它们不是一样的，跟前文的observer也不一样；
发送消息
//截取Demo中消息发送片段的代码
observer.onNext("RxSwift")
observer.onError(NSError(domain: "com.zm", code: 10087, userInfo: nil))
observer.onCompleted()
func on(_ event: Event<Element>)
public func onNext(_ element: Element) {
self.on(.next(element))
}
public func onCompleted() {
self.on(.completed)
}
public func onError(_ error: Swift.Error) {
self.on(.error(error))
}
-
onNext、onCompleted、onError是ObserverType协议协议拓展方法； -
ObserverType中定义了on方法，没有具体实现，交有子类AnyObserver实现；
public func on(_ event: Event<Element>) {
return self.observer(event)
}
- 执行
observer闭包的回调，前文说过self.observe保存的是AnonymousObservableSink中的on方法，那么接下来的流程肯定会走到on方法的具体实现
func on(_ event: Event<Element>) {
//省略部分代码
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()
}
}
}
-
on方法中调用了forwardOn方法，forwardOn是在AnonymousObservableSink的父类Sink中实现的
final func forwardOn(_ event: Event<Observer.Element>) {
//省略部分无关代码
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
-
self._observer序列的订阅中已经说过是一个AnonymousObserver对象，on方法在AnonymousObserver没有实现，在它的父类ObserverBase中实现的；
func on(_ event: Event<Element>) {
switch event {
case .next:
if load(self._isStopped) == 0 {
self.onCore(event)
}
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.onCore(event)
}
}
}
- 调用
AnonymousObserver的onCore方法，执行_eventHandler闭包的回调
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
- 此时终于回到了
subscribe订阅方法中的创建的observer对象是的闭包实现中了：
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()
}
}
- 在
event事件中执行subscribe方法中闭包参数onNext、onError、onCompleted的回调，完成元素的传递。
总结：
- 序列的创建过程是将一个
观察者对象保存到一个subscribe闭包的过程，在闭包的实现中onNext产生元素，元素完成时onComplete，产生元素时出错的处理onError；- 序列的订阅
-
最后附一张整个流程图
网友评论