local Rx = require 'Rx'

local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable
local SerialDisposable = Rx.SerialDisposable
local Subject = Rx.Subject

local notificationCreateOnError = Rx.Notification.createOnError
local notificationCreateOnNext = Rx.Notification.createOnNext

local unpack = table.unpack or unpack

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

local function combineLatestSource(source, subject, resultSelector)

    return AnonymousObservable(function(o)

        local value
        local isNotPaused
        local err
        local isDone = false

        local function next(notPaused, ...)

            if notPaused == nil then
                value = { n = select('#', ... ), ... }
            else
                isNotPaused = notPaused
            end

            if value ~= nil and isNotPaused ~= nil then

                o:onNext(resultSelector(value, isNotPaused))
                   -- err is only set when paused so send onNext first to unpause.

                if err then
                    o:onError(unpack(err, 1, err.n))
                    return
                end

            end

            if isDone and isNotPaused then
                o:onCompleted()
            end

        end

        return CompositeDisposable(
            source:subscribe(
                function(...)

                    next(nil, ...)

                end,
                function(...)

                    if isNotPaused then
                        o:onError(...)
                    else
                        err = { n = select('#', ...), ... }
                    end

                end,
                function()

                    isDone = true

                    if isNotPaused then
                        o:onCompleted()
                    end

                end),
            subject:subscribe(
                function(x)

                    next(not not x)

                end,
                function(...)

                    if isNotPaused then
                        o:onError(...)
                    else
                        err = { n = select('#', ...), ... }
                        next(true)
                            -- Pauser error, unpause and send queue.
                    end

                end,
                function()

                    isDone = true
                    if not isNotPaused then
                        next(true)
                            -- Pauser completed, unpause and send queue.
                    end

                end)
            )

    end, source)

end

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

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

    Observable,
        -- superclass

    _source = false,
        -- (Observable)

    _controller = false,
        -- (Subject)

    _pauser = false,
        -- (Observable)

    __init = function(self, source, pauser)

        self._source = source
        self._controller = Subject()

        if pauser and pauser.subscribe then
            self._pauser = self._controller:merge(pauser)
        else
            self._pauser = self._controller
        end

    end,

    _subscribe = function(self, observer)

        local q = {}
        local shouldFire
        local previousShouldFire

        local scheduledItem = SerialDisposable()

        local function sendQueuedValues()

            local cts = Rx.Scheduler.currentThread()

            scheduledItem:setDisposable( cts:scheduleRecursive(function(_, recurseFn)

                if shouldFire and #q > 0 then
                    local first = table.remove( q, 1 )
                    first:accept(observer)
                    recurseFn()
                end

            end ) )

        end

        local subscription =
            combineLatestSource(
                self._source,
                self._pauser:startWith(false):distinctUntilChanged(),
                function(data, newShouldFire)
                    shouldFire = newShouldFire
                    return data, newShouldFire
                end
            )
            :catch(function(...)

                if #q > 0 then

                    q[#q + 1] = notificationCreateOnError(...)

                    return self._pauser:distinctUntilChanged()
                        :map(function(newShouldFire)
                            shouldFire = newShouldFire
                            return nil, newShouldFire
                                -- Passing nil for data as it won't be used.
                        end)

                else

                    return Observable.throw(...)

                end

            end)
            :subscribe(
                function(data, shouldFire)

                    if previousShouldFire ~= nil and shouldFire ~= previousShouldFire then

                        previousShouldFire = shouldFire
                            -- Change in shouldFire.

                        if shouldFire then
                            sendQueuedValues()
                        end

                    else

                        previousShouldFire = shouldFire
                            -- shouldFire didn't change, (but may need to be initialized)
                            -- send or queue data.

                        if shouldFire then
                            observer:onNext(unpack(data, 1, data.n))
                        else
                            q[#q + 1] = notificationCreateOnNext(unpack(data, 1, data.n))
                        end

                    end

                end,
                function(...)

                    observer:onError(...)
                        -- Don't need to send queued values here because error will be
                        -- queued if it occurred while the queue contained values.

                end,
                function()

                    observer:onCompleted()

                end
            )

        return CompositeDisposable(subscription, self._controller, scheduledItem)

    end,

    pause = function(self)
        self._controller:onNext(false)
    end,

    resume = function(self)
        self._controller:onNext(true)
    end

}

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

return function(self, pauser)
    return PausableBufferedObservable(self, pauser)
end
