local Rx = require 'Rx'

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

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

local ColdObservable = {

    Observable,

    class = {},

    _scheduler = false,

    _messages = false,

    _subscriptions = false,

}

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

function ColdObservable:__init(scheduler, messages)

    self._scheduler = scheduler
    self._messages = messages
    self._subscriptions = {}

end

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

function ColdObservable:_subscribe(observer)

    local subscriptions = self._subscriptions
    local index = #subscriptions + 1
    subscriptions[index] = Subscription(self._scheduler:clock())

    local d = CompositeDisposable()

    for i, message in ipairs(self._messages) do
        local notification = message.value
        d:add(self._scheduler:scheduleRelative(message.time, function()
            notification:accept(observer)
            return Disposable.empty()
        end))
    end

    return Disposable.create(function()
        subscriptions[index] = Subscription(subscriptions[index].subscribe,
            self._scheduler:clock())

        d:dispose()
    end)

end

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

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