local Rx = require 'Rx'

local Disposable = Rx.Disposable
local Observer = Rx.Observer
local ScheduledObserver = Rx.ScheduledObserver
local Scheduler = Rx.Scheduler

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

local observeOnObserverPrototype = {

    ScheduledObserver,
        -- superclass

    _cancel = false,
        -- (Disposable)

}

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

function observeOnObserverPrototype:__init(scheduler, observer, cancel)

    if cancel and not Disposable:isClassOf(cancel) then
        error(string.format("Rx.ObserveOnObserver: cancel (%d) is not a Disposable",
            tostring(cancel)), 3)
    end

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

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

    self._cancel = cancel

    return scheduler, observer

end

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

function observeOnObserverPrototype:next(...)

    self:Rx_ScheduledObserver_next(...)

    self:ensureActive()

end

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

function observeOnObserverPrototype:error(...)

    self:Rx_ScheduledObserver_error(...)

    self:ensureActive()

end

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

function observeOnObserverPrototype:completed()

    self:Rx_ScheduledObserver_completed()

    self:ensureActive()

end

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

function observeOnObserverPrototype:dispose()

    self:Rx_ScheduledObserver_dispose()

    if self._cancel then
        self._cancel:dispose()
    end

    self._cancel = nil

end

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

return Rx.Class('Rx.ObserveOnObserver', observeOnObserverPrototype)
