local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _predicateFn = false,
        -- (function)

    _source = false,
        -- (Observable)

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

    __init = function(self, observer, predicateFn, source)

        self._observer = observer
        self._predicateFn = predicateFn
        self._source = source

    end,

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

    next = function(self, ...)

        local success, shouldRunOrError = pcall(self._predicateFn, ...)

        if success then

            if shouldRunOrError then

                self._observer:onNext(...)

            end

        else

            self._observer:onError(shouldRunOrError)

        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

local FilterObservable
    -- forward declare

FilterObservable = Rx.Class 'Rx.internals.FilterObservable' {

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _predicateFn = false,
        -- (function)

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

    __init = function(self, source, predicateFn)

        self._source = source
        self._predicateFn = predicateFn

    end,

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

    internalFilter = function(self, predicateFn)

        return FilterObservable(self._source,
            function(...)
                return self._predicateFn(...) and predicateFn(...)
            end)

    end,

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

    subscribeCore = function(self, observer)
        local filterObserver = FilterObserver(observer, self._predicateFn, self)
        return CompositeDisposable(self._source:subscribe(filterObserver), filterObserver)
    end,

}

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

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

    if FilterObservable:isClassOf(self) then
        return self:internalFilter(predicateFn)
    else
        return FilterObservable(self, predicateFn)
    end

end
