local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable

local unpack = table.unpack or unpack

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

return function(self, ...)

    local selector = (...)
    local selectorFn

    if type(selector) == 'function' then

        selectorFn = selector

        local thisArg = select(2, ...)

        if thisArg then
            error("Rx.Observable:select: thisArg not supported", 2)
        end

    else

        local values = { ... }
        local nValues = select('#', ...)

        selectorFn = function() return unpack(values, 1, nValues) end

    end

    return AnonymousObservable(function(observer)

        return self:subscribe(
            function(...)

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

                handlePredicateResult(pcall(selectorFn, ...))

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

    end, self)

end
