注:代码块中的...是一些不需要显示的代码,直接用...替代了,rx的思维导图我后面重新画了一张,老漂亮了,画出经验了。
一.创建序列:
let ob = Observable<Any>.create { (obserber) -> Disposable in
obserber.onNext("onNext1")
return Disposables.create()
}
Observable
遵守了ObservableType
协议,create
方法是协议的一个扩展方法。
create
的具体实现可以在工程目录下搜索create.swift
查看。
create
方法返回了一个AnonymousObservable(subscribe)
类,AnonymousObservable
初始化保存了闭包。保存在了self._subscribeHandler
属性中,self
就是AnonymousObservable
,也就是 AnonymousObservable. _subscribeHandler,这个很重要,后面会用到(函数式编程思想)。
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
AnonymousObservable
,继承于父类Producer
,父类Producer
继承于Observable
,重写了subscribe
方法。
下面是父类Observable
:
public class Observable<Element> : ObservableType {
/// Type of elements in sequence.
public typealias E = Element
init() {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
}
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod()
}
public func asObservable() -> Observable<E> {
return self
}
deinit {
#if TRACE_RESOURCES
_ = Resources.decrementTotal()
#endif
}
解释:
func asObservable()
,返回一个self
,实现了非序列向序列的转化,这个是万物皆序列的第二种解释,第一种解释前面的博客介绍过了。例如UISwitch().rx.value
并不是一个序列,我们可以通过UISwitch().rx.value.asObservable()
,转成一个序列使用。
func subscribe
实现的是一个抽象方法,里面没有具体的实现。
这里画了一张序列继承链的思维导图,供大家参考,如有不对的地方望不吝赐教:
序列继承链.png
二.序列执行订阅操作
let _ = ob.subscribe(onNext: { (text) in
print("订阅到:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
ob.subscribe
方法中就下面的这段代码需要我们去分析。
let observer = AnonymousObserver<E> { 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
)
首先通过let observer = AnonymousObserver<E>{...}
创建了一个观察者。AnonymousObserver<E>
中的E
来自于public func subscribe(onNext: ((E) -> Void)? = nil, ....)
中的E
,而这个E
(序列类型),要从序列的subscribe
方法中寻找,序列的subscribe
方法是来自于:class AnonymousObservable<Element>
---->父类Producer
----> Observable<Element>
public class Observable<Element> : ObservableType {
/// Type of elements in sequence.
public typealias E = Element
...
}
接着说let observer = AnonymousObserver<E>
的AnonymousObserver
类,let observer = AnonymousObserver<E>{...}
这个其实是进行了一个初始化的操作, self._eventHandler = eventHandler
保存了尾随闭包:
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
typealias Element = ElementType
typealias EventHandler = (Event<Element>) -> Void
private let _eventHandler : EventHandler
init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self._eventHandler = eventHandler
}
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
...
}
这里的AnonymousObserver
继承的ObserverBase
,其实是一个基类订阅者,包含核心方法on
, onCore
发送消息,和dispose
方法执行销毁操作。
这里我们能够看到,如果AnonymousObserver<E>
后面的闭包能够被调用,就能实现序列执行订阅操作中尾随闭包的调动,也就完成了序列的响应,也就是说后面我们要搞清楚序列和订阅是如何关联上的。
而observer
的调用是在:
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
self.asObservable().subscribe(observer)
中 self.asObservable()
代表的就是序列,也就变成 ob.subscribe(observer)
(ob是刚开始创建的序列), ob
的subscribe
,需要到其父类(AnonymousObservable
)的父类(Producer
)中寻找。
class Producer<Element> : Observable<Element> {
override init() {
super.init()
}
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == 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
}
}
}
...
}
可以看到其实就是执行了一个 let sinkAndSubscription = self.run(observer, cancel: disposer)
的run
方法,同时返回了一个销毁者,run
的调用者不是AnonymousObservable
就是Producer
,所以查找run
方法的实现,如下。
以下是run方法的实现:
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<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
我们看到run
方法中创建了一个let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
,保存了observer
(订阅者)和cancel
(销毁者)。然后调用了sink.run(self)
,这里的self
就是对应的类AnonymousObservable
,也就是sink.run(AnonymousObservable)
,也就是sink.run
传入了一个序列,这一点非常重要,说明在sink
中获得了序列,后面sink
会把序列和订阅者关联。这个run
方法的执行如下:
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
parent._subscribeHandler(AnyObserver(self))
其实就是 AnonymousObservable._subscribeHandler(AnyObserver(self))
,这个方法是不是很熟悉,是的,就是create
初始化保存的代码块。这里我把run
方法的整个代码块贴出来,因为非常重要:
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
typealias E = O.E
typealias Parent = AnonymousObservable<E>
// state
private let _isStopped = AtomicInt(0)
#if DEBUG
fileprivate let _synchronizationTracker = SynchronizationTracker()
#endif
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
func on(_ event: Event<E>) {
#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 {
return parent._subscribeHandler(AnyObserver(self))
}
}
AnonymousObservable._subscribeHandler(AnyObserver(self))
,这个方法是怎么实现订阅者和序列的关联,主要是通过AnyObserver(self)
,也就是 AnyObserver(AnonymousObservableSink)
,在AnyObserver
这个结构体中,自定义属性self.observer = observer.on
(注意啦,这里的self.observer
是一个结构体的一个function
),保存了 observer
的on
方法的代码块,observer
是sink
传过来的,也就是
on
方法(AnonymousObservableSink类内的方法)
的内部就是如下(核心逻辑):
func on(_ event: Event<E>) {
#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()
}
}
}
上面的意思总结起来就是 结构体AnyObserver的一个函数AnyObserver. observer保存了一个on函数块
回归到 AnonymousObservable._subscribeHandler(AnyObserver(self))
,外界的obserber.onNext("onNext--1")
,其实就是是调用的AnyObserver.onNext
.而onNext
其实就是调用了一个on
方法(AnyObserver
遵从了ObserverType
协议,这个onNext
是协议的扩展方法)
extension ObserverType {
public func onNext(_ element: E) {
self.on(.next(element))
}
public func onCompleted() {
self.on(.completed)
}
public func onError(_ error: Swift.Error) {
self.on(.error(error))
}
}
这个on
方法,其实就是前面提到的AnonymousObservableSink
中的on
方法,再上一次代码块:
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
...
func on(_ event: Event<E>) {
#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()
}
}
}
...
}
走到这里,在执行switch
,---> 里面的self.forwardOn(event)
,调用的而其实是父类的forwardOn
方法:
class Sink<O : ObserverType> : Disposable {
...
final func forwardOn(_ event: Event<O.E>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
...
}
最后通过self._observer.on(event)
调用 我们之前创建订阅者时的闭包(let observer = AnonymousObserver<E> {闭包} )
:
let observer = AnonymousObserver<E> { 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()
}
}
然后就可以快乐的调用 onNext
,onError
,onCompleted
输出print(订阅到....)
了,终于完了,好累。以上就是RXSwift底层实现的核心逻辑了。如有不对的地方,烦请不吝赐教,本人不胜感激~~~
网友评论