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

WFWorkQueueScheduler.lua

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

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

NOTICE: All information contained herein is, and remain 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 may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  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'

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

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFDebugTraceback = require 'WFDebugTraceback'


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

local check = assert( WFArguments.check )

local logger = WFLogger 'WFWorkQueueScheduler'
--	:setLevel 'debug'


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

-- @mark WFWorkQueueScheduler class

local workQueueSchedulerMethods = {

	_hasNewWorkB = false,
		-- (B< boolean >) true if there is unstarted work to perform
	
	_startNewWorkFn = false,
		-- (function) closure that returns a promise for next item in work queue

	_traceback = false,
		-- (WFDebugTraceback) traceback for creation of this limiter

	_nOutstandingTasks = 0,
		-- (number) how many tasks are currently in flight?

	_maxConcurrentTasks = false,
		-- (number) how many concurrent tasks are permitted?

	WFCloseSupport.addCloseSupportMethods,

}

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

local startNewWork -- forward declaration

--------------------------------------------------------------------------------
--[[!
	\brief Create an object that will create promises from a work queue and
		execute them at a restricted rate.
	
	\param hasNewWorkB (B< boolean >) should be true when there is unstarted
		work to be performed
	\param maxConcurrentTasks (number) maximum number of tasks to
		execute at any time
	\param startNewWorkFn (function) this function should return a new promise
		for the next item in the work queue (may return \c nil if there is no
		new work to be done)
	
	\warning The caller's \c startNewWorkFn must manage the lifetime of this
		promise appropriately. Typically this is done by attaching an action
		and calling \c :attachTo(something) on the action.

	\warning The \c startNewWorkFn receives a function \c attachObserverFn
		which must be called with the promise to be created. You will see log
		traffic warning you if you fail to do this. This function attaches
		an observer to release the rate-limit queue when the promise's execution
		has completed.
]]--

function workQueueSchedulerMethods:__init( hasNewWorkB,
										   maxConcurrentTasks,
										   startNewWorkFn )

	check( 'WFworkQueueScheduler', '_', 'number', 'function' )
	
	assert( maxConcurrentTasks >= 1 )
	self._maxConcurrentTasks = maxConcurrentTasks
	
	self:setOwned { _hasNewWorkB = hasNewWorkB:autoCloseRelayB() }
	
	self._startNewWorkFn = startNewWorkFn

	self._traceback = WFDebugTraceback.traceback { 'WFWorkQueueScheduler' }

	-- Try to queue up a new task whenever has new work transitions to true.
	
	self._hasNewWorkB:ownedOnDataB( self, function( value )
		
		if value then
			startNewWork( self )
		end
	
	end )

end

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

startNewWork = function( self )

	if WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		return
	end
	
	while self._nOutstandingTasks < self._maxConcurrentTasks do
	
		local resultP = self._startNewWorkFn()
		
		if resultP == nil then
			break
				-- No more new work. We'll wait for self._hasNewWorkB
				-- to transition back to true as a signal that there
				-- is new work to schedule.
		end
		
		resultP
			:whenResolvedDo( function()

				self._nOutstandingTasks = self._nOutstandingTasks - 1

				logger:debugf( "task finished, #outstanding tasks = %d",
							   self._nOutstandingTasks )
	
				startNewWork( self )
			
			end )
			:attachTo( self )
			:suppressCloseLogging()
		
		self._nOutstandingTasks = self._nOutstandingTasks + 1

		logger:debugf( "task started, #outstanding tasks = %d",
					   self._nOutstandingTasks )

	end

end

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

return Rx.Class 'WFworkQueueScheduler'( workQueueSchedulerMethods )
