local Rx = require 'Rx'

local AutoDetachObserver = Rx.AutoDetachObserver
local Disposable = Rx.Disposable
local Observable = Rx.Observable
local Scheduler = Rx.Scheduler

--------------------------------------------------------------------------------
-- Fix subscriber to check for nil or function returned to decorate as Disposable.

local function fixSubscriber(subscriber)

    local ts = type(subscriber)

    if ts == 'table' and Disposable:isClassOf(subscriber) then
        return subscriber
    end

    if ts == 'function' then
        return Disposable.create(subscriber)
    end

    return Disposable.empty()

end

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

local function setDisposable(s, state)

    local ado = state[1]
    local self = state[2]

    local success, subcriberOrErr = pcall(self.__subscribe, ado)

    if success then

        ado:setDisposable(fixSubscriber(subcriberOrErr))

    else

        if not ado:fail(subcriberOrErr) then
            error(subcriberOrErr)
        end

    end

end

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

local AnonymousObservable = Rx.Class 'Rx.AnonymousObservable' {

    Observable,

    _source = false,
        -- (Observable) parent source

    __subscribe = false,
        -- (function) subscribe

    __init = function(self, subscribe, parent)

        if type(subscribe) ~= 'function' then
            error("Rx.Observable: subscribe argument must be a function", 3)
        end

        if parent then
            if Observable:isClassOf(parent) then
                self._source = parent
            else
                error("Rx.Observable: parent argument must be an Observable or nil", 3)
            end
        end

        self.__subscribe = subscribe

    end,

    _subscribe = function(self, observer)

        local ado = AutoDetachObserver(observer)

        local state = { ado, self }

        local currentThreadScheduler = Scheduler.currentThread()

        if currentThreadScheduler:scheduleRequired() then
            currentThreadScheduler:schedule(state, setDisposable)
        else
            setDisposable(nil, state)
        end

        return ado

    end,

}

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

return AnonymousObservable
