local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local SchedulerArgument = Rx.internals.SchedulerArgument
local SerialDisposable = Rx.SerialDisposable

local unpack = table.unpack or unpack

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

return function(self, windowDuration, scheduler)

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

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

    return AnonymousObservable(function(observer)

        local cancelable = SerialDisposable()
        local lastOnNext = 0
        local hasValue = false
        local value

        local function clearValue()
            cancelable:dispose()
            value = nil
            hasValue = false
        end

        local subscription = self:subscribe(
            function(...)

                local now = scheduler:now()

                local delta = now - lastOnNext

                if lastOnNext == 0 or delta >= windowDuration then

                    if hasValue then

                        -- Timer overrun, kill off the scheduled callback as it
                        -- would just use this item anyways.

                        cancelable:setDisposable(nil)
                        value = nil
                        hasValue = false

                    end

                    lastOnNext = now

                    observer:onNext(...)

                else

                    value = { n = select('#', ...), ... }

                    if not hasValue then

                        hasValue = true

                        local si = scheduler:scheduleFuture(windowDuration - delta,
                            function()

                                if hasValue then

                                    lastOnNext = scheduler:now()

                                    observer:onNext(unpack(value, 1, value.n))

                                    value = nil
                                    hasValue = false

                                end

                            end)

                        cancelable:setDisposable(si)

                    end

                end

            end,
            function(...)

                clearValue()
                observer:onError(...)

            end,
            function()

                if hasValue then
                    observer:onNext(unpack(value, 1, value.n))
                    clearValue()
                end

                observer:onCompleted()

            end)

        return CompositeDisposable(subscription, cancelable)

    end, self)

end
