RxSwift 源碼學習之just

項目中一直使用着 RxSwift，如今也只是停留在使用的階段，最近利用閒暇的時間拜讀了一下 RxSwift 的源碼。

介紹

RxSwift 是一個能夠幫助咱們簡化異步編程的框架，它是 ReactiveX（簡寫：Rx） 的 Swift 版本。 RxSwift 拓展了觀察者模式，它能幫你自由組合多個異步事件，而不須要去關心線程，同步，線程安全，併發數據以及I/O阻塞。

RxSwift 屬於函數響應式編程。它是經過構建函數操做數據序列，而後對這些序列作出響應的編程方式。它結合了函數式編程以及響應式編程。

開篇

let observable =  Observable.just("hello")

observable.subscribe(onNext: { (str) in

    }, onError: { (error) in

    }, onCompleted: {
        //finished
    }).disposed(by: self.disposeBag)
複製代碼

個人記錄就從上面這個簡單的 RxSwift 使用示例開始，一步步去探尋 RxSwift 源碼。

源碼閱讀

Observable

Observable 定義：

public class Observable<Element> : ObservableType {
    init() {
    }
    
    public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        rxAbstractMethod()
    }
    
    public func asObservable() -> Observable<Element> {
        return self
    }

    /// Optimizations for map operator
    internal func composeMap<Result>(_ transform: @escaping (Element) throws -> Result) -> Observable<Result> {
        return _map(source: self, transform: transform)
    }
}
複製代碼

• init 方法被 internal 修飾，模塊外部不可見。主要是爲了監測 Rx 中的資源分配。
• 關鍵方法 public func subscribe<O: ObserverType>(_ observer: O) 實現中直接拋出異常；rxAbstractMethod 是對 fatalError的封裝。

just 函數

public static func just(_ element: Element) -> Observable<Element> {
    return Just(element: element)
}
複製代碼

Observable 的實例方法不可用，可是 Rx 中爲咱們擴展了許多類方法用於構造 Observable 實例。本文就以 just 方法爲例。

調用 just 函數時，傳入了一個數據，這裏利用泛型，咱們能夠傳入任何咱們須要的數據。而且返回一個 Observable 實例對象。 函數內部調用了一個私有類 Just：

final private class Just<Element>: Producer<Element> {
    private let _element: Element
    
    init(element: Element) {
        self._element = element
    }
    
    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        observer.on(.next(self._element))
        observer.on(.completed)
        return Disposables.create()
    }
}
複製代碼

私有類 Just 中持有了咱們傳入的數據 Element。

那麼，後續咱們就看看當咱們對 Observable 進行訂閱時，咱們是如何收到消息的。

subscribe 函數

咱們利用 subscribe 函數訂閱剛纔生成的 Observable 實例。咱們先來看看 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()
        }
    }
    return Disposables.create(
        self.asObservable().subscribe(observer),
        disposable
    )
}
複製代碼

能夠看到，函數有四個可選參數：onNext、onError、onCompleted、onDisposed。

而且在函數內部生成了一個 observer 對象，而且又利用 create 函數建立了一個 Observable 實例供外界使用。而且調用了 subscribe 方法將 observer 傳入。

此時，會來到前面咱們生成 Observable 實例時提到的的私有類 Just 中：

override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
    observer.on(.next(self._element))
    observer.on(.completed)
    return Disposables.create()
}
複製代碼

這裏出現了一個在 RxSwift 中常常出現的概念 Event，而且這個 Event 如今持有了 Just 中咱們生成 Observable 時傳入的 Element 對象：

public enum Event<Element> {
    /// Next element is produced.
    case next(Element)

    /// Sequence terminated with an error.
    case error(Swift.Error)

    /// Sequence completed successfully.
    case completed
}
複製代碼

這裏的 observer 就是在最開始時的 subscribe 函數中生成的。接下來，咱們來看看這個生成的 observer 究竟是何方神聖吧？

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()
    }
}
複製代碼

這是 subscribe 時，生成 observer 的代碼，咱們能夠看到這是一個 AnonymousObserver 的對象：

final class AnonymousObserver<Element>: ObserverBase<Element> {
    typealias EventHandler = (Event<Element>) -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
        self._eventHandler = eventHandler
    }

    override func onCore(_ event: Event<Element>) {
        return self._eventHandler(event)
    }
}
複製代碼

咱們能夠看到這裏持有了建立時傳入的閉包，而且在 onCore 方法調用時，執行了這個閉包。咱們繼續往上找，咱們來到 ObserverBase 中：

class ObserverBase<Element> : Disposable, ObserverType {
    private let _isStopped = AtomicInt(0)

    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)
            }
        }
    }

    func onCore(_ event: Event<Element>) {
        rxAbstractMethod()
    }

    func dispose() {
        fetchOr(self._isStopped, 1)
    }
}
複製代碼

這裏面咱們終於找到了處理私有類 Just 內 subscribe 函數中 observer.on(event) 的地方了，這裏判斷了 event 的類型，並調用了 onCore(event) 方法。下面咱們就回到了上面 AnonymousObserver 中的 onCore(event) 方法，並最終執行了建立 observer 時建立的閉包。

咱們再回頭看一下生成 observer 時的代碼。在傳入的閉包中，處理了 event，根據 event 的不一樣傳入對應的數據。咱們這裏關注 next 函數，這裏最終將咱們生成 Observable 時傳入的 Element 傳給了調用 subscribe 時傳入的 onNext 閉包。

這樣，當利用 just 函數生成一個 Observable 對象，而且訂閱值以後的相應處理也就完成了。

總結

這篇文章追尋一個簡單的 just 函數，探尋了 RxSwift 的源碼，咱們能夠看到 RxSwift 將咱們的處理事件的邏輯封裝成閉包，保存在合適的地方，並在須要處理相應事件時，將對應處理的閉包取出來處理。這個處理事件的方式在異步編程的優點就很是明顯了。

RxSwift 的源碼真的是很是簡潔明瞭，在閱讀源碼的過程當中不只可讓本身在使用時更加駕輕就熟，也能學習到大牛的一些編程思想。