local Rx = require 'Rx'

local ObservableBase = Rx.ObservableBase
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

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

local function scheduleMethod(state, recurseFn)

    local observer = state.observer

    local success, iterIndexOrErr, value = pcall(state.iterFn, state.iterState, state.iterIndex)

    if not success then
        observer:onError(iterIndexOrErr)
        return
    end

    state.iterIndex = iterIndexOrErr

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

    local result = value

    local mapFn = state.mapFn

    if mapFn then

        local mapSuccess, mappedResult = pcall(mapFn, value, state.iterIndex)

        if mapSuccess then
            result = mappedResult
        else
            observer:onError(mappedResult)
            return
        end

    end

    observer:onNext(result)

    recurseFn(state)

end

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

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

    ObservableBase,
        -- superclass

    _iterable = false,
        -- (table or string)

    _mapFn = false,
        -- (function)

    _scheduler = false,
        -- (Scheduler)

    __init = function(self, iterable, mapFn, scheduler)
        self._iterable = iterable
        self._mapFn = mapFn
        self._scheduler = scheduler
    end,

    subscribeCore = function(self, observer)

        local iterFn, iterState, iterIndex = getIterator(self._iterable)

        local state = {
            observer = observer,
            mapFn = self._mapFn,
            iterFn = iterFn,
            iterState = iterState,
            iterIndex = iterIndex,
        }

        return self._scheduler:scheduleRecursive(state, scheduleMethod)

    end,

}

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

return function(iterable, 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 FromObservable(iterable, mapFn, scheduler)

end
