local Rx = require 'Rx'

local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable
local SerialDisposable = Rx.SerialDisposable
local Subject = Rx.Subject

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

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

    Observable,
        -- superclass

    _source = false,
        -- (Observable)

    _controller = false,
        -- (Subject)

    _pauser = false,
        -- (Observable)

    __init = function(self, source, pauser)

        self._source = source

        self._controller = Subject()

        if pauser and pauser.subscribe then
            self._pauser = self._controller:merge(pauser)
        else
            self._pauser = self._controller
        end

    end,

    _subscribe = function(self, observer)

        local conn = self._source:publish()

        local subscription = conn:subscribe(observer)

        local disposable = SerialDisposable()

        local pausable = self._pauser
            :distinctUntilChanged()
            :subscribe(function(notPaused)

                if notPaused then
                    disposable:setDisposable(conn:connect())
                else
                    disposable:setDisposable(nil)
                end

            end)

        return CompositeDisposable(disposable, subscription,
            pausable, self._controller, conn)

    end,

    pause = function(self)
        self._controller:onNext(false)
    end,

    resume = function(self)
        self._controller:onNext(true)
    end

}

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

return function(self, pauser)
    return PausableObservable(self, pauser)
end
