local Rx = require 'Rx'

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

local observableOf = Rx.Observable.of

local unpack = table.unpack or unpack

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

local function setCompletion(observer, errors)

    if #errors == 0 then

        observer:onCompleted()

    elseif #errors == 1 then

        local error = errors[1]
        observer:onError(unpack(error, 1, error.n))

    else

        observer:onError("multiple errors occurred", errors)

    end

end

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

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

    AbstractObserver,
        -- superclass

    _inner = false,
        -- (Disposable)

    _group = false,
        -- (CompositeDisposable)

    _state = false,
        -- (table)

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

    __init = function(self, inner, group, state)

        self._inner = inner
        self._group = group
        self._state = state

    end,

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

    next = function(self, ...)

        self._state.observer:onNext(...)

    end,

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

    error = function(self, ...)

        local errors = self._state.errors
        errors[#errors + 1] = { n = select('#', ...), ... }

        self._group:remove(self._inner)

        if self._state.isStopped and self._group:length() == 1 then
            setCompletion(self._state.observer, self._state.errors)
        end

    end,

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

    completed = function(self)

        self._group:remove(self._inner)

        if self._state.isStopped and self._group:length() == 1 then
            setCompletion(self._state.observer, self._state.errors)
        end

    end,

}

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

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

    AbstractObserver,
        -- superclass

    _group = false,
        -- (Disposable)

    _state = false,
        -- (table)

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

    __init = function(self, group, state)

        self._group = group
        self._state = state

    end,

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

    next = function(self, x)

        local inner = SingleAssignmentDisposable()

        local group = self._group

        group:add(inner)

        local observer = MergeDelayErrorInnerObserver(inner, group, self._state)

        inner:setDisposable(CompositeDisposable(x:subscribe(observer), observer))

    end,

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

    error = function(self, ...) error("observableOf sent error, fix code in mergeDelayError") end,
    --[[ Not possible to reach the above line because observableOf doesn't send errors.

        local errors = self._state.errors
        errors[#errors + 1] = { n = select('#', ...), ... }
        self._state.isStopped = true

        if self._group:length() == 1 then
            setCompletion(self._state.observer, self._state.errors)
        end

    end, ]]--

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

    completed = function(self)

        self._state.isStopped = true

        if self._group:length() == 1 then
            setCompletion(self._state.observer, self._state.errors)
        end

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

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

    __init = function(self, source)

        self._source = source

    end,

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

    subscribeCore = function(self, observer)

        local group = CompositeDisposable()

        local m = SingleAssignmentDisposable()

        local state = {
            isStopped = false,
            errors = {},
            observer = observer,
        }

        group:add(m)

        local mergeDelayObserver = MergeDelayErrorObserver(group, state)

        m:setDisposable(CompositeDisposable(
            self._source:subscribe(mergeDelayObserver),
            mergeDelayObserver))

        return group

    end

}

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

return function(...)

    local source = observableOf(...)

    return MergeDelayErrorObservable(source)

end
