local Rx = require 'Rx'

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

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _count = false,
        -- (number)

    _remaining = false,
        -- (number)

    __init = function(self, observer, count)

        self._observer = observer
        self._count = count
        self._remaining = count

    end,

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

    next = function(self, ...)

        local remaining = self._remaining - 1

        if remaining >= 0 then

            self._remaining = remaining

            self._observer:onNext(...)

            if remaining == 0 then
                self._observer:onCompleted()
            end

        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- Observable

    _count = false,
        -- number

    __init = function(self, source, count)

        self._source = source
        self._count = count

    end,

    subscribeCore = function(self, observer)

        local innerObserver = TakeObserver(observer, self._count)
        return CompositeDisposable(self._source:subscribe(innerObserver), innerObserver)

    end,

}

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

return function(self, count, scheduler)

    if type(count) ~= 'number' then
        error("Rx.Observable:take: count must be a number", 2)
    end

    if count < 0 or count > (2 ^ 53) then
        error("Rx.Observable:take: count must be a positive number < 2^53", 2)
    end

    scheduler = SchedulerArgument.validateWithDefault("Rx.Observable:take",
        'currentThread', scheduler)

    if count == 0 then
        return Observable.empty(scheduler)
    end

    return TakeObservable(self, count)

end
