local Rx = require 'Rx'

local ObservableBase = Rx.ObservableBase
local SchedulerArgument = Rx.internals.SchedulerArgument

local unpack = table.unpack or unpack

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

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

    ObservableBase,
        -- superclass

    _error = false,
        -- (table with n = length)

    _scheduler = false,
        -- (Scheduler)

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

    __init = function(self, error, scheduler)
        self._error = error
        self._scheduler = scheduler
    end,

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

    subscribeCore = function(self, observer)

        local function scheduleItem(scheduler, error)
            observer:onError(unpack(error, 1, error.n))
        end

        return self._scheduler:schedule(self._error, scheduleItem)

    end,

}

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

return function(...)

    local lastArgIsScheduler, scheduler = SchedulerArgument.fromVarArgs(
        'Rx.Observable.throw', 'immediate', ...)

    local numArgs = select('#', ...)
    if lastArgIsScheduler then
        numArgs = numArgs - 1
    end

    local error = { n = numArgs, ... }

    return ThrowObservable(error, scheduler)

end
