     }
 }
 
-fileprivate final class MergeLimitedSink<S: ObservableConvertibleType, O: ObserverType>
+fileprivate final class MapConcatSink<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) {
+        _selector = selector
+        super.init(maxConcurrent: 1, observer: observer, cancel: cancel)
+    }
+    
+    override func performMap(_ element: SourceType) throws -> S {
+        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 {
+        return element
+    }
+}
+
+fileprivate class MergeLimitedSink<SourceType, S: ObservableConvertibleType, O: ObserverType>
     : Sink<O>
-    , ObserverType
     , LockOwnerType
-    , SynchronizedOnType where S.E == O.E {
-    typealias E = S
+    , SynchronizedOnType where O.E == S.E  {
     typealias QueueType = Queue<S>
 
     let _maxConcurrent: Int

    }
}

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

    init(source: Observable<S>) {

        return _group
    }
    
    func subscribe(_ innerSource: S, group: CompositeDisposable) {
        let subscription = SingleAssignmentDisposable()
        
        let key = group.insert(subscription)
        
        if let key = key {
            let observer = MergeLimitedSinkIter<SourceType, S, O>(parent: self, disposeKey: key)
            
            let disposable = innerSource.asObservable().subscribe(observer)
            subscription.setDisposable(disposable)
        }
    }
    
    func performMap(_ element: SourceType) throws -> S {
        rxAbstractMethod()
    }
    
    func on(_ event: Event<SourceType>) {
        synchronizedOn(event)
    }

    func _synchronized_on(_ event: Event<SourceType>) {
        switch event {
        case .next(let element):
            var subscribe: Bool = false
            if _activeCount < _maxConcurrent {
                _activeCount += 1
                subscribe = true
            }
            else {
                do {
                    let value = try performMap(element)
                    _queue.enqueue(value)
                    subscribe = false
                } catch {
                    forwardOn(.error(error))
                    dispose()
                }
            }

            if subscribe {
                do {
                    let value = try performMap(element)
                    self.subscribe(value, group: _group)
                } catch {
                    forwardOn(.error(error))
                    dispose()
                }
            }
        case .error(let error):
            forwardOn(.error(error))
