local Rx = require 'Rx'

local Observable = Rx.Observable

local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local SerialDisposable = Rx.SerialDisposable
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

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

return function(self)

    return AnonymousObservable(function(observer)

        local hasLatest = false
        local innerSubscription = SerialDisposable()
        local isStopped = false
        local latest = 0

        local subscription = self:subscribe(

            function(innerSource, ...)

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

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

                local d = SingleAssignmentDisposable()
                latest = latest + 1

                local id = latest
                hasLatest = true
                innerSubscription:setDisposable(d)

                d:setDisposable(innerSource:subscribe(
                    function(...)
                        if latest == id then
                            observer:onNext(...)
                        end
                    end,
                    function(...)
                        if latest == id then
                            observer:onError(...)
                        end
                    end,
                    function()
                        if latest == id then
                            hasLatest = false
                            if isStopped then
                                observer:onCompleted()
                            end
                        end
                    end))

            end,
            function(...)
                return observer:onError(...)
            end,
            function()
                isStopped = true
                if not hasLatest then
                    observer:onCompleted()
                end
            end)

        return CompositeDisposable(subscription, innerSubscription)

    end, self)

end
