local Rx = require 'Rx'

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

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

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

    ObservableBase,
        -- superclass

    _tab = false,
        -- (table)

    _keys = false,
        -- (table)

    _scheduler = false,
        -- (Scheduler)

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

    __init = function(self, tab, scheduler)

        local keys = {}
        for k in pairs(tab) do
            keys[#keys + 1] = k
        end

        self._tab = tab
        self._keys = keys
        self._scheduler = scheduler

    end,

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

    subscribeCore = function(self, observer)

        local tab = self._tab
        local keys = self._keys
        local len = #keys

        local function loopRecursive(idx, recurseFn)

            if idx <= len then

                local key = keys[idx]
                observer:onNext(key, tab[key])

                recurseFn(idx + 1)

            else

                observer:onCompleted()

            end

        end

        return self._scheduler:scheduleRecursive(1, loopRecursive)

    end,

}

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

return function(tab, scheduler)

    if type(tab) ~= 'table' then
        error("Rx.Observable.pairs tab must be a table", 2)
    end

    scheduler = SchedulerArgument.validateWithDefault("Rx.Observable.pairs",
        'currentThread', scheduler)

    return PairsObservable(tab, scheduler)

end
