local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local AnonymousSubject = Rx.AnonymousSubject
local Disposable = Rx.Disposable
local InnerSubscription = Rx.internals.InnerSubscription
local Observable = Rx.Observable
local Observer = Rx.Observer

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

local unpack = table.unpack or unpack

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

local subjectPrototype = {

    Observable,
    Observer,
    Disposable,
        -- superclasses

    class = {},

    _isStopped = false,
        -- (boolean)

    _hasError = false,
        -- (boolean) true if an error has occurred on source

    _error = false,
        -- (any) store err in onError

    _observers = false,
        -- (table)

    _observableHasObservers = false,
        -- (Observable) signals true if Subject has Observables, false if not

    _ohoObservers = false,
        -- (array of Observers) Observers watching :observableHasObservers

}

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

local function tableShallowCopy(orig)

    local copy = {}

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

    return copy

end

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

local function notifySubjectIsDone(self, hadObservers)

    if self._ohoObservers then
        for _, ohoObserver in ipairs(self._ohoObservers) do
            if hadObservers then
                ohoObserver:onNext(false)
            end
            ohoObserver:onCompleted()
        end
    end

    self._observableHasObservers = nil
    self._ohoObservers = nil

end

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

function subjectPrototype:__init()

    self._isStopped = false
    self._observers = {}

end

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

function subjectPrototype:_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))
        return disposableEmpty()

    end

    observer:onCompleted()
    return disposableEmpty()

end

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

function subjectPrototype:dispose()

    if not self:isDisposed() then

        notifySubjectIsDone(self, #self._observers > 0)

        self._observers = {}
        self._hasError = nil
        self._error = nil
        self._isStopped = true

        self:Rx_Disposable_dispose()

    end

end

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

function subjectPrototype:hasObservers()

    return #self._observers > 0

end

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

function subjectPrototype:isStopped()

    return self._isStopped

end

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

function subjectPrototype:observableHasObservers()

    if not self._observableHasObservers then

        self._ohoObservers = {}

        self._observableHasObservers = AnonymousObservable(

            function(observer)

                self._ohoObservers[#self._ohoObservers + 1] = observer
                observer:onNext(self:hasObservers())

                return function()

                    for i = 1, #self._ohoObservers do
                        if self._ohoObservers[i] == observer then
                            table.remove(self._ohoObservers, i)
                            return
                        end
                    end

                end

            end)

    end

    return self._observableHasObservers

end

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

function subjectPrototype:onCompleted()

    checkDisposed(self)

    if not self._isStopped then

        local os = tableShallowCopy(self._observers)
        self._isStopped = true

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

        notifySubjectIsDone(self, #os > 0)
        self._observers = {}

    end

end

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

function subjectPrototype:onError(...)

    checkDisposed(self)

    if not self._isStopped then

        local os = tableShallowCopy(self._observers)

        self._isStopped = true
        self._hasError = true
        self._error = { ... }
        self._error.n = select('#', ...)

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

        notifySubjectIsDone(self, #os > 0)
        self._observers = {}

    end

end

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

function subjectPrototype:onNext(...)

    checkDisposed(self)

    if not self._isStopped then

        local os = tableShallowCopy(self._observers)

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

    end

end

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

function subjectPrototype.class:create(observer, observable)

    return AnonymousSubject(observer, observable)

end

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

return Rx.Class('Rx.Subject', subjectPrototype)
