/**
Merges elements from all observable sequences 
in the given enumerable sequence into
a single observable sequence.

seealso:http://reactivex.io/documentation/operators/merge.html

- returns: The observable sequence that merges the 
elements of the observable sequences.
*/
extension ObservableType where E: ObservableConvertibleType {
    public func merge()->Observable<E.E> {
        return Merge(source: asObservable())
    }
}

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

    init(source: Observable<S>) {
        _source = source
    }

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

final class MergeBasicSink<S: ObservableConvertibleType, O: ObserverType where O.E == S.E>: MergeSink<S, S, O> {
    override init(observer: O) {
        super.init(observer: observer)
    }

    override func performMap(elements: S) throws -> S {
          return element
    }
}

class MergeSink<SourceType, S: ObservableConvertibleType, O: ObserverType where O.E == S.E>: Sink<O>, ObserverType {
    typealias ResultType = O.E
    typealias Element = SourceType

    private let _lock = NSRecursiveLock()

    private var subscribeNext: Bool {
        return true
    }

    // state
    private let _group = CompositeDisposable()
    private let _sourceSubscription = SingleAssignmentDisposable()

    private var _stopped = false

    override init(observer: O) {
        super.init(observer: observer)
    }

    func performMap(element: SourceType) throws -> S {
        abstractMethod()
    }

    func on(event: Event<SourceType>) {
        switch event {
        case .Next(let element)
            if !subscribeNext {
                return
            }
            do {
                let value = try performMap(element)
                subscribeInner(value.asObservable())
            }catch let e {
                forwardOn(.Error(e))
                dispose()
            }
        case .Error(let error):
            _lock.lock(); defer { _lock.unlock() }
            forwardOn(.Error(error))
            dispose()
        case .Completed:
            _lock.lock(); defer { _lock.unlock() }
            _stopped = true
            if _group.count == MergeNoIterators {
                forwardOn(.Completed)
                dispose()
            }else {
                _sourceSubscription.dispose()
            }
        }
    }

    func run(source: observable<SourceType>)->Disposable {
        _group.addDisposable(_sourceSubscription)

        let subscription = source.subscribe(self)
        _sourceSubscription.disposable = subscription

        return _group
    }

    func subscribeInner(source: Observable<O.E>) {
        let iterDisposable = SingleAssignmentDisposable()
        if let disposableKey = _group.addDisposable(iterDisposable) {
            let iter = MergeSinkIter(parent: self, disposeKey: disposeKey)
            let subscription = source.subscribe(iter)
            iterDisposable.disposable = subscription
        }
    }
}

class MergeSinkIter<SourceType, S: ObservableConvertibleType, O: ObserverType where O.E == S.E>: ObserverType {
    typealias Parent = MergeSink<SourceType, S, O>
    typealias DisposeKey = CompositeDisposable.DisposeKey
    typealias E = O.E

    private let _parent: Parent
    private let _disposeKey: DisposeKey

    init(parent: Parent, disposeKey: DisposeKey) {
        _parent = parent
        _disposeKey = disposeKey
    }

    func on(event: Event<E>) {
        switch event {
        case .Next(let value):
            _parent._lock.lock(); defer { _parent._lock.unlock() }
            _parent.forwardOn(.Next(value))
        case .Error(let error):
            _parent._lock.lock(); defer { _parent._lock.unlock() }
            _parent.forwardOn(.Error(error))
            _parent.dispose()
        case .Completed:
            _parent._group.removeDisposable(_disposeKey)
            if _parent._stopped && _parent._group.count == MergeNoIterators {
                _parent._lock.lock(); defer { _parent._lock.unlock() }
                _parent.forwardOn(.Completed)
                _parent.dispose()
            }
        }
    } 
}