RxSWift Combination Operators of

使用示例

example("switchLatest") {
    let disposeBag = DisposeBag()
    
    let subject1 = BehaviorSubject(value: "⚽️")
    let subject2 = BehaviorSubject(value: "🍎")
    
    let variable = Variable(subject1)
        
    variable.asObservable()
        .switchLatest()
        .subscribe(onNext: { print($0) })
        .disposed(by: disposeBag)
    
    subject1.onNext("🏈")
    subject1.onNext("🏀")
    
    variable.value = subject2
    
    subject1.onNext("⚾️")
    
    subject2.onNext("🍐")
}

// out log 
--- switchLatest example ---
ℹ️ [DEPRECATED] `Variable` is planned for future deprecation. Please consider `BehaviorRelay` as a replacement. Read more at: https://git.io/vNqvx
⚽️
🏈
🏀
🍎
🍐

switchLatest

switchLatest操作符的实现和Merge操作符原理相似。

extension ObservableType where E : ObservableConvertibleType {

    /**
     Transforms an observable sequence of observable sequences into an observable sequence
     producing values only from the most recent observable sequence.

     Each time a new inner observable sequence is received, unsubscribe from the
     previous inner observable sequence.

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

     - returns: The observable sequence that at any point in time produces the elements of the most recent inner observable sequence that has been received.
     */
    public func switchLatest() -> Observable<E.E> {
        return Switch(source: asObservable())
    }
}

结合函数原型很容易推断出示例代码中 variable 为 Variable<BehaviorSubject<String>> 类型。

final fileprivate class Switch<S: ObservableConvertibleType> : Producer<S.E> {
    fileprivate let _source: Observable<S>
    
    init(source: Observable<S>) {
        _source = source
    }
    
    override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == S.E {
        let sink = SwitchIdentitySink<S, O>(observer: observer, cancel: cancel)
        let subscription = sink.run(_source)
        return (sink: sink, subscription: subscription)
    }
}

还是得看看Sink 具体是怎么实现的

final fileprivate class SwitchIdentitySink<S: ObservableConvertibleType, O: ObserverType> : SwitchSink<S, S, O> where O.E == S.E {
    override init(observer: O, cancel: Cancelable) {
        super.init(observer: observer, cancel: cancel)
    }

    override func performMap(_ element: S) throws -> S {
        return element
    }
}

SwitchIdentitySink 继承于SwitchSink， 仅仅是简单的override performMap函数

// SwitchSink
    func run(_ source: Observable<SourceType>) -> Disposable {
        let subscription = source.subscribe(self)
        _subscriptions.setDisposable(subscription)
        return Disposables.create(_subscriptions, _innerSubscription)
    }

SwitchSink 果断订阅Observable。

    func on(_ event: Event<E>) {
        switch event {
        case .next(let element):
            if let (latest, observable) = nextElementArrived(element: element) {
                let d = SingleAssignmentDisposable()
                _innerSubscription.disposable = d
                   
                let observer = SwitchSinkIter(parent: self, id: latest, _self: d)
                let disposable = observable.subscribe(observer)
                d.setDisposable(disposable)
            }
        case .error(let error):
            _lock.lock(); defer { _lock.unlock() }
            forwardOn(.error(error))
            dispose()
        case .completed:
            _lock.lock(); defer { _lock.unlock() }
            _stopped = true
            
            _subscriptions.dispose()
            
            if !_hasLatest {
                forwardOn(.completed)
                dispose()
            }
        }
    }

顺利成章的就会走on函数，这里对比一下：Merge.on

// Merge
 func on(_ event: Event<SourceElement>) {
        switch event {
        case .next(let element):
            if let value = nextElementArrived(element: element) {
                subscribeInner(value.asObservable())
            }
            ...
          }
       }
       
       
   func subscribeInner(_ source: Observable<Observer.E>) {
        let iterDisposable = SingleAssignmentDisposable()
        if let disposeKey = _group.insert(iterDisposable) {
            let iter = MergeSinkIter(parent: self, disposeKey: disposeKey)
            let subscription = source.subscribe(iter)
            iterDisposable.setDisposable(subscription)
        }
    }

Merge 使用的CompositeDisposable 管理每一个Disposable， 而SwitchLatest只使用一个变量_innerSubscription存储Disposable，这样它永远只会持有一个Disposable，当它持有一个新的Disposable的时候，那个老的就不再持有，自然老的最终会被释放掉。

// SwitchSink
    @inline(__always)
    final private func nextElementArrived(element: E) -> (Int, Observable<S.E>)? {
        _lock.lock(); defer { _lock.unlock() } // {
            do {
                let observable = try performMap(element).asObservable()
                _hasLatest = true
                _latest = _latest &+ 1
                return (_latest, observable)
            }
            catch let error {
                forwardOn(.error(error))
                dispose()
            }

            return nil
        // }
    }

这里更新了_latest的值。

// SwitchSinkIter
    func _synchronized_on(_ event: Event<E>) {
        switch event {
        case .next: break
        case .error, .completed:
            _self.dispose()
        }
        
        if _parent._latest != _id {
            return
        }
       
        switch event {
        case .next:
            _parent.forwardOn(event)
        case .error:
            _parent.forwardOn(event)
            _parent.dispose()
        case .completed:
            _parent._hasLatest = false
            if _parent._stopped {
                _parent.forwardOn(event)
                _parent.dispose()
            }
        }
    }

这里再次检测 _parent._latest 的值 是否与自身_id相匹配， 如果不匹配则无效。