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 的源码真的是很是简洁明了，在阅读源码的过程当中不只可让本身在使用时更加驾轻就熟，也能学习到大牛的一些编程思想。