local Rx = require 'Rx'

local addRef = Rx.internals.addRef
local AnonymousObservable = Rx.AnonymousObservable
local RefCountDisposable = Rx.RefCountDisposable
local CompositeDisposable = Rx.CompositeDisposable
local Subject = Rx.Subject

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

return function(source, count, skip)

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

        if count < 1 or count > (2 ^ 53) then
            error("Rx.Observable:windowWithCount: count must be a positive integer < 2^53", 2)
        end

        skip = skip or count -- if skip is not provided, default is count
        if type(skip) ~= 'number' then
            error("Rx.Observable:windowWithCount: skip must be a number", 2)
        end

        if skip < 1 or skip > (2 ^ 53) then
            error("Rx.Observable:windowWithCount: skip must be a positive integer < 2^53", 2)
        end

        return AnonymousObservable(
            function(observer)
                local m = CompositeDisposable()
                local refCountDisposable = RefCountDisposable(m)
                local n = 0
                local q = {}

                local function createWindow()
                    local s = Subject()
                    table.insert(q, s)
                    m:add(s)
                    observer:onNext(addRef(s, refCountDisposable))
                end

                createWindow()

                m:add(source:subscribe(
                    function(...)
                        local len = #q
                        for i = 1, len, 1 do
                            q[i]:onNext(...)
                        end
                        local c = n - count + 1
                        if c >= 0 and c % skip == 0 then
                            local o = table.remove(q, 1)
                            o:onCompleted()
                            m:remove(o)
                        end
                        n = n + 1
                        if n % skip == 0 then
                            createWindow()
                        end
                    end,

                    function(...)
                        while #q > 0 do
                            local o = table.remove(q, 1)
                            if not o:isDisposed() then
                                o:onError(...)
                            end
                        end

                        observer:onError(...)
                    end,

                    function()
                        while #q > 0 do
                            local o = table.remove(q, 1)
                            o:onCompleted()
                        end
                        observer:onCompleted()
                    end))

            return refCountDisposable
        end, source)
end
