local Rx = require 'Rx'

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

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

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

    AbstractObserver,
        -- superclass

    _parent = false,
        -- (SwitchObservable)

    _id = false,
        -- (number)

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

    __init = function(self, parent, id)

        self._parent = parent
        self._id = id

    end,

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

    next = function(self, ...)

        local parent = self._parent

        if parent._latest == self._id then
            parent._observer:onNext(...)
        end

    end,

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

    error = function(self, ...)

        local parent = self._parent

        if parent._latest == self._id then
            parent._observer:onError(...)
        end

    end,

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

    completed = function(self)

        local parent = self._parent

        if parent._latest == self._id then

            parent._hasLatest = false

            if self._parent._isStopped then
                parent._observer:onCompleted()
            end

        end

    end,

}

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _inner = false,
        -- (SerialDisposable)

    _latest = 0,
        -- (number)

    _hasLatest = false,
        -- (boolean)

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

    __init = function(self, observer, inner)

        self._observer = observer
        self._inner = inner
        self._latest = 0
        self._hasLatest = false

    end,

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

    next = function(self, innerSource, ...)

        if select('#', ...) > 0 then
            self._observer:onError("Rx.Observable:switch: switched values should be a single Observable only")
            return
        end

        if not Observable:isClassOf(innerSource) then
            self._observer:onError("Rx.Observable:switch: switched value must be an Observable")
            return
        end

        local d = SingleAssignmentDisposable()

        local id = self._latest + 1
        self._latest = id

        self._hasLatest = true

        self._inner:setDisposable(d)

        local innerObserver = SwitchInnerObserver(self, id)
        d:setDisposable(CompositeDisposable(innerSource:subscribe(innerObserver), innerObserver))

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        if not self._hasLatest then
            self._observer:onCompleted()
        end

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

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

    __init = function(self, source)
        self._source = source
    end,

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

    subscribeCore = function(self, observer)

        local inner = SerialDisposable()

        local switchObserver = SwitchObserver(observer, inner)
        local s = self._source:subscribe(switchObserver)

        return CompositeDisposable(s, switchObserver, inner)

    end,

}

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

return function(self)
    return SwitchObservable(self)
end
