local Rx = require 'Rx'

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

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    __init = function(self, observer)
        self._observer = observer
    end,

    next = function(self, ...)
        self._observer:onCompleted()
    end,

    error = function(self, ...)
        self._observer:onError(...)
    end,

    completed = function(self)
    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _other = false,
        -- (Observable)

    __init = function(self, source, other)

        self._source = source
        self._other = other

    end,

    subscribeCore = function(self, observer)

        local innerObserver = TakeUntilObserver(observer)

        return CompositeDisposable(
            innerObserver,
            self._source:subscribe(observer),
            self._other:subscribe(innerObserver)
        )

    end,

}

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

return function(self, other)

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

    return TakeUntilObservable(self, other)

end
