local Rx = require 'Rx'

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

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

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(scheduler, state)

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

    local success, subscriberOrErr = pcall(self.subscribeCore, self, ado)

    if success then

        ado:setDisposable(fixSubscriber(subscriberOrErr))

    else

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

    end

end

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

local ObservableBase = Rx.Class 'Rx.ObservableBase' {

    Observable,
        -- superclass

    subscribeCore = false,

    _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 ObservableBase
