local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

local unpack = table.unpack or unpack

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

local function _allTrue(t, startIndex, endIndex)

    for i = startIndex, endIndex do

        local x = t[i]
        if not x then
            return false
        end

    end

    return true

end

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

local function allTrue(t, optExceptIndex)

    local n = t.n

    if optExceptIndex then

        local result = true
        if optExceptIndex > 1 then
            result = _allTrue(t, 1, optExceptIndex - 1)
        end
        if optExceptIndex < n then
            result = result and _allTrue(t, optExceptIndex + 1, n)
        end
        return result

    else

        return _allTrue(t, 1, n)

    end

end

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

local function argumentsToResult(...)
    return ...
end

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

return function(resultSelectorOrObservable, ...)

    local firstParamType = type(resultSelectorOrObservable)

    local resultSelector
    local params

    local n = select('#', ...)

    if firstParamType == 'function' then

        resultSelector = resultSelectorOrObservable
        params = { ... }

    elseif Observable:isClassOf(resultSelectorOrObservable) then

        resultSelector = argumentsToResult
        params = { resultSelectorOrObservable, ... }
        n = n + 1

    else

        error("Rx.Observable.combineLatest: resultSelector must be a function if provided", 3)

    end

    return AnonymousObservable(function(observer)

        local hasValue = { n = n }
        local hasValueAll = false

        local isDone = { n = n }
        local values = {}
        local nValues = n

        local function nextStep(i)

            hasValue[i] = true
            hasValueAll = hasValueAll or allTrue(hasValue)

            if hasValueAll then

                local success, ex = pcall(function()
                    observer:onNext(resultSelector(unpack(values, 1, nValues)))
                end)

                if not success then
                    observer:onError(ex)
                    return
                end

            elseif allTrue(isDone, i) then

                observer:onCompleted()

            end

        end

        local function done(i)

            isDone[i] = true

            if allTrue(isDone) then
               observer:onCompleted()
            end

        end

        local subscriptions = {}
        for idx = 1, n do

            local source = params[idx]
            local sad = SingleAssignmentDisposable()

            local nextFn
            if idx < n then
                nextFn = function(x)
                    values[idx] = x
                    nextStep(idx)
                end
            else
                -- Tuple support; keep all of the last observables values
                nextFn = function(...)
                    local newNumValues = select('#', ...)
                    for i = 1, newNumValues do
                        values[idx + i - 1] = select(i, ...)
                    end
                    for i = newNumValues, nValues do
                        values[idx + i] = nil
                    end
                    nValues = n + newNumValues - 1
                    nextStep(idx)
                end
            end
            sad:setDisposable(source:subscribe(
                nextFn,
                function(...)
                    return observer:onError(...)
                end,
                function()
                    done(idx)
                end))

            subscriptions[idx] = sad

        end

        return CompositeDisposable(unpack(subscriptions))

    end)

end
