local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase
local Scheduler = Rx.Scheduler
local SerialDisposable = Rx.SerialDisposable
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

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

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

    AbstractObserver,
        -- superclass

    _state = false,
        -- (table)

    _recurseFn = false,
        -- (function)

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

    __init = function(self, state, recurseFn)
        self._state = state
        self._recurseFn = recurseFn
    end,

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

    next = function(self, ...)
        self._state.observer:onNext(...)
    end,

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

    error = function(self)
        self._recurseFn(self._state)
    end,

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

    completed = function(self)
        self._recurseFn(self._state)
    end,

}

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

local function scheduleMethod(state, recurseFn)

    if state.pos < #state.sources then

        local pos = state.pos + 1
        state.pos = pos
        local current = state.sources[pos]

        local d = SingleAssignmentDisposable()
        state.subscription:setDisposable(d)

        local resumeNextObserver = OnErrorResumeNextObserver(state, recurseFn)

        d:setDisposable(CompositeDisposable(
            current:subscribe(resumeNextObserver),
            resumeNextObserver))

    else

        state.observer:onCompleted()

    end

end

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

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

    ObservableBase,
        -- superclass

    _sources = false,
        -- (Observable)

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

    __init = function(self, sources)

        self._sources = sources

    end,

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

    subscribeCore = function(self, observer)

        local subscription = SerialDisposable()

        local state = {
            pos = 0,
            subscription = subscription,
            observer = observer,
            sources = self._sources,
        }

        local cancellable = Scheduler.immediate():scheduleRecursive(state,
            scheduleMethod)

        return CompositeDisposable(subscription, cancellable)

    end,

}

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

return function(...)

    local sources = { ... }

    return OnErrorResumeNextObservable(sources)

end
