local Rx = require 'Rx'

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

local unpack = table.unpack or unpack

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _accumulatorFn = false,
        -- (function)

    _seeds = false,
        -- (table)

    _accumulation = false,
        -- (table)

    _isStopped = false,
        -- (boolean)

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

    __init = function(self, observer, parent)

        self._observer = observer
        self._accumulatorFn = parent._accumulatorFn
        self._seeds = parent._seeds
        self._accumulation = nil
        self._isStopped = false

    end,

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

    next = function(self, ...)

        local function pcallPack(success, ...)
            return success, { length = select('#', ...), ... }
        end

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

        local accumulation = self._accumulation
        local seeds = self._seeds

        local accumParams

        if accumulation ~= nil then

            accumParams = accumulation

        else

            local numSeeds = seeds.length
            if numSeeds == 0 then
                accumParams = { length = 1, nil }
            else
                accumParams = { length = numSeeds }
                for i = 1, numSeeds do
                    accumParams[i] = seeds[i]
                end
            end

        end

        local values = { ... }
        local numValues = select('#', ...)

        local numAccumParams = accumParams.length
        for i = 1, numValues do
            accumParams[numAccumParams + i] = values[i]
        end
        numAccumParams = numAccumParams + numValues

        local success

        success, accumulation = pcallPack(pcall(self._accumulatorFn,
            unpack(accumParams, 1, numAccumParams)))

        if success then

            if accumulation.length > 1 and seeds.length == 0 then

                self._observer:onError("Rx.Observable:scan: must specify seed values when returning multiple values from accumulator function")

            else

                if accumulation.length == 0 or seeds.length > 0 and accumulation ~= seeds.length then
                    accumulation.length = math.max(seeds.length, 1)
                end

                self._accumulation = accumulation

                self._observer:onNext(unpack(accumulation, 1,
                    accumulation.length))

            end

        else

            local err = accumulation[1]
            self._observer:onError(err)

        end

    end,

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

    error = function(self, ...)

        self._observer:onError(...)

    end,

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

    completed = function(self)

        local seeds = self._seeds
        if self._accumulation == nil and #seeds > 0 then
            self._observer:onNext(unpack(seeds, 1, seeds.length))
        end

        self._observer:onCompleted()

    end,

}

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

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

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _accumulatorFn = false,
        -- (function)

    _seeds = false,
        -- (table)

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

    __init = function(self, source, accumulatorFn, seeds)

        self._source = source
        self._accumulatorFn = accumulatorFn
        self._seeds = seeds

    end,

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

    subscribeCore = function(self, observer)

        local scanObserver = ScanObserver(observer, self)
        return CompositeDisposable(
            self._source:subscribe(scanObserver),
            scanObserver)

    end,

}

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

return function(self, accumulatorFn, ...)

    if type(accumulatorFn) ~= 'function' then
        error("Rx.Observable:scan: accumulator must be a function", 2)
    end

    local seeds = { length = select('#', ...), ... }

    return ScanObservable(self, accumulatorFn, seeds)

end
