local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local SchedulerArgument = Rx.internals.SchedulerArgument

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

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 AnonymousObservable(function(observer)

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

        return scheduler:scheduleRecursive(0, function(idx, recurseFn)

            if idx < #keys then

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

                recurseFn(idx + 1)

            else
                observer:onCompleted()
            end

        end)

    end)

end
