Rx Transforming Operators of Map

使用示例

Applies a transforming closure to elements emitted by an Observable sequence, and returns a new Observable sequence of the transformed elements

example("map") {
    let disposeBag = DisposeBag()
    Observable.of(1, 2, 3)
        .map { $0 * $0 }
        .subscribe(onNext: { print($0) })
        .disposed(by: disposeBag)
}

// output log
--- map example ---
1
4
9

Map 操作入口

extension ObservableType {

    /**
     Projects each element of an observable sequence into a new form.

     - seealso: [map operator on reactivex.io](http://reactivex.io/documentation/operators/map.html)

     - parameter transform: A transform function to apply to each source element.
     - returns: An observable sequence whose elements are the result of invoking the transform function on each element of source.

     */
    public func map<R>(_ transform: @escaping (E) throws -> R)
        -> Observable<R> {
        return self.asObservable().composeMap(transform)
    }
}

// Observable Class
    internal func composeMap<R>(_ transform: @escaping (Element) throws -> R) -> Observable<R> {
        return _map(source: self, transform: transform)
    }

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

Map 实现

map 操作最终借助于Map Class实现， 看看 Map究竟是怎么实现转换操作的

// Map Class
    override func run<O: ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == ResultType {
        let sink = MapSink(transform: _transform, observer: observer, cancel: cancel)
        let subscription = _source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }

这个应该比较熟悉，MapSink才是主菜

final fileprivate 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) {
        _transform = transform
        super.init(observer: observer, cancel: cancel)
    }

    func on(_ event: Event<SourceType>) {
        switch event {
        case .next(let element):
            do {
                let mappedElement = try _transform(element)
                forwardOn(.next(mappedElement))
            }
            catch let e {
                forwardOn(.error(e))
                dispose()
            }
        case .error(let error):
            forwardOn(.error(error))
            dispose()
        case .completed:
            forwardOn(.completed)
            dispose()
        }
    }
}

MapSink 订阅了 source,（let subscription = _source.subscribe(sink)） , 那么source的变化最终会通过AnyObserver或者其他形式Observer传递给MapSink， 触发MapSink.on 方法，MapSink.on 将next事件元素通过transform 闭包转换成最终形式，然后通过forwardOn 传递给observer

override composeMap

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

Map 对composeMap 方法进行重写，能够有效节省内存，我们经常会写类似这样的代码：
Observable.of(1, 2, 3).map(transform0).map(transform1)， Map 通过重写composeMap 方法，有效的避免了中间对象的产生。
上述示例如果使用默认实现会产生2个 Map对象， 2个MapSink 对象
而如果使用优化的代码则无论多少个map操作，最终都只会产生1个Map对象，1个MapSink对象。

Map.composeMap 每次都保留原始的Observable， 然后通过创建新的transform 达到转换的目的。