local Rx = require 'Rx'

local Subject = Rx.Subject
local InnerSubscription = Rx.internals.InnerSubscription

local checkDisposed = Rx.Disposable.checkDisposed
local disposableEmpty = Rx.Disposable.empty

local unpack = table.unpack or unpack

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

local asyncSubjectPrototype = {

    Subject,
        -- superclasses

    _hasValue = false,
        -- (boolean) true if value is set

    _value = false,
        -- (any)

}

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

local function tableShallowCopy(orig)

    local copy = {}

    for orig_key, orig_value in pairs(orig) do
        copy[orig_key] = orig_value
    end

    return copy

end

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

function asyncSubjectPrototype:__init()

    self._hasValue = false

end

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

function asyncSubjectPrototype:_subscribe(observer)

    checkDisposed(self)

    if not self._isStopped then

        if self._ohoObservers and #self._observers == 0 then
            for _, ohoObserver in ipairs(self._ohoObservers) do
                ohoObserver:onNext(true)
            end
        end

        table.insert(self._observers, observer)
        return InnerSubscription(self, observer)

    end

    if self._hasError then

        observer:onError(unpack(self._error, 1, self._error.n))

    elseif self._hasValue then

        observer:onNext(unpack(self._value, 1, self._value.n))
        observer:onCompleted()

    else

        observer:onCompleted()

    end

    return disposableEmpty()

end

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

function asyncSubjectPrototype:onCompleted()

    checkDisposed(self)

    if not self._isStopped then

        self._isStopped = true

        local os = tableShallowCopy(self._observers)

        if self._hasValue then

            for i = 1, #os do

                local o = os[i]

                o:onNext(unpack(self._value, 1, self._value.n))

                o:onCompleted()

            end

        else

            for i = 1, #os do
                os[i]:onCompleted()
            end

        end

        self._observers = {}

        if self._ohoObservers then

            for _, ohoObserver in ipairs(self._ohoObservers) do

                if #os > 0 then
                    ohoObserver:onNext(false)
                end

                ohoObserver:onCompleted()

            end

            self._ohoObservers = {}

        end

    end

end

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

function asyncSubjectPrototype:onNext(...)

    checkDisposed(self)

    if not self._isStopped then

        self._value = { n = select('#', ...), ... }
        self._hasValue = true

    end

end

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

function asyncSubjectPrototype:dispose()

    self._hasValue = nil
    self._value = nil

    self:Rx_Subject_dispose()

end

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

return Rx.Class('Rx.AsyncSubject', asyncSubjectPrototype)
