local Rx = require 'Rx'

local AsyncSubject = Rx.AsyncSubject

local unpack = table.unpack or unpack

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

local function packResults(success, ...)

    return success, { n = select('#', ...), ... }

end

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

return function(fn, selector, ignored)

    if selector and type(selector) ~= 'function' or ignored ~= nil then
        error("Rx.Observable.fromCallback: context not supported", 2)
    end

    return function(...)

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

        local subject = AsyncSubject()

        local function handler(...)

            if selector then

                local success, results = packResults(pcall(selector, ...))

                if success then

                    subject:onNext(unpack(results, 1, results.n))

                else

                    subject:onError(unpack(results, 1, results.n))

                end

            else

                subject:onNext(...)

            end

            subject:onCompleted()

        end

        args.n = args.n + 1
        args[args.n] = handler

        local success, errMsg = pcall(fn, unpack(args, 1, args.n))

        if not success then
            subject:onError(errMsg)
        end

        return subject:asObservable(), subject

    end

end
