local Rx = require 'Rx'

local Disposable = Rx.Disposable
local Observable = Rx.Observable
local Subscription = Rx.testing.Subscription

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

local HotObservable = {

    Observable,

    class = {},

    _scheduler = false,

    _messages = false,

    _subscriptions = false,

    _observers = false,

}

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

function HotObservable:__init(scheduler, messages)

    self._scheduler = scheduler
    self._messages = messages

    self._subscriptions = {}

    self._observers = {}

end

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

function HotObservable:__inited()

    for i, message in ipairs(self._messages) do

        local notification = message.value

        self._scheduler:scheduleAbsolute(message.time, function()

            local obs = {}
            for idx, observer in ipairs(self._observers) do
                obs[idx] = observer
            end

            for idx, observer in ipairs(obs) do
                notification:accept(observer)
            end

        end)

    end

end

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

function HotObservable:_subscribe(observer)

    local observers = self._observers
    local subscriptions = self._subscriptions
    local scheduler = self._scheduler

    observers[#observers + 1] = observer
    subscriptions[#subscriptions + 1] = Subscription(scheduler:clock())

    local index = #subscriptions

    return Disposable.create(function()
        for idx, nextObserver in ipairs(observers) do
            if nextObserver == observer then
                table.remove(observers, idx)
                break
            end
        end
        subscriptions[index] = Subscription(subscriptions[index].subscribe, scheduler:clock())
    end)

end

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

return Rx.Class('Rx.testing.HotObservable', HotObservable)
