local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local Observable = Rx.Observable
local SchedulerArgument = Rx.internals.SchedulerArgument

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

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 AnonymousObservable(function(observer)

        local remaining = count

        return self:subscribe(
            function(...)
                remaining = remaining - 1
                if remaining >= 0 then
                    observer:onNext(...)
                end
                if remaining <= 0 then
                    observer:onCompleted()
                end
            end,
            function(...)
                observer:onError(...)
            end,
            function()
                observer:onCompleted()
            end)

    end, self)

end
