local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable

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

return function(self, predicateFn, thisArg)

    if type(predicateFn) ~= 'function' then
        error("Rx.Observable:where: predicateFn must be a function", 2)
    end

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

    return AnonymousObservable(function(observer)

        return self:subscribe(
            function(...)

                local success, shouldRunOrError = pcall(predicateFn, ...)

                if success then
                    if shouldRunOrError then
                        observer:onNext(...)
                    end
                else
                    observer:onError(shouldRunOrError)
                end

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

    end, self)

end
