RxSwift-map源码解析

map 操作符为每一个序列元素提供转换，并返回到原序列。

看一段代码示例：

Observable.of(1,2,3,4,5,6)
    .subscribe(onNext: { (val) in
        print(val)
    }).disposed(by: disposeBag)
复制代码

输出：1，2，3，4，5，6

Observable.of(1,2,3,4,5,6)
    .map{
        $0+10
    }
    .subscribe(onNext: { (val) in
        print(val)
    }).disposed(by: disposeBag)
复制代码

输出：11，12，13，14，15，16

of
map

那么 map 是如何给序列重新设置新值的呢？闭包就像加工零件的数控机床，设定好加工程序 $0+10 就会对 of 中的每一个元素加工产出新的零件，看一下 map 源码都做了哪些事情：

extension ObservableType {
    public func map(_ transform: @escaping (E) throws -> R)
        -> Observable {
        return self.asObservable().composeMap(transform)
    }
}
复制代码

transform
asObservable()

首先看到 map 函数是一个带闭包参数的 ObservableType 的扩展函数，内部调用了 composeMap 并传入了外部的闭包以便内部调用。

由前边的源码探索经验可猜测，该处闭包会被保留在内部，在订阅时被使用，那么根据断点一步步探索，看看外界的闭包最终会保留在何处。 composeMap 所在类：

public class Observable : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
          // 此处代码有省略
    internal func composeMap(_ transform: @escaping (Element) throws -> R) -> Observable {
        return _map(source: self, transform: transform)
    }
}
复制代码

• source 向 _map 函数传入了 self 即为当前的序列对象
• transform 一路追踪的外部闭包

是 ObservableType 的子类 Observable 实现了 composeMap 方法，返回 Observable 类型的对象，在内部调用了 _map 方法：

internal func _map(source: Observable, transform: @escaping (Element) throws -> R) -> Observable {
    return Map(source: source, transform: transform)
}
复制代码

还是向 Map 内部传入序列，及业务层闭包，一直强调序列和业务层闭包，主要由于结构复杂，以免被遗忘，后续和订阅难以被联系在一起。继续查看 Map 类：

final private class Map: Producer {
    typealias Transform = (SourceType) throws -> ResultType

    private let _source: Observable

    private let _transform: Transform

    init(source: Observable, transform: @escaping Transform) {
        self._source = source
        self._transform = transform

#if TRACE_RESOURCES
        _ = increment(&_numberOfMapOperators)
#endif
    }

    override func composeMap(_ selector: @escaping (ResultType) throws -> R) -> Observable {
        let originalSelector = self._transform
        return Map(source: self._source, transform: { (s: SourceType) throws -> R in
            let r: ResultType = try originalSelector(s)
            return try selector(r)
        })
    }

    override func run(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == ResultType {
        let sink = MapSink(transform: self._transform, observer: observer, cancel: cancel)
        let subscription = self._source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }
}
复制代码

• 继承自 Producer ，在 《RxSwift核心源码探索》 中我们已经很熟悉了，继承自 Observable ，提供了连接序列和观察者的方法对象 sink ，及发送序列元素到观察者，再返回到订阅，这里不再叙述。
• Map 中保留了源序列及业务层闭包方法
• 此处 run 方法会在父类 Producer 类中方法调用，父类指针指向子类对象

继续断点运行就到达了我们的订阅，该处方法和 《RxSwift核心源码探索》 中的订阅方法为同一方法：

extension ObservableType {
    //业务层订阅调用
    public func subscribe(onNext: ((E) -> 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()
            }
            
            #if DEBUG
                let synchronizationTracker = SynchronizationTracker()
            #endif
            
            let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
            
            let observer = AnonymousObserver { event in
                
                #if DEBUG
                    synchronizationTracker.register(synchronizationErrorMessage: .default)
                    defer { synchronizationTracker.unregister() }
                #endif
                
                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
            )
    }
}
复制代码

self.asObservable().subscribe(observer) 此处调用的则是 Producer 中的 subscribe 方法，看一下该处方法：

class Producer : Observable {
    override init() {
        super.init()
    }

