local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase
local SchedulerArgument = Rx.internals.SchedulerArgument
local SerialDisposable = Rx.SerialDisposable
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

local unpack = table.unpack or unpack

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _source = false,
        -- (Observable)

    _dueTime = 0,
        -- (number)

    _scheduler = false,
        -- (Scheduler)

    _cancelable = false,
        -- (Disposable)

    _value = false,
        -- (tuple)

    _hasValue = false,
        -- (boolean)

    _id = 0,
        -- (number)

    __init = function(self, observer, source, dueTime, scheduler, cancelable)

        self._observer = observer
        self._source = source
        self._dueTime = dueTime
        self._scheduler = scheduler
        self._cancelable = cancelable
        self._hasValue = false
        self._value = nil
        self._id = 0

    end,

    next = function(self, ...)

        self._hasValue = true
        self._value = { n = select('#', ...), ... }

        local currentId = self._id

        local d = SingleAssignmentDisposable()
        self._cancelable:setDisposable(d)

        d:setDisposable(self._scheduler:scheduleFuture(self, self._dueTime,
            function(_, self)

                if self._hasValue and self._id == currentId then
                    self._observer:onNext(unpack(self._value, 1, self._value.n))
                end

                self._hasValue = false
                self._value = nil

            end))

    end,

    error = function(self, ...)

        self._cancelable:dispose()
        self._observer:onError(...)
        self._hasValue = false
        self._value = nil
        self._id = self._id + 1

    end,

    completed = function(self)

        self._cancelable:dispose()

        if self._hasValue then

            self._observer:onNext(unpack(self._value, 1, self._value.n))
            self._hasValue = false
            self._value = nil

        end

        self._observer:onCompleted()

        self._id = self._id + 1

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _dueTime = 0,
        -- (number)

    _scheduler = false,
        -- (Scheduler)

    __init = function(self, source, dueTime, scheduler)

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

        self._source = source
        self._dueTime = dueTime
        self._scheduler = scheduler

    end,

    subscribeCore = function(self, observer)

        local cancelable = SerialDisposable()

        local debounceObserver = DebounceObserver(observer, self._source,
            self._dueTime, self._scheduler, cancelable)

        return CompositeDisposable(
            self._source:subscribe(debounceObserver),
            debounceObserver,
            cancelable)

    end,

}

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

local function debounceWithSelector(source, durationSelector)

    return AnonymousObservable(function(observer)

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

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

                local success, throttle = pcall(durationSelector, ...)

                if not success then
                    return observer:onError(throttle)
                end

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

                id = id + 1
                local currentId = id

                local d = SingleAssignmentDisposable()
                cancelable:setDisposable(d)

                local function sendValue()

                    if hasValue and id == currentId then
                        observer:onNext(unpack(value, 1, value.n))
                    end

                    hasValue = false
                    value = nil

                    d:dispose()

                end

                local function sendError(...)
                    observer:onError(...)
                end

                d:setDisposable(
                    throttle:subscribe(sendValue, sendError, sendValue))

            end,
            function(...)

                cancelable:dispose()

                observer:onError(...)

                hasValue = false
                value = nil
                id = id + 1

            end,
            function()

                cancelable:dispose()

                if hasValue then
                    observer:onNext(unpack(value, 1, value.n))
                    hasValue = false
                    value = nil
                end

                observer:onCompleted()
                id = id + 1

            end
        )

        return CompositeDisposable(subscription, cancelable)

    end, source)

end

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

return function(self, dueTimeOrSelector, scheduler)

    local dueTimeOrSelectorType = type(dueTimeOrSelector)

    if dueTimeOrSelectorType == 'function' then
        return debounceWithSelector(self, dueTimeOrSelector)
    elseif dueTimeOrSelectorType == 'number' then
        return DebounceObservable(self, dueTimeOrSelector, scheduler)
    else
        error("Rx.Observable:debounce: invalid arguments")
    end

end
