RxSwift高階函數(shù)map解讀

1.map

通過一個轉(zhuǎn)換函數(shù)，將 Observable 的每個元素轉(zhuǎn)換一遍。

map.jpg

demo

let ob = Observable.of(1,2,3,4)
ob.map { (number) -> Int in
        return number + 2
    }
    .subscribe{
        print("\($0)")
    }
    .disposed(by: disposeBag)

輸出：
next(3)
next(4)
next(5)
next(6)
completed

解析

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

public class Observable<Element> : ObservableType {
    public func asObservable() -> Observable<Element> {
        return self
    }
    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)
}

源序列調(diào)用composeMap構(gòu)建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
    }

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

Map訂閱后，把MapSink作為觀察者轉(zhuǎn)交給源序列去訂閱。

final private class MapSink<SourceType, Observer: ObserverType>: Sink<Observer>, ObserverType {
    typealias Transform = (SourceType) throws -> ResultType
    typealias ResultType = Observer.Element 
    typealias Element = SourceType
    private let _transform: Transform

    init(transform: @escaping Transform, observer: Observer, 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()
        }
    }
}

源序列的響應(yīng)通過MapSink中_transform的映射后再響應(yīng)Map序列的訂閱。

2.flatMap

flatMap 操作符將源 Observable 的每一個元素應(yīng)用一個轉(zhuǎn)換方法，將他們轉(zhuǎn)換成 Observables。 然后將這些 Observables 的元素合并之后再發(fā)送出來。將可觀測序列發(fā)射的元素轉(zhuǎn)換為可觀測序列，并將兩個可觀測序列的發(fā)射合并為一個可觀測序列。

flatMap.jpg

這個操作符是非常有用的，例如，當(dāng) Observable 的元素本身擁有其他的 Observable 時，你可以將所有子 Observables 的元素發(fā)送出來。

demo

let disposeBag = DisposeBag()
let first = BehaviorSubject(value: "????")
let second = BehaviorSubject(value: "???")
let variable = BehaviorSubject(value: first)

variable.asObservable()
    .flatMap { $0 }
    .subscribe(onNext: { print("訂閱到了：\($0)") })
    .disposed(by: disposeBag)

first.onNext("??")
variable.onNext(second)
second.onNext("???")
first.onNext("??")

輸出：
訂閱到了：????
訂閱到了：??
訂閱到了：???
訂閱到了：???
訂閱到了：??

解析

先看flatMap函數(shù)：

extension ObservableType {
    public func flatMap<Source: ObservableConvertibleType>(_ selector: @escaping (Element) throws -> Source)
        -> Observable<Source.Element> {
            return FlatMap(source: self.asObservable(), selector: selector)
    }

}

和map一樣，內(nèi)部構(gòu)建了另一個序列FlatMap。

final private class FlatMap<SourceElement, SourceSequence: ObservableConvertibleType>: Producer<SourceSequence.Element> {
    init(source: Observable<SourceElement>, selector: @escaping Selector) {
        self._source = source
        self._selector = selector
    }
    
    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == SourceSequence.Element {
        let sink = FlatMapSink(selector: self._selector, observer: observer, cancel: cancel)
        let subscription = sink.run(self._source)
        return (sink: sink, subscription: subscription)
    }
}

FlatMap保存了_source和_selector后，就轉(zhuǎn)手給FlatMapSink。

fileprivate final class FlatMapSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType> : MergeSink<SourceElement, SourceSequence, Observer> where Observer.Element == SourceSequence.Element {
    init(selector: @escaping Selector, observer: Observer, cancel: Cancelable) {
        self._selector = selector
        super.init(observer: observer, cancel: cancel)
    }

    override func performMap(_ element: SourceElement) throws -> SourceSequence {
        return try self._selector(element)
    }
}

FlatMapSink正是我們上一篇 merge 函數(shù)解讀中的 MergeSink 的子類（這里不再重復(fù)貼源碼了）。FlatMapSink作為子類，也是只實現(xiàn)了performMap，通過回調(diào) flatMap 的閉包獲取源序列映射后的新的序列。

最后將這個新的合并后的序列的訂閱響應(yīng)通過MergeSinkIter轉(zhuǎn)交給FlatMapSink，外界就會收到這個新序列的信號。