local Rx = require 'Rx'

local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase
local SerialDisposable = Rx.SerialDisposable

local disposableCreate = Rx.Disposable.create
local immediateScheduler = Rx.Scheduler.immediate()

local function ConcatSink(sources, observer)

    return {

        _sources = sources,

        _observer = observer,

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

        run = function(self)

            local isDisposed = false
            local sources = self._sources
            local length = sources.n
            local observer = self._observer

            local subscription = SerialDisposable()

            local cancelable = immediateScheduler:scheduleRecursive(1,
                function(i, recurseFn)

                    if not isDisposed then

                        if i > length then
                            return observer:onCompleted()
                        end

                        local hasCompleted = false

                        local disposable = sources[i]:subscribe(
                            function(...)
                                observer:onNext(...)
                            end,
                            function(...)
                                observer:onError(...)
                            end,
                            function()
                                hasCompleted = true
                                recurseFn(i + 1)
                            end)

                        if hasCompleted then
                            disposable:dispose()
                        else
                            subscription:setDisposable(disposable)
                        end

                    end

                end
            )

            return CompositeDisposable(subscription, cancelable, disposableCreate(function()
                isDisposed = true
            end))

        end,

    }

end

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

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

    ObservableBase,
        -- superclass

    _sources = false,

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

    __init = function(self, ...)
        self._sources = { n = select('#', ...), ... }
    end,

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

    subscribeCore = function(self, observer)
        local sink = ConcatSink(self._sources, observer)
        return sink:run()
    end,

}

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

return function(...)

    return ConcatObservable(...)

end
