--[[----------------------------------------------------------------------------

WFRequestQueue.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and are protected by all applicable
intellectual property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

----------------------------------------------------------------------------]]--

	-- RxLua
local Rx = require 'Rx'
local ReplaySubject = Rx.ReplaySubject
local Disposable = Rx.Disposable

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.utils
local WFQueueing = require 'WFQueueing'


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

local log = WFLogger 'WFRequestQueue'
--	:setLevel( 'DEBUG' )

--============================================================================--

-- @mark WFRequestQueue class

--------------------------------------------------------------------------------
--[[!
	\brief WFRequestQueue maintains a list of requests to be processed.
	
]]--

local WFRequestQueue = {

	Disposable,
		-- Base class

	_size = false,
		-- (Rx.ReplaySubject< number >) how many items are in queue?

	_totalSize = false,
		-- (Rx.ReplaySubject< number >) how many items have been processed
		-- plus the current number of items in the queue

	_sizeNow = 0,
		-- (number) how many items are in queue? (feeds _size)

	_processedCount = 0,
		-- (number) how many items have been processed? (feeds _totalSize)

	_enqueueFunc = false,
		-- (function) enqueue an item

	_dequeueFunc = false,
		-- (function) dequeue an item

}

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

function WFRequestQueue:__init( )

	self._size = ReplaySubject( 1 )
	self._size:onNext( 0 )

	self._totalSize = ReplaySubject( 1 )
	self._totalSize:onNext( 0 )

	local enqueue, dequeue = WFQueueing.makeQueue()

	self._enqueueFunc = enqueue
	self._dequeueFunc = dequeue

end

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

function WFRequestQueue:dispose()
	
	if self._size then
		local size = self._size
		self._size = nil
		
		size:onCompleted()
		size:dispose()
	end

	if self._totalSize then
		local totalSize = self._totalSize
		self._totalSize = nil
		
		totalSize:onCompleted()
		totalSize:dispose()
	end

	self._enqueueFunc = nil
	self._dequeueFunc = nil
	
	return self:Rx_Disposable_dispose()

end

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

local function adjustQueueSize( self, delta )

	local newSize = self._sizeNow + delta
	assert( newSize >= 0 )

	local totalSize = newSize + self._processedCount
	self._totalSize:onNext( totalSize )
	self._sizeNow = newSize
	self._size:onNext( newSize )
	
end

--------------------------------------------------------------------------------
--[[!
	\brief Resets the state of the queue.

]]--

function WFRequestQueue:reset()
	
	self._processedCount = 0
	self._totalSize:onNext( 0 )
	self._sizeNow = 0
	self._size:onNext( 0 )

end

--------------------------------------------------------------------------------
--[[!
	\brief Add a request to the queue.

	\param request (?) request object to be added

]]--

function WFRequestQueue:addToQueue( request )

	log:debugf( "queueing request %s", request )

	self._enqueueFunc( request )
	adjustQueueSize( self, 1 )

end

--------------------------------------------------------------------------------
--[[!
	\brief Reports number of pending items in the queue

	\return Observable which reports size of the queue

]]--

function WFRequestQueue:size()

	return self._size

end

--------------------------------------------------------------------------------
--[[!
	\brief Reports total number of items which is the sum
	of processed items and the pending items in the queue.

	\return Observable which reports total size of the queue

]]--

function WFRequestQueue:totalSize()

	return self._totalSize

end

--------------------------------------------------------------------------------
--[[!
	\brief Removes and returns the next item to be 
	processed. 
	Once the queue becomes persistent, it might be
	required for us to have a peek API which 
	returns but doesnt remove the item to be processed.

	\return the next item to be processed.

]]--

function WFRequestQueue:nextRequest()

	local result = self._dequeueFunc()

	if result then
		self._processedCount = self._processedCount + 1
		adjustQueueSize( self, -1 )
	end

	return result

end


--============================================================================--

return Rx.Class 'WFRequestQueue'( WFRequestQueue )
