local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable

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

local function defaultPackedTupleComparer(prevPackedTupleN, prevPackedTuple, packedNewTupleN, packedNewTuple) -- intentionally long

    if prevPackedTupleN ~= packedNewTupleN then

        return false

    end

    for i = 1, prevPackedTupleN do

        local a, b = prevPackedTuple[i], packedNewTuple[i]

        if a == 0 and b == 0 then

            if 1/a == -1/b then

                  -- -0 does not equal 0 for our purposes

                return false

            end

        elseif not (a == b or (a ~= a and b ~= b)) then

            -- a ~= a and b ~= b implies a and b both NaN

            return false

        end

    end

    return true

end

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

local function compareTuples(comparer, maybePrevPackedTupleN, maybePrevPackedTuple, packedNewTupleN, packedNewTuple) -- intentionally long

    if maybePrevPackedTupleN == nil then

        return false

    end

    return comparer(maybePrevPackedTupleN, maybePrevPackedTuple,
                        packedNewTupleN, packedNewTuple)

end

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

local function dispatchCompareErrorsToOnError(observer, success, ...)

    if not success then

        observer:onError(...)

    end

end

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

return function(self, keySelector, comparer)

    if keySelector ~= nil and type(keySelector) ~= 'function' then
        error("Rx.Observable:distinctUntilChanged: first argument, keySelector, must be a function", 2)
    end

    if comparer ~= nil and type(comparer) ~= 'function' then
        error("Rx.Observable:distinctUntilChanged: second argument, comparer, must be nil or a function", 2)
    end

    comparer = comparer or defaultPackedTupleComparer

    return AnonymousObservable(function(observer)

        local prevPackedTupleN, prevPackedTuple

        local function try(...)

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

            local currTuple = keySelector and { keySelector(...) } or { ... }

            if not compareTuples(comparer,
                                    prevPackedTupleN, prevPackedTuple,
                                    currTupleN, currTuple) then

                prevPackedTupleN, prevPackedTuple = currTupleN, currTuple

                return observer:onNext(...)

            end

        end

        return self:subscribe(

            function(...)
                dispatchCompareErrorsToOnError(observer, pcall(try, ...))
            end,
            function(...)
                return observer:onError(...)
            end,
            function(...)
                return observer:onCompleted(...)
            end)

    end, self)

end
