local Rx = require 'Rx'

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

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _takeWhileObservable = false,
        -- (TakeWhileObservable)

    _running = false,
        -- (boolean)

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

    __init = function(self, observer, takeWhileObservable)

        self._observer = observer
        self._takeWhileObservable = takeWhileObservable
        self._running = true

    end,

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

    next = function(self, ...)

        if self._running then

            local success, runningOrErr = pcall(self._takeWhileObservable._predicateFn, ...)

            if success then

                self._running = runningOrErr

                if self._running then
                    self._observer:onNext(...)
                else
                    self._observer:onCompleted()
                end

            else

                self._observer:onError(runningOrErr)

            end

        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _predicateFn = false,
        -- (function)

    __init = function(self, source, predicateFn)

        self._source = source
        self._predicateFn = predicateFn

    end,

    subscribeCore = function(self, observer)

        local takeWhileObserver = TakeWhileObserver(observer, self)

        return CompositeDisposable(self._source:subscribe(takeWhileObserver),
            takeWhileObserver)

    end,

}

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

return function(self, predicateFn, thisArg)

    if type(predicateFn) ~= 'function' then
        error("Rx.Observable:takeWhile: predicateFn must be a function", 2)
    end

    if thisArg then
        error("Rx.Observable:takeWhile: thisArg not supported", 2)
    end

    return TakeWhileObservable(self, predicateFn)

end
