-- Gets a scheduler that schedules work as soon as possible on the current thread.
-- Note that this is a singleton.

local Rx = require 'Rx'

local PriorityQueue = Rx.internals.PriorityQueue
local Scheduler = Rx.Scheduler
local ScheduledItem = Rx.internals.ScheduledItem

local queue

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

local function runTrampoline()

    while queue:length() > 0 do

        local item = queue:dequeue()

        if not item:isCancelled() then
            item:invoke()
        end

    end

end

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

local currentThreadSchedulerClass = Rx.Class('Rx.CurrentThreadScheduler', {

    Scheduler,
        -- superclass

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

    schedule = function(self, state, action)

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

        local si = ScheduledItem(self, state, action, self:now())

        if not queue then

            queue = PriorityQueue()
            queue:enqueue(si)

            local success, message = pcall(runTrampoline)

            queue = nil

            if not success then
                error(message)
            end

        else

            queue:enqueue(si)

        end

        return si._disposable

    end,

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

    scheduleRequired = function(self)

        return not queue

    end,

})

local currentThreadSchedulerInstance = currentThreadSchedulerClass()

return currentThreadSchedulerInstance
