local Rx = require 'Rx'

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

local validateNonNegativeInteger = Rx.internals.NumericArgument.validateNonNegativeInteger

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

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

    AbstractObserver,
        -- superclass

    _parent = false,
        -- (Observer)

    _sad = false,
        -- (Single Assignment Disposable)

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

    __init = function(self, parent, sad)

        self._parent = parent
        self._sad = sad

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

        if #parent._q > 0 then

            parent:handleSubscribe(table.remove(parent._q, 1))

        else

            parent._activeCount = parent._activeCount - 1

            if parent._done and parent._activeCount == 0 then
                parent._observer:onCompleted()
            end

        end

    end,

}

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _maxConcurrent = false,
        -- (number)

    _disposableList = false,
        -- (CompositeDisposable)

    _done = false,
        -- (boolean)

    _q = false,
        -- (table)

    _activeCount = 0,
        -- (number)

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

    __init = function(self, observer, maxConcurrent, disposableList)

        self._observer = observer
        self._maxConcurrent = maxConcurrent
        self._disposableList = disposableList
        self._done = false
        self._q = {}
        self._activeCount = 0

    end,

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

    handleSubscribe = function(self, xs)

        local sad = SingleAssignmentDisposable()

        self._disposableList:add(sad)

        local innerObserver = InnerObserver(self, sad)
        self._disposableList:add(innerObserver)

        sad:setDisposable(xs:subscribe(innerObserver))

    end,

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

    next = function(self, innerSource)

        if self._activeCount < self._maxConcurrent then
            self._activeCount = self._activeCount + 1
            self:handleSubscribe(innerSource)
        else
            self._q[#self._q + 1] = innerSource
        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._done = true

        if self._activeCount == 0 then
            self._observer:onCompleted()
        end

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,

    _maxConcurrent = false,

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

    __init = function(self, source, maxConcurrent)

        self._source = source
        self._maxConcurrent = maxConcurrent

    end,

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

    subscribeCore = function(self, observer)

        local disposables = CompositeDisposable()

        local mergeObserver = MergeObserver(observer, self._maxConcurrent, disposables)
        disposables:add(self._source:subscribe(mergeObserver))
        disposables:add(mergeObserver)

        return disposables

    end,

}

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

return function(self, ...)

    local maxConcurrent = ...

    if maxConcurrent == nil then
        return self:flatMergeAll()
    end

    maxConcurrent = validateNonNegativeInteger('maxConcurrent', 1, 1, ...)

    return MergeObservable(self, maxConcurrent)

end
