local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

--------------------------------------------------------------------------------

local MergeAllInnerObserver = Rx.Class 'Rx.internals.MergeAllInnerObserver' {

    AbstractObserver,
        -- superclass

    _parent = false,
        -- (Observer)

    _sad = false,
        -- (SingleAssignmentDisposable)

    ----------------------------------------------------------------------------

    __init = function(self, parent, sad)

        self._parent = parent
        self._sad = sad

        return function()
            self._sad:dispose()
        end

    end,

    ----------------------------------------------------------------------------

    next = function(self, ...)

        self._parent._observer:onNext(...)

    end,

    ----------------------------------------------------------------------------

    error = function(self, ...)

        self._parent._observer:onError(...)

    end,

    ----------------------------------------------------------------------------

    completed = function(self)

        local parent = self._parent
        parent._disposableList:remove(self)

        if parent._done and parent._disposableList:length() == 1 then

            -- Only the MergeAllObserver is left in the _disposableList send
            -- onCompleted as all the MergeAllInnerObserver's have observed
            -- completion and have been removed from the list.

            parent._observer:onCompleted()

        end

    end,

}

--------------------------------------------------------------------------------

local MergeAllObserver = Rx.Class 'Rx.internals.MergeAllObserver' {

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _disposableList = false,
        -- (CompositeDisposable)

    _done = false,
        -- (boolean)

    ----------------------------------------------------------------------------

    __init = function(self, observer, disposableList)

        self._observer = observer
        self._disposableList = disposableList
        self._done = false

    end,

    ----------------------------------------------------------------------------

    next = function(self, innerSource)

        local sad = SingleAssignmentDisposable()
        local innerObserver = MergeAllInnerObserver(self, sad)
        self._disposableList:add(innerObserver)
        sad:setDisposable(innerSource:subscribe(innerObserver))

    end,

    ----------------------------------------------------------------------------

    error = function(self, ...)

        self._observer:onError(...)

    end,

    ----------------------------------------------------------------------------

    completed = function(self)

        self._done = true

        if self._disposableList:length() == 1 then
            self._observer:onCompleted()
        end

    end,

}

--------------------------------------------------------------------------------

local MergeAllObservable = Rx.Class 'Rx.internals.MergeAllObservable' {

    ObservableBase,
        -- superclass

    _source = false,

    ----------------------------------------------------------------------------

    __init = function(self, source)

        self._source = source

    end,

    ----------------------------------------------------------------------------

    subscribeCore = function(self, observer)

        local disposables = CompositeDisposable()

        local d = CompositeDisposable()
        disposables:add(d)

        local mergeAllObserver = MergeAllObserver(observer, disposables)
        d:add(mergeAllObserver)
        d:add(self._source:subscribe(mergeAllObserver))

        return disposables

    end,

}

--------------------------------------------------------------------------------

return function(self, ...)

    if select('#', ...) > 0 then
        error("observable:flatMergeAll() accepts no parameters. (Maybe you meant :flatMerge(n)?)")
    end

    return MergeAllObservable(self)

end
