local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable

local validateNonNegativeInteger = Rx.internals.NumericArgument.validateNonNegativeInteger

local unpack = table.unpack or unpack

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

return function(self, ...)

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

    return AnonymousObservable(function(observer)

        local q = {}
        local qHead = 1

        return self:subscribe(

            function(...)

                if count == 0 then
                    -- do nothing
                elseif #q == count then

                    q[qHead] = { n = select('#', ...), ... }

                    qHead = qHead + 1
                    if qHead > count then
                        qHead = 1
                    end

                else
                    q[#q + 1] = { n = select('#', ...), ... }
                end

            end,
            function(...)
                return observer:onError(...)
            end,
            function()

                for i = qHead, #q do

                    local values = q[i]
                    observer:onNext(unpack(values, 1, values.n))

                end

                if qHead > 1 then

                    for i = 1, qHead-1 do

                        local values = q[i]
                        observer:onNext(unpack(values, 1, values.n))

                    end

                end

                q = {}

                observer:onCompleted()

            end)

        end, self)

    end
