local Rx = require 'Rx'

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

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

local SumObserver = Rx.Class('Rx.internals.SumObserver', {

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _transformFn = false,
        -- (function)

    _source = false,
        -- (Observable)

    _sum = false,
        -- (Number) sum of all values

    __init = function(self, observer, transformFn, source)

        self._observer = observer
        self._transformFn = transformFn
        self._source = source
        self._sum = 0

    end,

    next = function(self, x, noTuple)

        assert(noTuple == nil, "sum operator doesn't support tuples")

        if self._transformFn then

            local success, valueOrError = pcall(self._transformFn, x)
            if success then
                self._sum = self._sum + valueOrError
            else
                self._observer:onError( valueOrError )
            end

        else

            self._sum = self._sum + x

        end

    end,

    error = function(self, ...)

        self._observer:onError(...)

    end,

    completed = function(self)

        self._observer:onNext(self._sum)
        self._observer:onCompleted()

    end,

})

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

local SumObservable = Rx.Class('Rx.internals.SumObservable', {

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _transformFn = false,
        -- (function)

    __init = function(self, source, transformFn)
        self._source = source
        self._transformFn = transformFn
    end,

    subscribeCore = function(self, observer)

        local sumObserver = SumObserver(observer, self._transformFn, self)

        local subscriptionOrError = self._source:subscribe(sumObserver)

        return CompositeDisposable(subscriptionOrError, sumObserver)

    end,
})

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

return function(self, transformFn)

    if transformFn and not (type(transformFn) == 'function') then
        error("Rx.Observable:sum: transformFn is not a function", 2)
    end

    return SumObservable(self, transformFn)

end
