local Rx = require 'Rx'

local SchedulerArgument = Rx.internals.SchedulerArgument
local ScheduledObserver = Rx.ScheduledObserver
local Subject = Rx.Subject

local checkDisposed = Rx.Disposable.checkDisposed
local disposableCreate = Rx.Disposable.create

local unpack = table.unpack or unpack

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

local replaySubjectPrototype = {

    Subject,
        -- superclass

    _bufferSize = false,
        -- maximum element count of the replay buffer

    _windowSize = false,
        -- maximum time length of the replay buffer

    _scheduler = false,
        -- scheduler the observers are invoked on.

    _q = false,
        -- queued up values to replay

    _qHead = false,
        -- index of where to enqueue next value at

    _qTail = false,
        -- index to dequeue next value from or nil if empty

}

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

local function tableShallowCopy(orig)

    local copy = {}

    for orig_key, orig_value in pairs(orig) do
        copy[orig_key] = orig_value
    end

    return copy

end

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

local function findObserver(observers, observer)

    for i, next in ipairs(observers) do

        if next == observer then
            return i
        end

    end

end

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

local function sendQueuedValues(self, so)

    if self._qTail then

        if self._qHead >= self._qTail then

            for i = self._qTail, self._qHead do
                local value = self._q[i].value
                so:onNext(unpack(value, 1, value.n))
            end

        elseif self._qHead < self._qTail then

            for i = self._qTail, #self._q do
                local value = self._q[i].value
                so:onNext(unpack(value, 1, value.n))
            end

            for i = 1, self._qHead do
                local value = self._q[i].value
                so:onNext(unpack(value, 1, value.n))
            end

        end

    end

end

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

local function subjectIsDone(subject, hadObservers, isDone)

    if subject._ohoObservers then

        for _, ohoObserver in ipairs(subject._ohoObservers) do

            if hadObservers then
                ohoObserver:onNext(false)
            end

            if isDone then
                ohoObserver:onCompleted()
            end

        end

    end

    if isDone then
        subject._ohoObservers = {}
    end

end

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

local function createRemovableDisposable(subject, observer)

    return disposableCreate(function()

        observer:dispose()

        if not subject._isDisposed and subject._observers then

            local index = findObserver(subject._observers, observer)
            if index then
                table.remove(subject._observers, index)
            end

            if (not subject._isStopped) and index and #subject._observers == 0 then
                subjectIsDone(subject, true, false)
            end

        end

    end)

end

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

function replaySubjectPrototype:__init(bufferSize, windowSize, scheduler)

    -- in case you need to create an unbounded buffer, use math.huge as the bufferSize

    assert( type( bufferSize ) == 'number', 'ReplaySubject: bufferSize is invalid number' )

    self._bufferSize = bufferSize
    self._windowSize = windowSize
    self._q = {}

    self._scheduler = SchedulerArgument.validateWithDefault(
        'Rx.ReplaySubject', 'currentThread', scheduler)

end

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

function replaySubjectPrototype:_incQueueIndex(index)

    if index then

        index = index + 1

        if self._bufferSize and index > self._bufferSize then
            index = 1
        end

    else

        index = 1

    end

    return index

end

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

function replaySubjectPrototype:_trimQueuedValues(now)

    if self._windowSize ~= nil then

        while self._qTail and (now - self._q[self._qTail].interval) > self._windowSize do

            if self._qTail == self._qHead then
                self._qTail = false
                self._qHead = false
            else
                self._qTail = self:_incQueueIndex(self._qTail)
            end
        end

    end

end

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

function replaySubjectPrototype:_enqueueValue(value, now)

    self._qHead = self:_incQueueIndex(self._qHead)

    self._q[self._qHead] = { interval = now, value = value }

    if not self._qTail then

        self._qTail = 1

    elseif self._qTail == self._qHead then

        self._qTail = self:_incQueueIndex(self._qTail)
            -- queue was full, move the tail as we overwrote a value

    end

end

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

function replaySubjectPrototype:_subscribe(observer)

    local so = ScheduledObserver(self._scheduler, observer)

    local subscription = createRemovableDisposable(self, so)

    checkDisposed(self)

    if self._ohoObservers and not self._hasError and not self._isStopped
        and #self._observers == 0
    then
        for _, ohoObserver in ipairs(self._ohoObservers) do
            ohoObserver:onNext(true)
        end
    end

    table.insert(self._observers, so)

    self:_trimQueuedValues(self._scheduler:now())
    sendQueuedValues(self, so)

    if self._hasError then
        so:onError(unpack(self._error, 1, self._error.n))
    elseif self._isStopped then
        so:onCompleted()
    end

    so:ensureActive()

    return subscription

end

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

function replaySubjectPrototype:onNext(...)

    checkDisposed(self)

    if not self._isStopped then

        local value = { n = select('#', ...), ... }

        self:_enqueueValue(value, self._scheduler:now())
        self:_trimQueuedValues(self._scheduler:now())

        local os = tableShallowCopy(self._observers)

        for i = 1, #os do

            local observer = os[i]

            observer:onNext(...)

            observer:ensureActive()

        end

    end

end

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

function replaySubjectPrototype:onError(...)

    checkDisposed(self)

    if not self._isStopped then

        self._isStopped = true

        self._error = { n = select('#', ...), ... }
        self._hasError = true

        self:_trimQueuedValues(self._scheduler:now())

        local os = tableShallowCopy(self._observers)

        for i = 1, #os do

            local observer = os[i]

            observer:onError(...)

            observer:ensureActive()

        end

        self._observers = {}

        subjectIsDone(self, #os > 0, true)

    end

end

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

function replaySubjectPrototype:onCompleted()

    checkDisposed(self)

    if not self._isStopped then

        self._isStopped = true

        self:_trimQueuedValues(self._scheduler:now())

        local os = tableShallowCopy(self._observers)

        for i = 1, #os do

            local observer = os[i]

            observer:onCompleted()

            observer:ensureActive()

        end

        self._observers = {}

        subjectIsDone(self, #os > 0, true)

    end

end

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

function replaySubjectPrototype:dispose()

    if not self:isDisposed() then

        self._q = nil
        self._scheduler = nil

        subjectIsDone(self, #self._observers > 0, true)

        self:Rx_Subject_dispose()

    end

end

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

return Rx.Class('Rx.ReplaySubject', replaySubjectPrototype)
