Rx Combination Operators of Merg

使用示例

example("merge") {
    let disposeBag = DisposeBag()
    
    let subject1 = PublishSubject<String>()
    let subject2 = PublishSubject<String>()
    
    Observable.of(subject1, subject2)
        .merge()
        .subscribe(onNext: { print($0) })
        .disposed(by: disposeBag)
    
    subject1.onNext("🅰️")
    
    subject1.onNext("🅱️")
    
    subject2.onNext("①")
    
    subject2.onNext("②")
    
    subject1.onNext("🆎")
    
    subject2.onNext("③")
}

// output log 
--- merge example ---
🅰️
🅱️
①
②
🆎
③

类型推断

我们可能会写这种

Observable.of("🐶", "🐱", "🐭", "🐹")

但是

    let subject1 = PublishSubject<String>()
    let subject2 = PublishSubject<String>()
    
    Observable.of(subject1, subject2)

又是个什么鬼？

分析下of的函数原型

//  of operator
    public static func of(_ elements: E ..., scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<E> {
        return ObservableSequence(elements: elements, scheduler: scheduler)
    }

可以看到对范型E,没有做任何约束，所以Observable.of(subject1, subject2)至少从语法上来讲是合法的，结合上下文推断那应该是这样滴

    let observable: Observable<PublishSubject<String>> =  Observable.of(subject1, subject2)

准确的说是这样滴

ObservableSequence<S: Sequence> where S.Iterator.Element = PublishSubject<String>

验证一下：

    let observable: Observable<PublishSubject<String>> =  Observable.of(subject1, subject2)
    
    observable
       // .merge()
        .subscribe(onNext: { print($0) })
        .disposed(by: disposeBag)

//output log 
--- merge example ---
RxSwift.PublishSubject<Swift.String>
RxSwift.PublishSubject<Swift.String>

可以看到推断是正确的。

实现原理

final class Merge<SourceSequence: ObservableConvertibleType> : Producer<SourceSequence.E> {
    private let _source: Observable<SourceSequence>

    init(source: Observable<SourceSequence>) {
        _source = source
    }
    
    override func run<O: ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == SourceSequence.E {
        let sink = MergeBasicSink<SourceSequence, O>(observer: observer, cancel: cancel)
        let subscription = sink.run(_source)
        return (sink: sink, subscription: subscription)
    }
}

核心实现一如既往的放在Sink对象中

fileprivate final class MergeBasicSink<S: ObservableConvertibleType, O: ObserverType> : MergeSink<S, S, O> where O.E == S.E {
    override func performMap(_ element: S) throws -> S {
        return element
    }
}

MergeBasicSink 也没什么干料，还得看看它的父类

// MergeSink.run
    func run(_ source: Observable<SourceElement>) -> Disposable {
        let _ = _group.insert(_sourceSubscription)

        let subscription = source.subscribe(self)
        _sourceSubscription.setDisposable(subscription)
        
        return _group
    }

这里注意成功订阅了source, (注： source = ObservableSequence<S: Sequence> where S.Iterator.Element = PublishSubject<String>), 那么很快就会触发on方法啦

    func on(_ event: Event<SourceElement>) {
        switch event {
        case .next(let element):
            if let value = nextElementArrived(element: element) {
                subscribeInner(value.asObservable())
            }
        case .error(let error):
            _lock.lock(); defer { _lock.unlock() }
            forwardOn(.error(error))
            dispose()
        case .completed:
            _lock.lock(); defer { _lock.unlock() }
            _stopped = true
            _sourceSubscription.dispose()
            checkCompleted()
        }
    }

这个时候接收的事件是 onNext(PublishSubject<String>)

  @inline(__always)
    final private func nextElementArrived(element: SourceElement) -> SourceSequence? {
        _lock.lock(); defer { _lock.unlock() } // {
            if !subscribeNext {
                return nil
            }

            do {
                let value = try performMap(element)
                _activeCount += 1
                return value
            }
            catch let e {
                forwardOn(.error(e))
                dispose()
                return nil
            }
        // }
    }

nextElementArrived 核心操作就是performMap， 由于MergeBasicSink 只是简单的返回原始值，所以let value = nextElementArrived(element: element) 最终结果仍然是PublishSubject<String>

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

这一步相当关键，这里MergeSinkIter 成功订阅 iter， 注意 MergeSink 可以持有多个MergeSinkIter， MergeSinkIter的数量取决于被merge的数量

// MergeSinkIter.on
    func on(_ event: Event<E>) {
        _parent._lock.lock(); defer { _parent._lock.unlock() } // lock {
            switch event {
            case .next(let value):
                _parent.forwardOn(.next(value))
            case .error(let error):
                _parent.forwardOn(.error(error))
                _parent.dispose()
            case .completed:
                _parent._group.remove(for: _disposeKey)
                _parent._activeCount -= 1
                _parent.checkCompleted()
            }
        // }
    }

MergeSinkIter 在订阅事件后，会在收到信息的同时传递给MergeSink，至此整个流程走完.

总结

多个Observable<E>, 通过of操作聚焦为 ObservableSequence<S: Sequence> where S.Iterator.Element = Observable<E>
Merge操作会触发
MergeSink subscribe ObservableSequence,
MergeSink在接收到元素Observable<E>后， 会创建MergeSinkIter subscribe Observable<E>，
最终MergeSinkIter 在接收事件后又回传给MergeSink