     }
 }
 
-final fileprivate class Merge<S: ObservableConvertibleType> : Producer<S.E> {
+final class ConcatMap<SourceType, S: ObservableConvertibleType>: Producer<S.E> {
+    typealias Selector = (SourceType) throws -> S
+    
+    private let _source: Observable<SourceType>
+    private let _selector: Selector
+    
+    init(source: Observable<SourceType>, selector: @escaping Selector) {
+        _source = source
+        _selector = selector
+    }
+    
+    override func run<O: ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == S.E {
+        let sink = MapConcatSink<SourceType, S, O>(selector: _selector, observer: observer, cancel: cancel)
+        let subscription = sink.run(_source)
+        return (sink: sink, subscription: subscription)
+    }
+}
+
+final class Merge<S: ObservableConvertibleType> : Producer<S.E> {
     private let _source: Observable<S>
 
     init(source: Observable<S>) {

    }
}

fileprivate final class MergeLimitedSinkIter<S: ObservableConvertibleType, O: ObserverType>
    : ObserverType
    , LockOwnerType
    , SynchronizedOnType where S.E == O.E {
    typealias E = O.E
    typealias DisposeKey = CompositeDisposable.DisposeKey
    typealias Parent = MergeLimitedSink<S, O>
    
    private let _parent: Parent
    private let _disposeKey: DisposeKey

        super.init(observer: observer, cancel: cancel)
    }
    
    func run(_ source: Observable<SourceType>) -> Disposable {
        let _ = _group.insert(_sourceSubscription)
        
        let disposable = source.subscribe(self)
