local Rx = require 'Rx'
local AnonymousObservable = Rx.AnonymousObservable
local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable
local observableInterval = Rx.Observable.interval
local SchedulerArgument = Rx.internals.SchedulerArgument
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable

local unpack = table.unpack or unpack

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

local function sampleObservable(self, sampler)

    return AnonymousObservable(function(observer)

        local atEnd = false
        local hasValue = false
        local value

        local function sendSampledValue()

            if hasValue then
                observer:onNext(unpack(value, 1, value.n))
                hasValue = false
                value = nil
            end

            if atEnd then
                observer:onCompleted()
            end

        end

        local function sendError(...)
            return observer:onError(...)
        end

        local sourceSubscription = SingleAssignmentDisposable()
        sourceSubscription:setDisposable(self:subscribe(
            function(...)
                hasValue = true
                value = { n = select('#', ...), ... }
            end,
            function(...)
                return observer:onError(...)
            end,
            function()
                atEnd = true
                sourceSubscription:dispose()
            end)
        )

        return CompositeDisposable(
            sourceSubscription,
            sampler:subscribe(sendSampledValue, sendError, sendSampledValue)
        )

    end, self)

end

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

return function(self, intervalOrSampler, scheduler)

    if type(intervalOrSampler) == 'number' then
        scheduler = SchedulerArgument.validateWithoutDefault(
            'Rx.Observable:sample', scheduler)
        return sampleObservable(self, observableInterval(intervalOrSampler, scheduler))
    elseif Observable:isClassOf(intervalOrSampler) then
        return sampleObservable(self, intervalOrSampler)
    else
        error("Rx.Observable:sample: intervalOrSampler must be a number or an Observable", 2)
    end

end