    override func subscribe(_ 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
            }
        }
    }
}
复制代码

此方法很熟悉，主要看一下内部 self.run 方法调用，此处继承链和 《RxSwift核心源码探索》 中的继承链不同，继承链如下：

• RxSwift 核心源码探索中 Producer 的子类是 AnonymousObservable ， run 方法在此类实现
• Map 源码中Producer的子类是 Map ， run 方法在该处被实现

此处如果不太清楚可以追溯上文查看。上面有 Map 类的完整代码，此处只查看调用方法代码：

override func run(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == ResultType {
    let sink = MapSink(transform: self._transform, observer: observer, cancel: cancel)
    let subscription = self._source.subscribe(sink)
    return (sink: sink, subscription: subscription)
}
复制代码

• 调用了 MapSink 方法此处和 《RxSwift核心源码探索》 中的 AnnonymousObservableSink 类似
• self._source 此处为订阅时保存的闭包
• .subscribe(sink)Producer 类的方法，传入 sink 用来调用 sink 中的 on 方法

类似于 《RxSwift核心源码探索》 中的 Sink ，功能是一样的， MapSink 中保留的是观察者， Map 中保留的为可观察序列 Observable ，通过 Observable 来触发观察者的方法调用。 subscribe 方法中调用的

• sinkAndSubscription = self.run(observer, cancel: disposer)

final private class ObservableSequence: Producer {
    fileprivate let _elements: S
    fileprivate let _scheduler: ImmediateSchedulerType

    init(elements: S, scheduler: ImmediateSchedulerType) {
        self._elements = elements
        self._scheduler = scheduler
    }

    override func run(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == E {
        let sink = ObservableSequenceSink(parent: self, observer: observer, cancel: cancel)
        let subscription = sink.run()
        return (sink: sink, subscription: subscription)
    }
}
复制代码

final private class ObservableSequenceSink: Sink where S.Iterator.Element == O.E {
    typealias Parent = ObservableSequence

    private let _parent: Parent

    init(parent: Parent, observer: O, cancel: Cancelable) {
        self._parent = parent
        super.init(observer: observer, cancel: cancel)
    }

    func run() -> Disposable {
        return self._parent._scheduler.scheduleRecursive(self._parent._elements.makeIterator()) { iterator, recurse in
            var mutableIterator = iterator
            if let next = mutableIterator.next() {
                self.forwardOn(.next(next))
                recurse(mutableIterator)
            }
            else {
                self.forwardOn(.completed)
                self.dispose()
            }
        }
    }
}
复制代码

class Sink : Disposable {
    fileprivate let _observer: O
    fileprivate let _cancel: Cancelable
    fileprivate var _disposed = AtomicInt(0)

    #if DEBUG
        fileprivate let _synchronizationTracker = SynchronizationTracker()
    #endif

    init(observer: O, cancel: Cancelable) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self._observer = observer
        self._cancel = cancel
    }

    final func forwardOn(_ event: Event) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        if isFlagSet(&self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
}
复制代码

final private class MapSink: Sink, ObserverType {
    typealias Transform = (SourceType) throws -> ResultType

    typealias ResultType = O.E
    typealias Element = SourceType

    private let _transform: Transform

    init(transform: @escaping Transform, observer: O, cancel: Cancelable) {
        self._transform = transform
        super.init(observer: observer, cancel: cancel)
    }

    func on(_ event: Event) {
        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):
            self.forwardOn(.error(error))
            self.dispose()
        case .completed:
            self.forwardOn(.completed)
            self.dispose()
        }
    }
}
复制代码

do {
    let mappedElement = try self._transform(element)
    self.forwardOn(.next(mappedElement))
}
复制代码