RxSwift的核心邏輯很重要，是分析整個(gè)框架的基礎(chǔ)磕蛇。
希望通俗易通景描，沒(méi)有弄思維導(dǎo)圖，個(gè)人覺(jué)得太多對(duì)象和線條的思維導(dǎo)圖秀撇，還沒(méi)看就眼花超棺。不多說(shuō)直接奔主題，以Observable開始呵燕，重點(diǎn)是核心邏輯棠绘，其他細(xì)節(jié)先忽略。例子：
let observable = Observable<Any>.create { (anyObserver) -> Disposable in
anyObserver.onNext("發(fā)送響應(yīng)")
anyObserver.onCompleted()
return Disposables.create()
}
observable.subscribe { (text) in
print("收到響應(yīng)")
}.disposed(by: disposbag)
- 首先是創(chuàng)建序列:
let observable = Observable<Any>.create { (observer) -> Disposable in
...
}
- 開始進(jìn)入源碼虏等，
Observable繼承了ObservableType協(xié)議（協(xié)議方法需要在別的擴(kuò)展里找到）：
public class Observable<Element> : ObservableType {
...
}
ObservableType實(shí)現(xiàn)了create弄唧，返回AnonymousObservable（保存了序列閉包subscribeHandler）：
extension ObservableType {
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)//內(nèi)部保存了閉包subscribe
}
}
final private class AnonymousObservable<Element>: Producer<Element> {
...
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
}
- 然后回到外面進(jìn)行訂閱，也就是
AnonymousObservable.subscribe：
observable.subscribe { (event) in
...
}
-
AnonymousObservable沒(méi)有具體的subscribe方法霍衫，這時(shí)候找父類候引，然而這里面是層層套，最后能看到它實(shí)現(xiàn)了ObservableType協(xié)議：
final private class AnonymousObservable<Element>: Producer<Element> {
...
}
class Producer<Element> : Observable<Element> {
...
}
public class Observable<Element> : ObservableType {
...
}
具體協(xié)議方法里敦跌，創(chuàng)建了匿名內(nèi)部觀察者AnonymousObserver（保存了事件閉包eventHandler）澄干，返回銷毀者：
extension ObservableType {
...
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
...
let observer = AnonymousObserver<E> { event in
...
}
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
}
...
}
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
...
init(_ eventHandler: @escaping EventHandler) {
...
self._eventHandler = eventHandler
}
...
}
- 在返回銷毀者時(shí)調(diào)用
self.asObservable().subscribe，首先是self.asObservable()
public class Observable<Element> : ObservableType {
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod() //具體方法由子類實(shí)現(xiàn)
}
public func asObservable() -> Observable<E> {
return self //子類可能會(huì)實(shí)現(xiàn)具體方法柠傍，最后結(jié)果都是返回一個(gè)序列Observable
}
}
然后self.asObservable().subscribe =>
self.subscribe =>
AnonymousObservable.subscribe =>
Producer.subscribe
class Producer<Element> : Observable<Element> {
...
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
}
}
}
...
}
-
Producer最終會(huì)調(diào)用self.run,而子類AnonymousObservable實(shí)現(xiàn)了具體的run方法：
final private class AnonymousObservable<Element>: Producer<Element> {
...
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
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)
}
}
- 然后
AnonymousObservable的run里面創(chuàng)建了AnonymousObservableSink(業(yè)務(wù)下沉)麸俘，保存了這里的observer(即AnonymousObserver)和cancel，AnonymousObservableSink擁有所有的功能——觀察惧笛，發(fā)送从媚，響應(yīng)，銷毀患整，相當(dāng)于manager的作用拜效，然后調(diào)用sink.run
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
...
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
...
}
class Sink<O : ObserverType> : Disposable {
...
init(observer: O, cancel: Cancelable) {
...
self._observer = observer
self._cancel = cancel
}
...
}
-
sink.run內(nèi)部的parent便是傳進(jìn)來(lái)的AnonymousObservable喷众，然后創(chuàng)建AnyObserver（保存AnonymousObservableSink.on函數(shù)），并利用AnonymousObservable保存的序列閉包subscribeHandler發(fā)送到外面：
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
...
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
}
public struct AnyObserver<Element> : ObserverType {
...
public init<O : ObserverType>(_ observer: O) where O.E == Element {
self.observer = observer.on
}
...
}
- 外面的
_subscribeHandler就會(huì)調(diào)用anyObserver.onNext("")紧憾，而anyObserver就是AnyObserver：
let observable = Observable<Any>.create { (anyObserver) -> Disposable in
anyObserver.onNext("發(fā)送信號(hào)")
anyObserver.onCompleted() //發(fā)送完成
return Disposables.create() //銷毀者
}
-
AnyObserver繼承了ObserverType協(xié)議：
public struct AnyObserver<Element> : ObserverType {
...
}
extension ObserverType {
public func onNext(_ element: E) {
self.on(.next(element))
}
...
}
-
AnyObserver又具體實(shí)現(xiàn)了self.on：
public struct AnyObserver<Element> : ObserverType {
...
public func on(_ event: Event<Element>) {
return self.observer(event)
}
...
}
- 然后調(diào)用
self.observer到千，便是AnyObserver保存的AnonymousObservableSink.on：
inal private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
...
func on(_ event: Event<E>) {
...
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()
}
}
}
...
}
- 然后根據(jù)
event進(jìn)入.next，調(diào)用self.forwardOn赴穗，AnonymousObservableSink沒(méi)有具體的forwardOn憔四，找父類Sink：
class Sink<O : ObserverType> : Disposable {
...
final func forwardOn(_ event: Event<O.E>) {
...
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
...
}
-
self._observer便是AnonymousObservableSink保存的AnonymousObserver，AnonymousObserver沒(méi)有具體的on方法般眉，找父類ObserverBase：
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
...
}
class ObserverBase<ElementType> : Disposable, ObserverType {
...
func on(_ event: Event<E>) {
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：
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
...
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
...
}
-
self._eventHandler便是AnonymousObserver保存的事件閉包eventHandler：
extension ObservableType {
...
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
...
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
)
}
...
}
- 最后根據(jù)
event調(diào)用外面?zhèn)鬟M(jìn)來(lái)的onNext?(value)了赵，便回到最初外面的響應(yīng)閉包：
observable.subscribe { (text) in
print("收到響應(yīng)")
}.disposed(by: disposbag)
這樣才能基本完成一個(gè)訂閱響應(yīng)的流程，還是不懂的可以自己斷點(diǎn)慢慢走一遍煤篙。
