local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local ObservableBase = Rx.ObservableBase
local SchedulerArgument = Rx.internals.SchedulerArgument

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

local TimeIntervalObserver = Rx.Class 'Rx.internals.TimeIntervalObserver' {

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _scheduler = false,
        -- (Scheduler)

    _last = 0,
        -- (number)

    __init = function(self, observer, scheduler)

        self._observer = observer
        self._scheduler = scheduler
        self._last = scheduler:now()

    end,

    next = function(self, ...)

        local now = self._scheduler:now()

        self._observer:onNext(now - self._last, ...)

        self._last = now

    end,

    error = function(self, ...)

        self._observer:onError(...)

    end,

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

local TimeIntervalObservable = Rx.Class 'Rx.internals.TimeIntervalObservable' {

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    _scheduler = false,
        -- (Scheduler)

    __init = function(self, source, scheduler)

        self._source = source
        self._scheduler = scheduler

    end,

    subscribeCore = function(self, observer)

        local intervalObserver = TimeIntervalObserver(observer, self._scheduler)

        return CompositeDisposable(self._source:subscribe(intervalObserver),
            intervalObserver)
    end,

}

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

return function(self, scheduler)

    scheduler = SchedulerArgument.validateWithoutDefault(
        'Rx.Observable:timeInterval', scheduler)

    return TimeIntervalObservable(self, scheduler)

end
