简单分析RxSwift源码，这段代码到底做了什么？
let disposeBagTwo = DisposeBag()
Observable.just("🔴")
.subscribe { event in
print(event)
}
.disposed(by: disposeBagTwo)
RxSwift中有这些基础的东西：Observable、Observer、序列（好抽象，这个是Rx重点，本文只带过），那在上面的代码中到底对应的是什么？它们是怎么联系在一起的呢？先来看看它们的抽象类(进行了简化)
这是Observable所需要的。注意我把Just的父类跳过Producer修改为了Observable，这是为了主干分析时不被细枝影响，也可不修改，对于上面的代码块没有任何影响
public protocol ObservableConvertibleType {
associatedtype E
func asObservable() -> Observable<E>
}
public protocol ObservableType : ObservableConvertibleType {
func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E
}
public class Observable<Element> : ObservableType {
/// Type of elements in sequence.
public typealias E = Element
init() {
}
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod()
}
public func asObservable() -> Observable<E> {
return self
}
deinit {
}
}
extension ObservableType {
/**
Returns an observable sequence that contains a single element.
- seealso: [just operator on reactivex.io](http://reactivex.io/documentation/operators/just.html)
- parameter element: Single element in the resulting observable sequence.
- returns: An observable sequence containing the single specified element.
*/
public static func just(_ element: E) -> Observable<E> {
return Just(element: element)
}
}
final fileprivate class Just<Element> : Observable <Element> {
private let _element: Element
init(element: Element) {
_element = element
}
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
observer.on(.next(_element))
observer.on(.completed)
return Disposables.create()
}
}
下面是Observer所需要的。注意这些修改是需要的（为了后面修改的代码能正常运行），AnonymousObserver修改为公共的类，需要在4处添加public标识（在下面源码中已经用`***`标记）。后面的修改是不必需的，移除ObserverBase遵循Disposable协议，还是为了方便分析，如果你也一样在源码进行了修改，那会出现两处错误，处理方式是修改报错处的返回代码为return (sink: Disposables.create(), subscription: subscription)
public protocol ObserverType {
associatedtype E
func on(_ event: Event<E>)
}
//移除了Disposable协议
class ObserverBase<ElementType> : ObserverType {
typealias E = ElementType
private var _isStopped: AtomicInt = 0
func on(_ event: Event<E>) {
switch event {
case .next:
if _isStopped == 0 {
onCore(event)
}
case .error, .completed:
if AtomicCompareAndSwap(0, 1, &_isStopped) {
onCore(event)
}
}
}
func onCore(_ event: Event<E>) {
rxAbstractMethod()
}
}
//*** 添加public
public final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
//*** 添加public
public typealias Element = ElementType
//*** 添加public
public typealias EventHandler = (Event<Element>) -> Void
private let _eventHandler : EventHandler
//*** 添加public
public init(_ eventHandler: @escaping EventHandler) {
_eventHandler = eventHandler
}
override func onCore(_ event: Event<Element>) {
return _eventHandler(event)
}
}
这些是辅助代码块，可能它就解开了所有的联系😄
extension ObservableType {
/**
Subscribes an event handler to an observable sequence.
- parameter on: Action to invoke for each event in the observable sequence.
- returns: Subscription object used to unsubscribe from the observable sequence.
*/
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
}
现在我们就来改写最开始的代码，让所有的对象显示出来：
第一阶段
//通过 ObservableType 协议扩展的 just 方法创建 Just 对象->observable
let observable = Observable.just("🔴")
//定义一个闭包,这个闭包的内容和开始代码段subscribe后的一致
let onCall : ((RxSwift.Event<String>) -> Swift.Void) = { event in
print(event)
}
/*改写 .subscribe { event in
print(event)
}
这里为尾闭包，我们还原回去
注意：这里返回的是Disposable对象，我们的Observer去哪了？下面会进一步分解*/
let disposable = observable.subscribe(onCall)
//最后一步都很熟悉
disposable.disposed(by: disposeBagTwo)
我们那找说好的Observer去哪里。从observable.subscribe(onCall)分析
第二阶段
let disposable = observable.subscribe(onCall)
//注意这里的`subscribe`方法是通过`ObservableType`协议扩展的方法（前面提到的辅助代码块😄），
//注意与`Observable`类中`subscribe`方法进行区别(多态性)。
//分解为
let observer = AnonymousObserver(onCall)
let disposable = observable.asObservable().subscribe(observer)
//注意在此例中observable就是Observable类对象，asObservable返回的也是self，因此还可以简化为
let disposable = observable.subscribe(observer)
Final版本出现了
let disposeBagTwo = DisposeBag()
//Observable
let observable = Observable.just("🔴")
//事件处理
let onCall : ((RxSwift.Event<String>) -> Swift.Void) = { event in
print(event)
}
//Observer
let observer = AnonymousObserver(onCall)
//Disposable
let disposable = observable.subscribe(observer)
disposable.disposed(by: disposeBagTwo)
终结：
所有的关系现在已经很清楚了，在Just对象的subscribe方法中
observer.on(.next(_element))
observer.on(.completed)
return Disposables.create()
这里的Observer没有进行存放，Disposable也没有进行任何处理。
序列是什么呢？在哪？这里的序列就是【"🔴"】，只有一个元素。
其它由简入繁，提示:修改的源码注意还原错误处请指正😄
