local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable

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

local function noop() end

return function(self, other)

    if not Observable:isClassOf(other) then
        error("Rx.Observable:takeUntil: other value must be an Observable")
    end

    return AnonymousObservable(function(observer)

        return CompositeDisposable(
            self:subscribe(observer),
            other:subscribe(
                function(...)
                    return observer:onCompleted()
                end,
                function(...)
                    return observer:onError(...)
                end,
                noop)
            )

    end, self)

end
