local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local SchedulerArgument = Rx.internals.SchedulerArgument

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

return function(self, windowDuration, scheduler)

    scheduler = SchedulerArgument.validateWithoutDefault(
        'Rx.Observable:throttle', scheduler)

    if type(windowDuration) ~= 'number' or windowDuration <= 0 then
        error("Rx.Observable:throttle: windowDuration argument must be a number greater than 0", 2)
    end

    return AnonymousObservable(function(o)

        local lastOnNext = 0

        return self:subscribe(
            function(...)

                local now = scheduler:now()

                if lastOnNext == 0 or now - lastOnNext >= windowDuration then

                    lastOnNext = now

                    o:onNext(...)

                end

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

    end, self)

end
