RxSwift源码分析(8)——高阶函数核心逻辑

想要使用RxSwift使用得爽，肯定绕不开高阶函数，比如有map、combineLatest、Driver、publish和之前分析过的skip等等。熟练使用各个高阶函数肯定会更加感受到RxSwift的魅力。
想要更好的使用高阶函数，自然也要探讨一下源码。以map函数为例，看看下面的例子：

Observable.of(1,2,3)
.map { (num) -> Int in
    return num+1
}.subscribe { (num) in
    print(num)
}.disposed(by: disposeBag)

这段代码的结果是打印：2，3，4。
就是在print之前执行了num+1。如果不熟悉RxSwift的朋友，不知道你们觉得这个例子看上去是不是挺复杂的，但是我们可以把它拆分成几个部分：

//创建序列
let ob = Observable.of(1,2,3)
//map
let map = ob.map { (num) -> Int in
    return num+1
}
//订阅
let dispose = map.subscribe { (num) in
    print(num)
}
//销毁，这部分不看
dispose.disposed(by: disposeBag)

1. 我们首先来看第一个部分的源码，就是创建了一个序列，保存了一些属性：

let ob = Observable.of(1,2,3)

extension ObservableType {
    public static func of(_ elements: Element ..., scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<Element> {
        return ObservableSequence(elements: elements, scheduler: scheduler)
    }
}

final private class ObservableSequence<Sequence: Swift.Sequence>: Producer<Sequence.Element> {
    ...
    init(elements: Sequence, scheduler: ImmediateSchedulerType) {
        self._elements = elements
        self._scheduler = scheduler
    }
    ...
}

2. 然后来看第二部分源码，一路走下去，发现也就是返回了一个Map序列（Map继承了Producer），保存了source和transform：

let map = ob.map { (num) -> Int in
    return num+1
}

extension ObservableType {
    public func map<Result>(_ transform: @escaping (Element) throws -> Result)
        -> Observable<Result> {
        return self.asObservable().composeMap(transform)
    }
}

public class Observable<Element> : ObservableType {
    ...
    internal func composeMap<Result>(_ transform: @escaping (Element) throws -> Result) -> Observable<Result> {
        return _map(source: self, transform: transform)
    }
}

internal func _map<Element, Result>(source: Observable<Element>, transform: @escaping (Element) throws -> Result) -> Observable<Result> {
    return Map(source: source, transform: transform)
}

final private class Map<SourceType, ResultType>: Producer<ResultType> {
    ...
    init(source: Observable<SourceType>, transform: @escaping Transform) {
        self._source = source
        self._transform = transform
        ...
    }

而source和transform其实就是外面传进来的，一步一步往回走就能知道：

3. 接着我们看第三部分源码，Map序列进行订阅：

let dispose = map.subscribe { (num) in
    print(num)
}

根据RxSwift核心逻辑一步一步走流程，将会来到Map的run函数：

final private class Map<SourceType, ResultType>: Producer<ResultType> {
    ...
    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == ResultType {
        let sink = MapSink(transform: self._transform, observer: observer, cancel: cancel)
        let subscription = self._source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }
    ...
}

• 注意一下这里，这里是高阶函数重点。在之前核心逻辑里说的基础序列的run函数里，内部是调用sink.run：

而现在的高阶函数序列里，则是利用self.source源序列调用subscribe进行订阅sink：

这个self.source源序列才是我们之前说过的基础序列，所以在这里才开始进行真正的订阅。

4. 根据RxSwift核心逻辑，源序列订阅后来到MapSink的on函数：

final private class MapSink<SourceType, Observer: ObserverType>: Sink<Observer>, ObserverType {
    ...
    func on(_ event: Event<SourceType>) {
        switch event {
        case .next(let element):
            do {
                let mappedElement = try self._transform(element)
                self.forwardOn(.next(mappedElement))
            }
            catch let e {
                self.forwardOn(.error(e))
                self.dispose()
            }
        case .error(let error):  ...
        case .completed:  ...
        }
    }
}

5. 在进行self.forwardOn(.next())之前会先调用self._transform()，而self._transform就是之前保存的尾随闭包，这便是重点：

.map { (num) -> Int in
    return num+1
}

所以会先进行num+1返回新值，然后再self.forwardOn(.next())把新值发送出去，源数据(1,2,3)会变成(2,3,4)。所以在每个元素分别进行发送的时候，在最终响应之前都先进行self._transform()再self.forwardOn(.next())发送响应。

6. 根据RxSwift核心逻辑，走完剩下的流程，来到外面的响应闭包，便完整走完高阶函数的响应流程：

subscribe { (num) in
    print(num)
}

高阶函数可以简单理解为，另外创建一个新序列保存源序列，也就是序列嵌套序列，先用新序列进行处理，然后再用源序列进行订阅和响应