local Rx = require 'Rx'

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

local validateNonNegativeInteger = Rx.internals.NumericArgument.validateNonNegativeInteger

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _remaining = false,
        -- (number)

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

    __init = function(self, observer, count)

        self._remaining = count
        self._observer = observer

    end,

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

    next = function(self, ...)

        local remaining = self._remaining

        if remaining <= 0 then
            self._observer:onNext(...)
        else
            self._remaining = remaining - 1
        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

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

    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 = SkipObserver(observer, self._count)
        return CompositeDisposable(self._source:subscribe(innerObserver), innerObserver)
    end,

}

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

return function(self, ...)

    local count = validateNonNegativeInteger('count', 1, 1, ...)

    return SkipObservable(self, count)

end
