local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local Observer = Rx.Observer
local Queue = Rx.internals.Queue
local Scheduler = Rx.Scheduler
local SerialDisposable = Rx.SerialDisposable

local unpack = table.unpack or unpack

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

local function enqueueNext(observer, ...)

    local values = { ... }
    local nValues = select('#', ...)

    return function()
        observer:onNext(unpack(values, 1, nValues))
    end

end

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

local function enqueueError(observer, ...)

    local errors = { ... }
    local nErrors = select('#', ...)

    return function()
        observer:onError(unpack(errors, 1, nErrors))
    end

end

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

local function enqueueCompleted(observer)

    return function ()
        observer:onCompleted()
    end

end

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

local scheduledObserverPrototype = {

    AbstractObserver,
        -- superclass

    _scheduler = false,
        -- (Scheduler)

    _observer = false,
        -- (Observer)

    _isAcquired = false,
        -- (boolean)

    _hasFaulted = false,
        -- (boolean)

    _queue = false,
        -- (Rx.internals.Queue)

    _disposable = false,
        -- (Rx.SerialDisposable)

}

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

function scheduledObserverPrototype:__init(scheduler, observer)

    if not Scheduler:isClassOf(scheduler) then
        error(string.format("Rx.ScheduledObserver: scheduler (%s) is not a Scheduler",
            tostring(scheduler)), 3)
    end

    if not Observer:isClassOf(observer) then
        error(string.format("Rx.ScheduledObserver: observer (%s) is not an Observer",
            tostring(observer)), 3)
    end

    self._scheduler = scheduler
    self._observer = observer
    self._isAcquired = false
    self._hasFaulted = false
    self._queue = Queue()
    self._disposable = SerialDisposable()

end

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

function scheduledObserverPrototype:__inited()

    self._action = function()

        self._disposable:dispose()

        self._disposable = nil
        self._scheduler = nil

        self._queue:removeAll()
        self._queue = nil

    end

end

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

function scheduledObserverPrototype:next(...)

    self._queue:enqueue(enqueueNext(self._observer, ...))

end

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

function scheduledObserverPrototype:error(...)

    self._queue:enqueue(enqueueError(self._observer, ...))

end

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

function scheduledObserverPrototype:completed()

    self._queue:enqueue(enqueueCompleted(self._observer))

end

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

local function scheduleMethod(state, recurseFn)

    local work

    if state:isDisposed() then
        return
    end

    if state._queue:getCount() > 0 then

        work = state._queue:dequeue()

    else

        state._isAcquired = false
        return

    end

    local status, errorMsg = pcall(work)

    if not status then

        if state._queue then
            state._queue:removeAll()
        end

        state._hasFaulted = true

        error(errorMsg)

    end

    recurseFn(state)

end

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

function scheduledObserverPrototype:ensureActive()

    local isOwner = false

    local queue = self._queue

    if (not self._hasFaulted) and queue and queue:getCount() > 0 then
        isOwner = not self._isAcquired
        self._isAcquired = true
    end

    if isOwner then

        self._disposable:setDisposable(
            self._scheduler:scheduleRecursive(self, scheduleMethod))

    end

end

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

return Rx.Class('Rx.ScheduledObserver', scheduledObserverPrototype)
