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

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

return function(self, resultSelector, ...)

    local sources

    if type(resultSelector) == 'function' then
        sources = { ... }
    else
        sources = { resultSelector, ... }
        resultSelector = function(...) return ... end
    end

    local nSources = #sources

    if nSources < 1 then
        error('Rx.Observable:withLatestFrom: No source Observables provided', 2)
    end

    for i = 1, nSources do
        local source = sources[i]
        if not Observable:isClassOf(source) then
            error("Rx.Observable:withLatestFrom: source " .. i .. "is not an Observable", 2)
        end
    end

    return AnonymousObservable(function(observer)

        local numNeedingValue = nSources
        local hasValue = {}
        local hasValueAll = false
        local nValues
        local values = {}

        local subscriptions = {}

        for sourceIndex = 1, nSources do

            local other = sources[sourceIndex]
            local sad = SingleAssignmentDisposable()
            local isLastSource = sourceIndex == nSources

            sad:setDisposable(other:subscribe(function(...)

                local i = sourceIndex + 1
                    -- reserve first position in values array for self's value

                if isLastSource then
                    local lastN = select('#', ...)
                    nValues = nSources + lastN
                    for argN = 1, lastN do
                        values[i + argN - 1] = select(argN, ...)
                    end
                else
                    values[i] = ...
                end

                if not hasValue[i] then
                    hasValue[i] = true
                    numNeedingValue = numNeedingValue - 1
                    hasValueAll = numNeedingValue == 0
                end

            end,
            function(...)
                observer:onError(...)
            end,
            function() end))

            subscriptions[sourceIndex] = sad

        end

        local sad = SingleAssignmentDisposable()

        sad:setDisposable(self:subscribe(
            function(...)

                values[1] = ...

                if not hasValueAll then
                    return
                end

                local function tryFunc()
                    return resultSelector(unpack(values, 1, nValues))
                end

                local function forward(success, ...)
                    if success then
                        observer:onNext(...)
                    else
                        observer:onError(...)
                    end
                end

                forward(pcall(tryFunc))

            end,
            function(...)
                observer:onError(...)
            end,
            function()
                observer:onCompleted()
            end))

        subscriptions[nSources + 1] = sad

        return CompositeDisposable(unpack(subscriptions))

    end, self)

end
