local Rx = require 'Rx'

local ObservableBase = Rx.ObservableBase
local Scheduler = Rx.Scheduler

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

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

    _observer = false,
        --(Observer)

    _parent = false,
        --(Observable)

    __init = function(self, observer, parent)

        self._observer = observer
        self._parent = parent

    end,

    run = function(self)

        local observer = self._observer
        local array = self._parent._array
        local len = array.length

        local function loopRecursive(i, recurseFn)

            if i <= len then
                observer:onNext(array[i])
                recurseFn(i + 1)
            else
                observer:onCompleted()
            end

        end

        return self._parent._scheduler:scheduleRecursive(1, loopRecursive)

    end,

}

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

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

    ObservableBase,
        -- superclass

    _scheduler = false,
        -- (Observer)

    _array = false,

    __init = function(self, scheduler, array)
        self._scheduler = scheduler
        self._array = array
    end,

    subscribeCore = function(self, observer)
        local sink = FromArraySink(observer, self)
        return sink:run()
    end,

}

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

return function(scheduler, ...)

    if not Scheduler:isClassOf(scheduler) then
        error("Observable.ofWithScheduler requires first argument to be a scheduler", 2)
    end

    return FromArrayObservable(scheduler, { length = select('#', ...), ... })

end
