local Rx = require 'Rx'

local PriorityQueue = Rx.internals.PriorityQueue
local ScheduledItem = Rx.internals.ScheduledItem
local SchedulePeriodicRecursive = Rx.internals.SchedulePeriodicRecursive

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

local virtualSchedulerPrototype = {

    Rx.Scheduler,

    _clock = false,
        -- (any, typically number) current virtual clock reading

    _isEnabled = false,
        -- (boolean) true if scheduler is running

    _queue = false,
        -- (Rx.PriorityQueue)

}

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

local function validateTimeArgument(time, stackFrameDepth)

    if type(time) ~= 'number' then
        error("VirtualTimeScheduler: time argument must be a number", stackFrameDepth)
    end

end

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

function virtualSchedulerPrototype:__init(initialClock)

    if initialClock == nil then
        initialClock = 0
    end

    validateTimeArgument(initialClock, 4)

    self._clock = initialClock

    self._isEnabled = false

    self._queue = PriorityQueue()

end

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

function virtualSchedulerPrototype:clock()

    return self._clock

end

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

function virtualSchedulerPrototype:now()

    return self._clock

end

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

function virtualSchedulerPrototype:schedule(state, action)

    if action == nil then
        action = state
        state = nil
    end

    return self:scheduleAbsolute(state, self._clock, action)

end

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

function virtualSchedulerPrototype:scheduleFuture(state, dueTime, action)

    if action == nil then
        action = dueTime
        dueTime = state
        state = nil
    end

    validateTimeArgument(dueTime)

    return self:scheduleRelative(state, dueTime, action)

end

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

function virtualSchedulerPrototype:scheduleRelative(state, dueTime, action)

    if action == nil then
        action = dueTime
        dueTime = state
        state = nil
    end

    if type(action) ~= 'function' then
        error("VirtualTimeScheduler:scheduleRelative: action argument must be a function", 2)
    end

    validateTimeArgument(dueTime, 3)

    local runAt = self._clock + dueTime
    return self:scheduleAbsolute(state, runAt, action)

end

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

function virtualSchedulerPrototype:start()

    if not self._isEnabled then

        self._isEnabled = true

        repeat

            local next = self:getNext()

            if next ~= nil then

                if next._dueTime and self._clock and next._dueTime > self._clock then
                    self._clock = next._dueTime
                end

                next:invoke()

            else

                self._isEnabled = false

            end

        until not self._isEnabled

    end

end

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

function virtualSchedulerPrototype:stop()

    self._isEnabled = false

end

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

function virtualSchedulerPrototype:advanceTo(time)

    validateTimeArgument(time, 3)

    if time < self._clock then
        error("Rx.VirtualSchedulerPrototype:advanceTo: proposed new time is behind current time", 2)
    end

    if time == self._clock then
        return
    end

    if not self._isEnabled then

        self._isEnabled = true

        repeat

            local next = self:getNext()

            if next ~= nil and next._dueTime <= time then

                if next._dueTime > self._clock then
                    self._clock = next._dueTime
                end

                next:invoke()

            else

                self._isEnabled = false

            end

        until not self._isEnabled

        self._clock = time

    end

end

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

function virtualSchedulerPrototype:advanceBy(time)

    validateTimeArgument(time, 3)

    local dt = self._clock + time

    if self._clock > dt then
        error("Rx.VirtualSchedulerPrototype:advanceBy: can not roll back time", 2)
    end

    if self._clock == dt then
        return
    end

    self:advanceTo(dt)

end

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

function virtualSchedulerPrototype:sleep(time)

    validateTimeArgument(time, 3)

    local dt = self._clock + time

    if dt <= self._clock then
        error("Rx.VirtualSchedulerPrototype:sleep: sleep must advance clock", 2)
    end

    self._clock = dt

end

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

function virtualSchedulerPrototype:getNext()

    local queue = self._queue

    while queue:length() > 0 do

        local next = queue:peek()

        if next:isCancelled() then
            queue:dequeue()
        else
            return next
        end

    end

    return nil

end

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

function virtualSchedulerPrototype:scheduleAbsolute(state, dueTime, action)

    if action == nil then
        action = dueTime
        dueTime = state
        state = nil
    end

    if type(action) ~= 'function' then
        error("VirtualTimeScheduler:scheduleAbsolute: action argument must be a function", 2)
    end

    validateTimeArgument(dueTime, 3)

    local si

    local function run(scheduler, state1)
        self._queue:remove(si)
        return action(scheduler, state1)
    end

    si = ScheduledItem(self, state, run, dueTime)
    self._queue:enqueue(si)

    return si._disposable

end

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

function virtualSchedulerPrototype:schedulePeriodic(state, period, action)

    if action == nil then
        action = period
        period = state
        state = nil
    end

    local s = SchedulePeriodicRecursive(self, state, period, action)
    return s:start()

end

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

return Rx.Class('Rx.VirtualTimeScheduler', virtualSchedulerPrototype)
