         }
     }
     
-    func run(_ source: Observable<SourceType>) -> Disposable {
+    func run(_ source: Observable<SourceElement>) -> Disposable {
         let _ = _group.insert(_sourceSubscription)
 
         let subscription = source.subscribe(self)

    var _activeCount = 0
    var _stopped = false

    override init(observer: O, cancel: Cancelable) {
        super.init(observer: observer, cancel: cancel)
    }

    func performMap(_ element: SourceType) throws -> S {
        rxAbstractMethod()
    }
    
    func on(_ event: Event<SourceType>) {
        _lock.lock(); defer { _lock.unlock() } // lock {
            switch event {
            case .next(let element):


// MARK: flatMap

fileprivate final class FlatMapSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType> : MergeSink<SourceElement, SourceSequence, Observer> where Observer.E == SourceSequence.E {
    typealias Selector = (SourceElement) throws -> SourceSequence

    private let _selector: Selector

    init(selector: @escaping Selector, observer: Observer, cancel: Cancelable) {
        _selector = selector
        super.init(observer: observer, cancel: cancel)
    }

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

fileprivate final class FlatMapWithIndexSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType> : MergeSink<SourceElement, SourceSequence, Observer> where Observer.E == SourceSequence.E {
    typealias Selector = (SourceElement, Int) throws -> SourceSequence

    private var _index = 0
    private let _selector: Selector

    init(selector: @escaping Selector, observer: Observer, cancel: Cancelable) {
        _selector = selector
        super.init(observer: observer, cancel: cancel)
    }

    override func performMap(_ element: SourceElement) throws -> SourceSequence {
        return try _selector(element, try incrementChecked(&_index))
    }
}

// MARK: FlatMapFirst

fileprivate final class FlatMapFirstSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType> : MergeSink<SourceElement, SourceSequence, Observer> where Observer.E == SourceSequence.E {
    typealias Selector = (SourceElement) throws -> SourceSequence

    private let _selector: Selector

