     }
 }
 
-fileprivate final class ConcatMapSink<SourceType, S: ObservableConvertibleType, O: ObserverType>: MergeLimitedSink<SourceType, S, O> where O.E == S.E {
-    typealias Selector = (SourceType) throws -> S
-    
-    private let _selector: Selector
-    
-    init(selector: @escaping Selector, observer: O, cancel: Cancelable) {
+fileprivate final class ConcatMapSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>: MergeLimitedSink<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(maxConcurrent: 1, observer: observer, cancel: cancel)
     }
     
-    override func performMap(_ element: SourceType) throws -> S {
+    override func performMap(_ element: SourceElement) throws -> SourceSequence {
         return try _selector(element)
     }
 }
 
-fileprivate final class MergeLimitedBasicSink<S: ObservableConvertibleType, O: ObserverType>: MergeLimitedSink<S, S, O> where O.E == S.E {
-    
-    override func performMap(_ element: S) throws -> S {
+fileprivate final class MergeLimitedBasicSink<SourceSequence: ObservableConvertibleType, Observer: ObserverType>: MergeLimitedSink<SourceSequence, SourceSequence, Observer> where Observer.E == SourceSequence.E {
+    
+    override func performMap(_ element: SourceSequence) throws -> SourceSequence {
         return element
     }
 }
 
-fileprivate class MergeLimitedSink<SourceType, S: ObservableConvertibleType, O: ObserverType>
-    : Sink<O>
+fileprivate class MergeLimitedSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>
+    : Sink<Observer>
     , LockOwnerType
-    , SynchronizedOnType where O.E == S.E  {
-    typealias QueueType = Queue<S>
+    , SynchronizedOnType where Observer.E == SourceSequence.E  {
+    typealias QueueType = Queue<SourceSequence>
 
     let _maxConcurrent: Int
 

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

}


fileprivate class MergeSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>
    : Sink<Observer>
    , ObserverType where Observer.E == SourceSequence.E {
    typealias ResultType = Observer.E
    typealias Element = SourceElement

    let _lock = RecursiveLock()

