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

WFQueryDrivenWorkQueueScheduler.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.lua.threading
local WFTimer = require 'WFTimer'

	-- WF.core.reactive
local KSReactive = require 'KSReactive'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFUnknown = require 'WFUnknown'
local WFWorkQueueScheduler = require 'WFWorkQueueScheduler'

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

local check = assert( WFArguments.check )

local table_concat = assert( table.concat )


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

-- @mark WFQueryDrivenWorkQueueScheduler class

local queryDrivenWorkQueueSchedulerMethods = {

	_queryB = false,
		-- (B< array of array >) SQL query results
		
	_doneUpdateNotifierE = false,

	_blackoutWindow = false,
		-- (number) how many seconds to black out a given query row
		-- after the promise resolves
	
	_taskForRowFn = false,
		-- (function) returns a promise for given query row results

	_workQueueScheduler = false,
		-- (WFWorkQueueScheuler)
	
	_blackedOutItems = false,
		-- (set of string) items in this table
		
	_filteredQueryB = false,
		-- (a filtered query, tracked for ownership/closure)
		
	_hasNewWorkB = false,
		-- (tracked for ownership/closure)

	WFCloseSupport.addCloseSupportMethods,

}

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

local function queryRowKeyFromQueryRow( queryRow )

	local tmp = {}
	tmp[ 1 ] = queryRow[ 1 ]
	tmp[ 2 ] = queryRow[ 2 ]
	tmp[ 3 ] = queryRow[ 3 ]
	
	return table_concat( tmp, '\t' )

end

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

local function makeCounterBAndBumpFn()

	local counter = 0
	local counterB = KSReactive.variableB( counter )
	
	local function bumpCounterFn()
		counter = counter + 1
		counterB:set( counter )
	end
	
	return counterB, bumpCounterFn

end

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

local function makeFilteredQueryB( queryB, blackedOutItems )

	local counterB, bumpCounterFn = makeCounterBAndBumpFn()
	
	local filteredQueryB = KSReactive.closingDataCallB(
		function( queryRows, counter )
		
			if WFUnknown.isUnknown( queryRows ) then
				return queryRows
			end
		
			local filteredResults = {}
			local nFilteredResults = 0
			
			for i = 1, #queryRows do

				local queryRow = queryRows[ i ]
				local queryRowKey = queryRowKeyFromQueryRow( queryRow )
				
				if not blackedOutItems[ queryRowKey ] then
					
					nFilteredResults = nFilteredResults + 1
					filteredResults[ nFilteredResults ] = queryRow
				
				end
			
			end
			
			return filteredResults
		
		end,
		queryB,
		counterB )
	
	return filteredQueryB, bumpCounterFn
                                                                                                                                                                                                                                                                                                                                                                                      
end

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

local function makeQueryIsNotEmptyB( filteredQueryB )

	return filteredQueryB:mapB( function( filteredResults )
	
		if WFUnknown.isUnknown( filteredResults ) then
			return false
		end
		
		return filteredResults and #filteredResults > 0
	
	end )

end

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

local function makeUpdateBlackoutListFn( blackedOutItems, bumpCounterFn )

	return function( queryRow, hasTask )
	
		local queryRowKey = queryRowKeyFromQueryRow( queryRow )
		blackedOutItems[ queryRowKey ] = hasTask and true or nil

		bumpCounterFn()
	
	end

end

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

local function makeStartNewWorkFn( self, queryB, taskForRowFn, updateBlackoutListFn )

	return function()

		local queryRows = queryB:valueNow()
		local blackedOutItems = self._blackedOutItems
		
		if type( queryRows ) ~= 'table' or WFUnknown.isUnknown( queryRows ) then return end
			-- shouldn't happen, but just in case ...

		-- It's possible that the query behavior might fire twice or might
		-- not get updated immediately, so we scan the list looking for the first
		-- item for which we haven't already launched a task.
	
		for i = 1, #queryRows do
	
			local queryRow = queryRows[ i ]

			local queryRowKey = queryRowKeyFromQueryRow( queryRow )
			if not blackedOutItems[ queryRowKey ] then
	
				-- No prior task for this query row: Go for it!
	
				local taskP = taskForRowFn( queryRow )
				assert( taskP ~= nil )
				
				updateBlackoutListFn( queryRow, true )
				
				-- Keep this task in the blackout window for specified interval afterwards.
				
				taskP
					:mapP( function()
						return WFTimer.afterDelayP( self._blackoutWindow )
					end )
					:whenResolvedDo( function()
						updateBlackoutListFn( queryRow, false )
					end )
					:attachTo( self )
					:suppressCloseLogging()
				
				return taskP
			end
			
		end
		
		-- If we get here, we've already launched tasks for all of the query
		-- result rows.
		
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Create an object that will create promises from a work queue and
		execute them at a restricted rate.
	
	\param queryB (B< array of array >) raw query behavior (typically this
		is generated by WFIndexStore:queryB)
	\param maxConcurrentTasks (number) maximum number of tasks to
		execute at any time
	\param blackoutWindow (number) how many seconds to wait after a given
		query row's task has completed before attempting that same task
		again
	\param taskForRowFn (function) this function should return a new promise
		for given query row results
]]--

function queryDrivenWorkQueueSchedulerMethods:__init( queryB,
													  maxConcurrentTasks,
													  blackoutWindow,
													  taskForRowFn )

	check( 'WFQueryDrivenWorkQueueScheduler', '_', 'number', 'number', 'function' )
	
	assert( maxConcurrentTasks >= 1 )
	
	assert( blackoutWindow > 0 )
	self._blackoutWindow = blackoutWindow

	-- Derive filtered queryB by excluding items which have been recently
	-- attempted.

	self._blackedOutItems = {}

	self._doneUpdateNotifierE = KSReactive.sourceE()
	
	local filteredQueryB, bumpBlackoutListFn = makeFilteredQueryB( queryB,
																   self._blackedOutItems )

	local hasNewWorkB = makeQueryIsNotEmptyB( filteredQueryB )
	
	-- Derive a start-new-work function from caller's new task function
	-- and the additional logic necessary to enforce the blackout window.
	
	local updateBlackoutListFn = makeUpdateBlackoutListFn( self._blackedOutItems,
														   bumpBlackoutListFn )
	
	local startNewWorkFn = makeStartNewWorkFn( self,
											   filteredQueryB,
											   taskForRowFn,
											   updateBlackoutListFn )
	
	-- Now build the work scheduler, and we're off to the races!
	
	self:setOwned {
		_workQueueScheduler = WFWorkQueueScheduler(
			hasNewWorkB, maxConcurrentTasks, startNewWorkFn ),
		_filteredQueryB = filteredQueryB,
		_hasNewWorkB = hasNewWorkB,
	}
	
end

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

function queryDrivenWorkQueueSchedulerMethods:getDoneUpdateNotifierE()
	return self._doneUpdateNotifierE
end

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

return Rx.Class 'WFQueryDrivenWorkQueueScheduler'( queryDrivenWorkQueueSchedulerMethods )
