RxSwift-map源碼解析

map操作符為每一個序列元素提供轉換，并返回到原序列。

map.png

看一段代碼示例：

Observable<Int>.of(1,2,3,4,5,6)
    .subscribe(onNext: { (val) in
        print(val)
    }).disposed(by: disposeBag)

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

Observable<Int>.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操作符，操作序列每個元素加10后作為新元素，構成新的序列

那么map是如何給序列重新設置新值的呢？閉包就像加工零件的數(shù)控機床，設定好加工程序$0+10就會對of中的每一個元素加工產出新的零件，看一下map源碼都做了哪些事情：

extension ObservableType {
    public func map<R>(_ transform: @escaping (E) throws -> R)
        -> Observable<R> {
        return self.asObservable().composeMap(transform)
    }
}
  • transform逃逸閉包，轉換邏輯交給業(yè)務層
  • asObservable()保證協(xié)議的一致性

首先看到map函數(shù)是一個帶閉包參數(shù)的ObservableType的擴展函數(shù)，內部調用了composeMap并傳入了外部的閉包以便內部調用。

由前邊的源碼探索經驗可猜測，該處閉包會被保留在內部，在訂閱時被使用，那么根據(jù)斷點一步步探索，看看外界的閉包最終會保留在何處。composeMap所在類：

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
          // 此處代碼有省略
    internal func composeMap<R>(_ transform: @escaping (Element) throws -> R) -> Observable<R> {
        return _map(source: self, transform: transform)
    }
}
  • source向_map函數(shù)傳入了self即為當前的序列對象
  • transform一路追蹤的外部閉包

是ObservableType的子類Observable實現(xiàn)了composeMap方法，返回Observable類型的對象，在內部調用了_map方法：

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

還是向Map內部傳入序列，及業(yè)務層閉包，一直強調序列和業(yè)務層閉包，主要由于結構復雜，以免被遺忘，后續(xù)和訂閱難以被聯(lián)系在一起。繼續(xù)查看Map類：

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

    private let _source: Observable<SourceType>

    private let _transform: Transform

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

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

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

    override func run<O: ObserverType>(_ 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，及發(fā)送序列元素到觀察者，再返回到訂閱，這里不再敘述。
  • Map中保留了源序列及業(yè)務層閉包方法
  • 此處run方法會在父類Producer類中方法調用，父類指針指向子類對象

繼續(xù)斷點運行就到達了我們的訂閱，該處方法和《RxSwift核心源碼探索》中的訂閱方法為同一方法：

extension ObservableType {
    //業(yè)務層訂閱調用
    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<E> { 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<Element> : Observable<Element> {
    override init() {
        super.init()
    }

    override func subscribe<O : ObserverType>(_ 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方法在此類實現(xiàn)
  • Map源碼中Producer的子類是Map，run方法在該處被實現(xiàn)

此處如果不太清楚可以追溯上文查看。上面有Map類的完整代碼，此處只查看調用方法代碼：

override func run<O: ObserverType>(_ 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來觸發(fā)觀察者的方法調用。subscribe方法中調用的

  • sinkAndSubscription = self.run(observer, cancel: disposer)
final private class ObservableSequence<S: Sequence>: Producer<S.Iterator.Element> {
    fileprivate let _elements: S
    fileprivate let _scheduler: ImmediateSchedulerType

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

    override func run<O : ObserverType>(_ 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)
    }
}

是繼承自Producer的方法，內部創(chuàng)建了ObservableSequenceSink對象并傳入了當前Observable對象和observer對象，最終調用了run()方法，此處猜測內部為變量序列并調用觀察者閉包方法，向外界發(fā)送消息。代碼如下：

final private class ObservableSequenceSink<S: Sequence, O: ObserverType>: Sink<O> where S.Iterator.Element == O.E {
    typealias Parent = ObservableSequence<S>

    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()
            }
        }
    }
}
  • 注意此類繼承自Sink，由此可知會調用Sink中的forwardOn方法

_elements是由of創(chuàng)建時保留的序列集合，此處對序列元素進行遍歷，并調用forwardOn方法發(fā)送元素。forwardOn：

class Sink<O : ObserverType> : 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<O.E>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        if isFlagSet(&self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
}
  • _observer是上面?zhèn)魅氲?code>MapSink對象

清楚看到在此處調用了sink的on方法，self._observer.on(event)。繼續(xù)追蹤MapSink類的on方法：

final private class MapSink<SourceType, O: ObserverType>: Sink<O>, 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<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):
            self.forwardOn(.error(error))
            self.dispose()
        case .completed:
            self.forwardOn(.completed)
            self.dispose()
        }
    }
}

到此處就很熟悉了，此處on和《RxSwift核心源碼探索》中不同：

元素處理代碼：

do {
    let mappedElement = try self._transform(element)
    self.forwardOn(.next(mappedElement))
}
  • let mappedElement = try self._transform(element)調用外界閉包獲取新值
  • self.forwardOn(.next(mappedElement))通過forwardOn將新值發(fā)送至訂閱者

最終會調用ObserverBase中的on方法，再調用觀察者observer的onCore方法，向觀察者發(fā)送元素。在由觀察者調用業(yè)務層訂閱時實現(xiàn)的閉包將序列元素發(fā)送到了業(yè)務層，到此map就完成了對源序列的修改。

總結：

實際上map就是對sink做了一層封裝，根據(jù)業(yè)務層的map設置在ObservableSequenceSink中處理了序列元素再發(fā)送至forwardOn直至Observer對象，由此完成了對元素的加工處理。

RxSwift源碼比較繞，復雜的邏輯帶來的是高效的開發(fā)，高效的運行，因此對RxSwfit源碼我們還需要進一步探索理解。

最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯(lián)系作者
【社區(qū)內容提示】社區(qū)部分內容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內容（如有圖片或視頻亦包括在內）由作者上傳并發(fā)布，文章內容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內容

  • 開篇 在上一篇中，我們分析了在RxSwift中的整個訂閱流程。在開講變換操作之前，首先要弄清楚Sink的概念，不清...
    Maru閱讀 1,942評論 3贊 9
  • 當程序員原來越浮躁了,項目做多了大都是雷同的, 對技術沒啥幫助,讀一些牛逼的第三方框架,有助于提升,關于RxSwi...
    水落斜陽閱讀 829評論 0贊 1
  • 開篇 一直覺得自己似乎越來越浮躁了，可能當代的大多數(shù)年輕人都活在恐慌里，問題已經從小時候的不思進取變成了“太思進取...
    Maru閱讀 3,869評論 13贊 26
  • 響應式編程是RxSwift的核心思想，統(tǒng)一思想快速開發(fā)。同樣在ReactiveX系列中其他語言也都使用了這一思想，...
    yahibo閱讀 741評論 0贊 7
  • 前言 通過上一篇內容RxSwift學習--核心邏輯初探,對RxSwift有了些初步的認知，下面通過源碼來看一下Rx...
    Henry_Jeannie閱讀 467評論 0贊 3

友情鏈接更多精彩內容