local Rx = require 'Rx'

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

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

local function stringIterator(source)

    local i = 0
    local limit = #source

    return function()
        i = i + 1
        if i <= limit then
            return i, string.sub(source, i, i)
        end
    end

end

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

local function lengthIterator(source)

    local i = 0
    local limit = source.length

    return function()
        i = i + 1
        if i <= limit then
            return i, source[i]
        end
    end

end

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

local function functionIteratorAdapter(sourceFn)

    local i = 0

    local function wrapResults(firstResult, ...)
        if firstResult then
            i = i + 1
            return i, firstResult, ...
        else
            return nil
        end
    end

    return function()
        return wrapResults(sourceFn())
    end

end

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

local function getIterator(source)

    local sourceType = type(source)

    if sourceType == 'string' then

        return stringIterator(source)

    elseif sourceType == 'table' then

        if rawget(source, 'length') then
            return lengthIterator(source)
        end

        if source.__ipairs then
            return source:__ipairs()
        end

        return ipairs(source)

    elseif sourceType == 'function' then

        return functionIteratorAdapter(source)

    else

        error("Rx.Observable.from: source must be table, iterator function, or string", 3)

    end

end

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

return function(source, mapFn, scheduler)

    if mapFn ~= nil and type(mapFn) ~= 'function' then
        error("Rx.Observable.from: mapFn must be nil or function", 2)
    end

    scheduler = SchedulerArgument.validateWithDefault('Rx.Observable.from',
                                                      'currentThread', scheduler)
    return AnonymousObservable(function(observer)

        local iterFn, iterState, iterIndex = getIterator(source)

        return scheduler:scheduleRecursive(function(_, recurseFn)

            local success, iterNextIndex, value = pcall(iterFn, iterState, iterIndex)

            if not success then
                observer:onError(iterNextIndex)
                    -- If we fail, this will be the error message.
                return
            end

            iterIndex = iterNextIndex

            if iterIndex == nil then
                observer:onCompleted()
                return
            end

            local result = value

            if mapFn then
                local mapSuccess, mappedResult = pcall(mapFn, value, iterIndex)
                if mapSuccess then
                    result = mappedResult
                else
                    observer:onError(mappedResult)
                    return
                end
            end

            observer:onNext(result)

            recurseFn()

        end)

    end)

end
