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

    }
}

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

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