local Rx = require 'Rx'

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

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

return function(self, other)

    return AnonymousObservable(function(observer)

        local isOpen = false

        local disposables = CompositeDisposable(self:subscribe(
            function(...)
                if isOpen then
                    observer:onNext(...)
                end
            end,
            function(...)
                return observer:onError(...)
            end,
            function()
                if isOpen then
                    observer:onCompleted()
                end
            end))

        local rightSubscription = SingleAssignmentDisposable()
        disposables:add(rightSubscription)

        rightSubscription:setDisposable(other:subscribe(
            function()
                isOpen = true
                rightSubscription:dispose()
            end,
            function(...)
                return observer:onError(...)
            end,
            function()
                rightSubscription:dispose()
            end))

        return disposables

    end, self)

end
