local Rx = require 'Rx'

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

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

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 CombineLatestObserver = Rx.Class 'Rx.internals.CombineLatestObserver' {

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _i = false,
        -- (number)

    _resultSelector = false,
        -- (function)

    _state = {},
        -- (table)

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

    __init = function(self, observer, i, resultSelector, state)

        self._observer = observer
        self._i = i
        self._resultSelector = resultSelector
        self._state = state

    end,

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

    next = function(self, x, ...)

        local i = self._i

        local state = self._state
        local numObservables = state.numObservables

        state.values[i] = x
        state.hasValue[i] = true

        if i == numObservables then
            local extraTupleValues = select('#', ...)
            for tx = 1, extraTupleValues do
                state.values[numObservables + tx] = select(tx, ...)
            end
            state.numValues = numObservables + extraTupleValues
        end

        local hasValueAll = state.hasValueAll or allTrue(state.hasValue)
        state.hasValueAll = hasValueAll

        if hasValueAll then

            local success, ex = pcall(function()
                self._observer:onNext(self._resultSelector(
                    unpack(state.values, 1, state.numValues)))
            end)

            if not success then
                self._observer:onError(ex)
                return
            end

        elseif allTrue(state.isDone, i) then

            self._observer:onCompleted()

        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._state.isDone[self._i] = true

        if allTrue(self._state.isDone) then
            self._observer:onCompleted()
        end

    end,

}

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

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

    ObservableBase,
        --superclass

    _sources = false,
        -- (table)

    _resultSelector = false,
        -- (function)

    _length = false,

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

    __init = function(self, sources, length, resultSelector)

        self._sources = sources
        self._resultSelector = resultSelector
        self._length = length

    end,

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

    subscribeCore = function(self, observer)

        local sources = self._sources
        local n = self._length
        local subscriptions = {}

        local state = {
            hasValue = { n = n },
            hasValueAll = false,
            isDone = { n = n },
            values = {},
            numObservables = n,
            numValues = 0,
        }

        for idx = 1, n do

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

            local combineLatestObserver = CombineLatestObserver(observer, idx,
                self._resultSelector, state)

            sad:setDisposable(source:subscribe(combineLatestObserver))

            subscriptions[idx] = sad
            subscriptions[n+idx] = combineLatestObserver

        end

        return CompositeDisposable(unpack(subscriptions))

    end

}

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

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

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

return function(resultSelectorOrObservable, ...)

    local firstParamType = type(resultSelectorOrObservable)

    local resultSelector
    local sources

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

    if firstParamType == 'function' then

        resultSelector = resultSelectorOrObservable
        sources = { ... }

    elseif Observable:isClassOf(resultSelectorOrObservable) then

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

    else

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

    end

    return CombineLatestObservable(sources, n, resultSelector)

end
