local Rx = require 'Rx'

local CompositeDisposable = Rx.CompositeDisposable
local Observable = Rx.Observable
local SchedulerArgument = Rx.internals.SchedulerArgument
local Subject = Rx.Subject

local notificationCreateOnCompleted = Rx.Notification.createOnCompleted
local notificationCreateOnError = Rx.Notification.createOnError
local notificationCreateOnNext = Rx.Notification.createOnNext

local disposableCreate = Rx.Disposable.create

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

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

    Subject,
        -- superclass

    _enableQueue = false,
        -- (boolean)

    _queue = false,
        -- (table)

    _subject = false,
        -- (Subject)

    _requestedCount = false,
        -- (number)

    _requestedDisposable = false,
        -- (Disposable)

    _error = false,
        -- (table)

    _hasFailed = false,
        -- (boolean)

    _hasCompleted = false,
        -- (boolean)

    _scheduler = false,
        -- (Scheduler)

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

    __init = function(self, enableQueue, scheduler)

        self._enableQueue = enableQueue or enableQueue == nil
        self._queue = {}
        self._subject = Subject()
        self._requestedCount = 0
        self._requestedDisposable = nil
        self._error = nil
        self._hasFailed = false
        self._hasCompleted = false
        self._scheduler = scheduler

    end,

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

    _subscribe = function(self, observer)
        local subject = self._subject
        return CompositeDisposable(subject:subscribe(observer), subject)
    end,

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

    onCompleted = function(self)

        self._hasCompleted = true

        if not self._enableQueue or #self._queue == 0 then

            self._subject:onCompleted()
            self:disposeCurrentRequest()

        else

            table.insert(self._queue, notificationCreateOnCompleted())

        end

    end,

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

    onError = function(self, ...)

        self._hasFailed = true

        self._error = { n = select('#', ...), ... }

        if not self._enableQueue or #self._queue == 0 then

            self._subject:onError(...)
            self:disposeCurrentRequest()

        else

            table.insert(self._queue, notificationCreateOnError(...))

        end

    end,

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

    onNext = function(self, ...)

        if self._requestedCount <= 0 then

            if self._enableQueue then
                table.insert(self._queue, notificationCreateOnNext(...))
            end

        else

            if self._requestedCount == 1 then
                self:disposeCurrentRequest()
            end

            self._requestedCount = self._requestedCount - 1

            self._subject:onNext(...)

        end

    end,

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

    _processRequest = function(self, numberOfItems)

        if self._enableQueue then

            while #self._queue > 0 and (numberOfItems > 0
                or self._queue[1]:kind() ~= 'N') do

                local first = table.remove(self._queue, 1)
                first:accept(self._subject)

                if first:kind() == 'N' then
                    numberOfItems = numberOfItems - 1
                else
                    self:disposeCurrentRequest()
                    self._queue = {}
                end

            end

        end

        return numberOfItems

    end,

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

    request = function(self, number)

        self:disposeCurrentRequest()

        self._requestedDisposable = self._scheduler:schedule(number,
            function(_, i)

                local numberOfItemsRemaining = self:_processRequest(i)

                local stopped = self._hasCompleted or self._hasFailed

                if not stopped and numberOfItemsRemaining > 0 then

                    self._requestedCount = numberOfItemsRemaining

                    return disposableCreate(function()
                        self._requestedCount = 0
                    end)
                        -- Scheduled item is still in progress. Return a new
                        -- disposable to allow the request to be interrupted
                        -- via dispose.

                end

            end)

        return self._requestedDisposable

    end,

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

    disposeCurrentRequest = function(self)

        if self._requestedDisposable then

            self._requestedDisposable:dispose()
            self._requestedDisposable = nil

        end

    end

}

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

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

    Observable,
        -- superclass

    _source = false,
        -- (Observable)

    _subject = false,
        -- (ControlledSubject)

    _multicast = false,
        -- (ConnectableObservable)

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

    __init = function(self, source, enableQueue, scheduler)

        self._subject = ControlledSubject(enableQueue, scheduler)

        local multicast = source:multicast(self._subject)
        self._multicast = multicast
        self._source = multicast:refCount()

    end,

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

    _subscribe = function(self, observer)

        return CompositeDisposable(self._source:subscribe(observer),
            self._subject, self._multicast)

    end,

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

    request = function(self, numberOfItems)

        return self._subject:request(numberOfItems or -1)

    end,

}

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

return function(self, enableQueue, scheduler)

    scheduler = SchedulerArgument.validateWithDefault('Rx.Observable:controlled',
                                                      'currentThread', scheduler)

    if enableQueue == nil then
        enableQueue = true
    end

    return ControlledObservable(self, enableQueue, scheduler)

end
