     typealias Element = SourceType
 
     private let _lock = NSRecursiveLock()
-    private let _source: Observable<SourceType>
 
     private var subscribeNext: Bool {
         return true

    }

    override func run<O: ObserverType where O.E == S.E>(observer: O) -> Disposable {
        let sink = FlatMapFirstSink(source: _source, selector: _selector, observer: observer)
        sink.disposable = sink.run()
        return sink
    }
}

    private let _sourceSubscription = SingleAssignmentDisposable()
    private let _group = CompositeDisposable()
    
    init(maxConcurrent: Int, observer: O) {
        _maxConcurrent = maxConcurrent
        
        _group.addDisposable(_sourceSubscription)
        super.init(observer: observer)
    }
    
    func run(source: Observable<S>) -> Disposable {
        _group.addDisposable(_sourceSubscription)
        
        let disposable = source.subscribe(self)
        _sourceSubscription.disposable = disposable
        return _group
    }
