RxSwift高阶函数map解读

1.map

通过一个转换函数，将 Observable 的每个元素转换一遍。

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

源序列调用composeMap构建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作为观察者转交给源序列去订阅。

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

源序列的响应通过MapSink中_transform的映射后再响应Map序列的订阅。

2.flatMap

flatMap 操作符将源 Observable 的每一个元素应用一个转换方法，将他们转换成 Observables。 然后将这些 Observables 的元素合并之后再发送出来。将可观测序列发射的元素转换为可观测序列，并将两个可观测序列的发射合并为一个可观测序列。

flatMap.jpg

这个操作符是非常有用的，例如，当 Observable 的元素本身拥有其他的 Observable 时，你可以将所有子 Observables 的元素发送出来。

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函数：

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

}

和map一样，内部构建了另一个序列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后，就转手给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 函数解读中的 MergeSink 的子类（这里不再重复贴源码了）。FlatMapSink作为子类，也是只实现了performMap，通过回调 flatMap 的闭包获取源序列映射后的新的序列。

最后将这个新的合并后的序列的订阅响应通过MergeSinkIter转交给FlatMapSink，外界就会收到这个新序列的信号。