local Rx = require 'Rx'

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

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

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

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    _scheduler = false,
        -- (Scheduler)

    __init = function(self, observer, scheduler)

        self._observer = observer
        self._scheduler = scheduler

    end,

    next = function(self, ...)

        self._observer:onNext(self._scheduler:now(), ... )

    end,

    error = function(self, ...)

        self._observer:onError(...)

    end,

    completed = function(self)

        self._observer:onCompleted()

    end,

}

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

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

    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 timestampObserver = TimestampObserver(observer, self._scheduler)

        return CompositeDisposable(self._source:subscribe(timestampObserver),
            timestampObserver)

    end

}

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

return function(self, scheduler)

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

    return TimestampObservable(self, scheduler)

end
