local Rx = require 'Rx'

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

local unpack = table.unpack or unpack

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _selectorFn = false,
        -- (function)

    _source = false,
        -- (Observable)

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

        self._observer = observer
        self._selectorFn = selectorFn
        self._source = source

    end,

    next = function(self, ...)

        local function handlePredicateResult(success, ...)
            if success then
                self._observer:onNext(...)
            else
                self._observer:onError(...)
            end
        end

        handlePredicateResult(pcall(self._selectorFn, ...))

    end,

    error = function(self, ...)

        self._observer:onError(...)

    end,

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

local MapObservable
    -- forward declaration

MapObservable = Rx.Class 'Rx.internals.MapObservable' {

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _selectorFn = false,
        -- (function)

    __init = function(self, source, selectorFn)
        self._source = source
        self._selectorFn = selectorFn
    end,

    internalMap = function(self, selectorFn)
        return MapObservable(self._source, function(...)
            return selectorFn(self._selectorFn(...))
        end)
    end,

    subscribeCore = function(self, observer)

        local mapObserver = MapObserver(observer, self._selectorFn, self)

        local success, subscriptionOrError =
            pcall(self._source.subscribe, self._source, mapObserver)

        if success then
            return CompositeDisposable(subscriptionOrError, mapObserver)
        else
            mapObserver:dispose()
            error(subscriptionOrError)
        end

    end,

}

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

return function(self, ...)

    local selector = (...)
    local selectorFn

    if type(selector) == 'function' then

        selectorFn = selector

        local thisArg = select(2, ...)

        if thisArg then
            error("Rx.Observable:select: thisArg not supported", 2)
        end

    else

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

        selectorFn = function() return unpack(values, 1, nValues) end

    end

    if MapObservable:isClassOf(self) then
        return self:internalMap(selectorFn)
    else
        return MapObservable(self, selectorFn)
    end

end
