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

PixelPathProviderScheduler.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2016 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.

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

	-- wf.core
local KSTableUtils = require "KSTableUtils"
local WFArguments = require 'WFArguments'
local WFArguments_check = assert( WFArguments.check,
										"missing WFArguments.check" )
local WFCloseSupport = require "WFCloseSupport"
local WFFiniteStateMachine = require "WFFiniteStateMachine"
local WFFunctionUtils = require "WFFunctionUtils"
local WFPerformanceReport = require "WFPerformanceReport"

	-- scheduler helpers
local WFCatRenderSize = require "WFCatRenderSize"
local PixelPathProviderSchedulerDiagnosticLog =
		require "PixelPathProviderSchedulerDiagnosticLog"
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local PixelPathProviderSchedulerMainLoop =
		require "PixelPathProviderSchedulerMainLoop"
local PixelPathProviderPriorityQueue = require "PixelPathProviderPriorityQueue"
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelProviderSharedBlobCache = require "PixelProviderSharedBlobCache"
local PixelPathProviderStateMachine = require "PixelPathProviderStateMachine"

	-- Rx
local Rx = require "Rx"

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

local WFFunctionUtils_map = assert( WFFunctionUtils.map,
			"WFFunctionUtils.map does not exist" )

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

local WFFunctionUtils_filter = assert( WFFunctionUtils.filter,
			"WFFunctionUtils.filter does not exist" )

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

local WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

--------------------------------------------------------------------------------
-- The PixelPathProviderScheduler has three main responsibilities:
--
-- 1) It exposes an API for adding and removing pixel path requests.
--
-- 2) It dispatches pixel path requests to the appropriate task needed to
--    generate the pixel path. It does this on a dedicated run loop in
--    PixelPathProviderSchedulerMainLoop. It uses the
--    PixelPathProviderPriorityQueue and PixelPathProviderStateMachine to
--    prioritize and track a request through the various stages needed to fetch
--    the path.
--
-- 3) It exposes an Rx.Observable that is a stream of
--
-- 			pixStreamId, localAssetId, success, pixelPathOrError, maybeDigest,
-- 			maybeRenderSize 
--
-- 	  tuples
--
--------------------------------------------------------------------------------

local logger = require 'WFLogger'( "PixelPathProviderScheduler" )

--==============================================================================
-- shared functions
--==============================================================================

local function forwardToPriorityQueue( funcName )

	return function( self, ... )

		return self._requestPriorityQueue[ funcName ](
					self._requestPriorityQueue, ... )

	end

end

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

local function makeWeakRef( v )

	local weakRef = KSTableUtils.makeWithMode( "kv" )

	weakRef[ v ] = v

	return weakRef

end

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

local function derefWeakRef( weakRef )

	return next( weakRef )

end

--==============================================================================
-- PixelPathProviderScheduler prototype
--==============================================================================

local PixelPathProviderSchedulerPrototype = {

	_anyPixStreamIdPixelPathStream = false,
		-- an Rx.Observable on which we emit pixel streams

	_pixelPathStreamBacklogQueue = false,
		-- queue functions for pixelpath stream messages, we don't send
		-- messages from the _runLoopCoroutineThread, we queue them then send
		-- them on yield 

	_requestPriorityQueue = false,
		-- a PixelPathProviderPriorityQueue instance created by the scheduler.
		-- This is an in memory db priority queue of all the pixel provider
		-- requests.

	_stateMachine = false,
		-- a PixelPathProviderStateMachine created by the scheduler. This
		-- determines the next unit of work for each request in the priority
		-- queue

	_runLoopCoroutineThread = false,
		-- a couroutine instance of PixelPathProviderSchedulerMainLoop.run
		-- dispatches requests to the various pixel provider processes

	_batchScheduleRequests = false,
		-- a nesting number that when positive, batches up requests to schedule
		-- pixel generation.  Used when scheduling many requests at once

	_renderBlackoutCount = false,
		-- a boolean that the PixelProviderRenderTask uses to throttle 
		-- concurrent renders.
		
	_shouldRenderForNewRenditions = false,
		-- a boolean that the PixelProviderChooseRenditionStrategyTask uses to 
		-- decide whether to prefer downloads or renders
	
	_testData = false,
		-- optional table of data tests can use inject behaviorial changes

	_DEBUG_requestStatsStreamSubscriberCount = false,
	_DEBUG_requestStatsObservable = false,
	_DEBUG_requestStatsStreamData = false,
	_DEBUG_requestStatsChangeCounter = false,
		-- used to implement requestStatsStream

	_DEBUG_localAssetIdWatchCounts = false,
	_DEBUG_localAssetIdStatesStreams = false,
	_DEBUG_localAssetIdStatesChangeCounters = false,
		-- used to implement localAssetIdStatesStream
		
	_staminaPollerId = false,

	_daemonDownloadStatus = false,
		-- a table indiciating the current download status of the daemon.

	WFCloseSupport.addCloseSupportMethods,

}

--==============================================================================
-- private methods. These are mainly used by the main loop
--==============================================================================

--------------------------------------------------------------------------------
-- _maybeSetNextState is more expensive that _applyTransitionToRequest because
-- it must lookup the request in the _requestPriorityQueue before updating
-- the state.
--
-- That is needed in a worker thread callback because we have no guarantee
-- the request still exists and if it does, no guarantee it had the same state
-- as it did before we started the worker thread task
--------------------------------------------------------------------------------
PixelPathProviderSchedulerPrototype._maybeSetNextState =
	PixelPathProviderSchedulerPrivateMethods.maybeSetNextState

--------------------------------------------------------------------------------
-- _applyTransitionToRequest is much more efficient that _maybeSetNextState and
-- it should be used whenever you update can update a request without yielding
-- to worker thread for further work.
--------------------------------------------------------------------------------
PixelPathProviderSchedulerPrototype._applyTransitionToRequest =
	PixelPathProviderSchedulerPrivateMethods.applyTransitionToRequest

PixelPathProviderSchedulerPrototype._applyTransitionAllRequests =
	PixelPathProviderSchedulerPrivateMethods.applyTransitionAllRequests

PixelPathProviderSchedulerPrototype._getRequestsWithStatesInPriorityOrder =
	PixelPathProviderSchedulerPrivateMethods.getRequestsWithStatesInPriorityOrder

PixelPathProviderSchedulerPrototype._mapAllRequestsWithStates =
	PixelPathProviderSchedulerPrivateMethods.mapAllRequestsWithStates

PixelPathProviderSchedulerPrototype._mapAllRequestsWithPixStreamId =
	PixelPathProviderSchedulerPrivateMethods.mapAllRequestsWithPixStreamId

PixelPathProviderSchedulerPrototype._mapAllRequestsWithState =
	PixelPathProviderSchedulerPrivateMethods.mapAllRequestsWithState

PixelPathProviderSchedulerPrototype._getLocalAssetIdRequestsWithState =
	PixelPathProviderSchedulerPrivateMethods.getLocalAssetIdRequestsWithState

PixelPathProviderSchedulerPrototype._hasPixstreamRequestForLocalAssetId =
	PixelPathProviderSchedulerPrivateMethods.hasPixstreamRequestForLocalAssetId

PixelPathProviderSchedulerPrototype._expose_test_API =
	PixelPathProviderSchedulerPrivateMethods.expose_test_API

PixelPathProviderSchedulerPrototype._sendTestData =
	PixelPathProviderSchedulerPrivateMethods.sendTestData

--==============================================================================
-- request stats helpers
--==============================================================================

local function flattenRequestStat( flattened, requestStat )

	local requestTypeName, requestTypeCount = unpack( requestStat )

	flattened[ requestTypeName ] = requestTypeCount

	return flattened

end

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

local function generateStaminaPollingMetric( self )

	local stats = self._requestPriorityQueue:_DEBUG_getRequestCountsByState()

	return WFFunctionUtils_reduce( {}, flattenRequestStat, stats.all )

end

--==============================================================================
-- public object methods
--==============================================================================

function PixelPathProviderSchedulerPrototype:__init( wfCatalog, pixelProviderSettings )

	WFArguments_check( 'PixelPathProviderScheduler', 'WFCatalog' )

	self._anyPixStreamIdPixelPathStream = Rx.Subject()

	self:setOwned {

		_requestPriorityQueue =
			PixelPathProviderPriorityQueue(
				PixelPathProviderStateMachine.DocstoreInfoStates.NeedAssetInfoForRender )

	}

	self._stateMachine = PixelPathProviderStateMachine.make()

	self._daemonDownloadStatus = {}

	self._runLoopCoroutineThread = 
			PixelPathProviderSchedulerMainLoop.createRunLoopCoroutineThread( 
				self, wfCatalog, pixelProviderSettings )

	self._staminaPollerId =
			WFPerformanceReport.registerPollingMetric( 
				0.5, "PixelProviderPolingStat", 
					generateStaminaPollingMetric, self )

end

--------------------------------------------------------------------------------
-- close
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:close()

	logger:info( "scheduler is closing" )

	if self._runLoopCoroutineThread then

		if coroutine.status( self._runLoopCoroutineThread ) == "running" then

			logger:warn( "closing scheduler from run loop" )

		end

		local ok, maybeLoopExitStatus = 
				coroutine.resume( self._runLoopCoroutineThread, "quit" )

		if not ok then

			logger:errorf( "failed to quit the runLoop: %s", 
								maybeLoopExitStatus )

			return 

		end

		local finalStatus = coroutine.status( self._runLoopCoroutineThread )

		if maybeLoopExitStatus ~= "done" or finalStatus ~= "dead" then

			logger:errorf( "run loop did not exit cleanly. Loop exit status: %s. Coroutine status: %s",
								maybeLoopExitStatus, finalStatus )

		end

	end

	WFPerformanceReport.deregisterPollingMetric( self._staminaPollerId )

end

--------------------------------------------------------------------------------
-- render blackout management
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:startRenderBlackout()

	self._renderBlackoutCount = ( self._renderBlackoutCount or 0 ) + 1

end

--------------------------------------------------------------------------------
 
function PixelPathProviderSchedulerPrototype:endRenderBlackout()
	
	self._renderBlackoutCount = ( self._renderBlackoutCount or 0 ) - 1

	if self._renderBlackoutCount < 0 then

		logger:warn( "endRenderBlackout called more often than startRenderBlackout" )
	
		self._renderBlackoutCount = 0

	end

end

--------------------------------------------------------------------------------
 
function PixelPathProviderSchedulerPrototype:isRenderBlackoutActive()

	return ( self._renderBlackoutCount or 0 ) > 0

end

--------------------------------------------------------------------------------
-- render strategy override
--------------------------------------------------------------------------------
 
function PixelPathProviderSchedulerPrototype:setPreferRenderForNewRenditions(
												shouldRenderForNewRenditions )

	self._shouldRenderForNewRenditions = not not shouldRenderForNewRenditions

end

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

local _preferRenderForNewRenditionsFromDaemonConfig =
		not not require "WFConfiguration".Stamina.preferRenderForNewRenditions

--------------------------------------------------------------------------------
 
function PixelPathProviderSchedulerPrototype:shouldPreferDownloadsForNewRenditions()

	return not( _preferRenderForNewRenditionsFromDaemonConfig or
					self._shouldRenderForNewRenditions )

end

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

local function maybeUpdateLongSide( self, pixStreamId, localAssetId, longSide )

	WFArguments_check( "PixelPathProviderScheduler",
						"string", "number", "number|string" )

	local didUpdate =
			self._requestPriorityQueue:maybeUpdateLongSide(
										pixStreamId, localAssetId, longSide )

	if not didUpdate then

		return false

	end

	return self:_maybeSetNextState( pixStreamId, localAssetId, 'new_long_side' )

end

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

PixelPathProviderSchedulerPrototype.maybeUpdateLongSide = maybeUpdateLongSide

--------------------------------------------------------------------------------
-- addRequest
--------------------------------------------------------------------------------

local function addOrUpdateRequest( self, pixStreamId, localAssetId,
									newRequestPriority, longSide,
									maybeOverrideInitialState )

	WFArguments_check( "PixelPathProviderScheduler",
						"string", "number", "string", "number|string", "?string" )

	local didUpdate, prevLongSide =
			self._requestPriorityQueue:maybeUpdateRequest(
										pixStreamId, localAssetId,
										newRequestPriority, longSide )

	if didUpdate then

		if prevLongSide ~= longSide then

			self:_maybeSetNextState( pixStreamId, localAssetId, 'new_long_side' )

		end

		return

	end

	self._requestPriorityQueue:addNewRequest(
								pixStreamId, localAssetId,
								newRequestPriority, longSide, 
								maybeOverrideInitialState )

end

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

addOrUpdateRequest =
	PixelPathProviderSchedulerPrivateMethods.wrapInScheduleRequests( 
		addOrUpdateRequest )

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

PixelPathProviderSchedulerPrototype.addRequest = addOrUpdateRequest

--------------------------------------------------------------------------------
-- getHighResVideoThumb
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:getHighResVideoThumb(
												pixStreamId, localAssetId,
												priority )

	return addOrUpdateRequest( self, pixStreamId, localAssetId, priority, 
								2560 )


end

--------------------------------------------------------------------------------
-- getProxy
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:getProxy( pixStreamId,
														localAssetId,
														priority )

	return addOrUpdateRequest( 
			self, pixStreamId, localAssetId, priority, 'proxy',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedDownloadProxyInfo )

end

--------------------------------------------------------------------------------
-- requestOriginalForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestOriginalForExport( pixStreamId,
															localAssetId )

	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists, 
			'master',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedDownloadOriginalInfo
			)

end

--------------------------------------------------------------------------------
-- requestFullSizeRenditionForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestFullSizeRenditionForExport( 
				pixStreamId, localAssetId )

	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists, 
			2560, -- this size is arbitrary, we'll request the biggest rendition no matter what
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedAssetInfoForExportFullRendition
			)

end

--------------------------------------------------------------------------------
-- requestBiggestPossibleRenditionForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestBiggestPossibleRenditionForExport( 
				pixStreamId, localAssetId )

	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists, 
			2048,
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedAssetInfoForExportBiggestPossibleRendition
			)

end

--------------------------------------------------------------------------------
-- requestProxyForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestProxyForExport( pixStreamId,
															localAssetId )

	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists, 
			'proxy',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedDownloadProxyInfo
			)

end

--------------------------------------------------------------------------------
-- requestXmpDevelopForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestXmpDevelopForExport( pixStreamId, localAssetId )

	return addOrUpdateRequest(
			self,
			pixStreamId,
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists,
			'xmp_develop',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedXmpDevelopInfo
			)

end

--------------------------------------------------------------------------------
-- requestXmpDevelopForDevelop
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestXmpDevelopForDevelop( pixStreamId, localAssetId, needsASAP )

	local requestPriority = 
			needsASAP and PixelPathProviderRequestTypes.names.highest or PixelPathProviderRequestTypes.names.download_soon

	return addOrUpdateRequest(
			self,
			pixStreamId,
			localAssetId,
			requestPriority,
			'xmp_develop',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedXmpDevelopInfo
			)

end

--------------------------------------------------------------------------------
-- requestXmpSidecarForExport
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestXmpSidecarForExport( pixStreamId, localAssetId )

	return addOrUpdateRequest(
			self,
			pixStreamId,
			localAssetId,
			PixelPathProviderRequestTypes.names.ensure_local_original_exists,
			'xmp_sidecar',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedXmpSidecarInfo
			)

end

--------------------------------------------------------------------------------
-- requestBinaryWhenIdle
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:requestBinaryWhenIdle( 
												pixStreamId, localAssetId,
												binaryType, binarySize,
												isMarkedForOffline, hasLocalOriginal )

	local initialState =
	
			( binaryType == 'original' and
				PixelPathProviderStateMachine.DocstoreInfoStates.NeedDownloadOriginalInfo ) or
	
			( binaryType == 'proxy' and
				PixelPathProviderStateMachine.DocstoreInfoStates.NeedDownloadProxyInfo ) or

			( binaryType == 'xmp_develop' and
				PixelPathProviderStateMachine.DocstoreInfoStates.NeedXmpDevelopInfo ) or

			( binaryType == 'xmp_sidecar' and
				PixelPathProviderStateMachine.DocstoreInfoStates.NeedXmpSidecarInfo ) or

			nil
			
	local requestPriority = 
			isMarkedForOffline and PixelPathProviderRequestTypes.names.ensure_local_original_exists or
			hasLocalOriginal and PixelPathProviderRequestTypes.names.lowest or
			PixelPathProviderRequestTypes.names.idle
			
	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			requestPriority,
			binarySize,
			initialState )

end

--------------------------------------------------------------------------------
-- getBestPossibleLocalNegativePathASAP
--------------------------------------------------------------------------------

local function makeASAPNegativePathRequest( self, pixStreamId, localAssetId,
												negativeType,
												requestInitialState )


	return addOrUpdateRequest( 
			self,
			pixStreamId, 
			localAssetId,
			PixelPathProviderRequestTypes.names.highest,
			negativeType,
			requestInitialState )

end

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

function PixelPathProviderSchedulerPrototype:getBestPossibleLocalNegativePathASAP( 
												pixStreamId, localAssetId )

	return makeASAPNegativePathRequest( 
			self, pixStreamId, localAssetId, 'master',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedBestPossibleLocalNegativePathInfo )

end

--------------------------------------------------------------------------------
-- getLocalNegativePathASAP
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:getLocalNegativePathASAP( 
												pixStreamId, localAssetId )

	return makeASAPNegativePathRequest( 
			self, pixStreamId, localAssetId, 'any',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedAnyLocalNegativePathInfo )

end

--------------------------------------------------------------------------------
-- batchPrefetchNegativePaths
--------------------------------------------------------------------------------

local function notInCurrentSet( assetId, currAssetIdsSet )

	if not currAssetIdsSet[ assetId ] then

		return assetId

	end

end

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

local function addOrUpdateRequestForAssetId( localAssetId, self, pixStreamId,
											requestPriorityName, longSide,
											maybeOverrideInitialState )

	return addOrUpdateRequest( self, pixStreamId, localAssetId,
								requestPriorityName, longSide,
								maybeOverrideInitialState )

end

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

local function batchPrefetchNegativePaths( self, priorityHint, pixStreamId,
											assetIds )

	local requestPriority = 
			priorityHint == "soon" and
			PixelPathProviderRequestTypes.names.download_soon or
			PixelPathProviderRequestTypes.names.grid_prefetch

	WFFunctionUtils_map( 
		addOrUpdateRequestForAssetId, 
			assetIds, self, pixStreamId, requestPriority, 'any',
			PixelPathProviderStateMachine.DocstoreInfoStates.NeedAnyLocalNegativePathInfo )

end

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

batchPrefetchNegativePaths =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		'batchPrefetchNegativePaths',
		batchPrefetchNegativePaths )

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

function PixelPathProviderSchedulerPrototype:prefetchNegativePaths(
										priorityHint, pixStreamId,
										maybePrevAssetIds, assetIds )

	if #assetIds == 0 then

		self:abandonAllPixStreamRequests( pixStreamId )

		return

	end

	local prevAssetIds = maybePrevAssetIds or {}

	local currAssetIdsSet = KSTableUtils.mapArrayToSet( assetIds )

	local assetIdsToRemove = WFFunctionUtils_filter( 
								notInCurrentSet, prevAssetIds, currAssetIdsSet )

	self:batchRemoveRequests( pixStreamId, assetIdsToRemove )

	batchPrefetchNegativePaths( self, priorityHint, pixStreamId, assetIds )

end

--------------------------------------------------------------------------------
-- updateLongSideOnAll
--------------------------------------------------------------------------------

local function applyNewLongSideTransition( request, self )

	self:_applyTransitionToRequest( request, 'new_long_side' )

end

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

function PixelPathProviderSchedulerPrototype:updateLongSideOnAll( pixStreamId,
																	longSide )

	WFArguments_check( "PixelPathProviderScheduler", "number|string" )

	local udpatedRequests =
		self._requestPriorityQueue:updateLongSideOnAll( pixStreamId, longSide )

	WFFunctionUtils_map(
		applyNewLongSideTransition, 
			udpatedRequests, self )

end

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

PixelPathProviderSchedulerPrototype.updateLongSideOnAll =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"updateLongSideOnAll",
		PixelPathProviderSchedulerPrototype.updateLongSideOnAll )

--------------------------------------------------------------------------------
-- batchSetPriority
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.setPriority =
	forwardToPriorityQueue( 'setPriority' )

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

local function setPriorityForLocalAssetId( localAssetId, self, pixStreamId, 
								newPriorityName )

	self:setPriority( pixStreamId, localAssetId, newPriorityName )

end

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

function PixelPathProviderSchedulerPrototype:batchSetPriority( 
												pixStreamId, assetIds, 
												newPriorityName )

	WFArguments_check( "PixelPathProviderScheduler", "number|string", "table",
							"string" )

	WFFunctionUtils_map(
		setPriorityForLocalAssetId, 
			assetIds, self, pixStreamId, newPriorityName )

end

--------------------------------------------------------------------------------
-- updatePriorityOnDownloadingRequestsOrRemove
--------------------------------------------------------------------------------

local function updatePriorityOnDownloadingRequestOrRemove( 
							localAssetId, requestPriorityQueue, pixStreamId,
							pixStreamLocalAssetIdsWithActiveDownload,
							newPriorityName )

	if not pixStreamLocalAssetIdsWithActiveDownload[ localAssetId ] then

		requestPriorityQueue:removeRequest( pixStreamId, localAssetId )

	else

		requestPriorityQueue:setPriority( pixStreamId, localAssetId,
											newPriorityName )

	end

end

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

local function maybeAddRequestAssetIdToSet( set, request, wantedPixstreamId )

	local pixStreamId, localAssetId = unpack( request )

	if wantedPixstreamId == pixStreamId then

		set[ localAssetId ] = true

	end

	return set

end

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

local function updatePriorityOnDownloadingRequestsOrRemove( self, pixStreamId,
															localAssetIds,
															newPriorityName )

	WFArguments_check( "PixelPathProviderScheduler", "string", "table", 
						"string" )

	-- we expect there to only be about a couple dozen downloads at most. In 
	-- a fully synced catalog, there often be no downloads at all. On the other
	-- hand, it's possible there will 100+ localAssetIds if doing scrubby
	-- scrolling.
	--
	-- So it will almost always be more efficent to do a query to find out 
	-- which requests are donwloading and use that list to determine if
	-- any priorities need adjustment.
	local downloadingBlobRequests =
			self._requestPriorityQueue:getRequestsWithStatesInPriorityOrder( 
				unpack( PixelProviderSharedBlobCache.downloadingBlobStates ) )

	local pixStreamLocalAssetIdsWithActiveDownload =
			WFFunctionUtils_reduce( {},
				maybeAddRequestAssetIdToSet,
					downloadingBlobRequests, pixStreamId )

	WFFunctionUtils_map( updatePriorityOnDownloadingRequestOrRemove,
							localAssetIds,
							self._requestPriorityQueue, pixStreamId,
							pixStreamLocalAssetIdsWithActiveDownload,
							newPriorityName )

end

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

PixelPathProviderSchedulerPrototype.updatePriorityOnDownloadingRequestsOrRemove =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"updatePriorityOnDownloadingRequestsOrRemove",
		updatePriorityOnDownloadingRequestsOrRemove )

--------------------------------------------------------------------------------
-- createRenditionRequestOrUpdatePriorityForLocalAssetIdIndex
--------------------------------------------------------------------------------

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

local math_max = math.max

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

local WFCatRenderSize_defaultRenderSizes =
		assert( type( WFCatRenderSize.defaultRenderSizes ) == "table" and
						WFCatRenderSize.defaultRenderSizes,
					"WFCatRenderSize.defaultRenderSizes table is missing" )

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

local _smallestDefaultRenderSize = 
		assert( WFCatRenderSize_defaultRenderSizes[ 1 ],
				"WFCatRenderSize.defaultRenderSizes is empty!" )

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

local _largestDefaultRenderSize = 
		assert( WFCatRenderSize_defaultRenderSizes[ #WFCatRenderSize_defaultRenderSizes ],
				"WFCatRenderSize.defaultRenderSizes is empty!" )

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

local function getRenderSizeForCellNWithDefault( 
				cellSizeData, i, defaultRenderSize )

	local offset = 2 * ( i - 1 )

	local w = tonumber( cellSizeData[ offset + 1 ] ) or -1

	local h = tonumber( cellSizeData[ offset + 2 ] ) or -1

	local renderSize = math_max( w, h )

	if renderSize <= 0 then

		return defaultRenderSize

	end

	return renderSize

end

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

local function normalizeRenderSize( targetRenderSize )

	if targetRenderSize <= _smallestDefaultRenderSize then

		return _smallestDefaultRenderSize

	end

	if targetRenderSize >= _largestDefaultRenderSize then

		return _largestDefaultRenderSize

	end

	local currBiggestRenderSizeSeen = _smallestDefaultRenderSize

	for _, currDefaultRenderSize in ipairs( WFCatRenderSize_defaultRenderSizes ) do

			currBiggestRenderSizeSeen = 
				assert( 
					currBiggestRenderSizeSeen <= currDefaultRenderSize and
					currDefaultRenderSize,
					"default render size sequence cannot be descending" )

		if currBiggestRenderSizeSeen >= targetRenderSize then

			return currDefaultRenderSize

		end

	end

	return _largestDefaultRenderSize

end

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

local function getNormalizedRenderSize( cellSizeData, i, defaultRenderSize )

	return normalizeRenderSize( 

			getRenderSizeForCellNWithDefault( 
				cellSizeData, i, defaultRenderSize )

			)

end

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

local function createRenditionRequestOrUpdatePriorityForLocalAssetIdIndex( 
					requestData, 
					self, 
					localAssetIds,
					cellSizeData, pixStreamId,
					defaultRenderSize, priorityName,
					maybeOverrideInitialState )

	local action, localAssetIdIndex = unpack( requestData )

	local localAssetId = localAssetIds[ localAssetIdIndex ]

	if not localAssetId then

		logger:warnf( "no localAsset id at index %s of '%s'", localAssetIdIndex,
						pixStreamId )

		return

	end

	if action ~= "create" then

		self:setPriority( pixStreamId, localAssetId, priorityName )

		return

	end

	local renderSize = 
			getNormalizedRenderSize( 
				cellSizeData, localAssetIdIndex,
				defaultRenderSize )

	addOrUpdateRequest( 
		self, pixStreamId, localAssetId, priorityName, 
		renderSize, maybeOverrideInitialState )

end

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

local function createRenditionRequestOrUpdatePriorityImp( 
					self,
					pixStreamId, 
					assetIdIndexesToCreateOrUpdate,
					assetIds, cellSizeData,
					priorityName,
					maybeOverrideInitialState )

	WFArguments_check( "PixelPathProviderScheduler", 
							"string", 
							"table",
							"table", "table", 
							"string", "?string" )

	local firstCellBasedDefaultRenderSize = 
			getRenderSizeForCellNWithDefault( 
				cellSizeData, 1, _smallestDefaultRenderSize )

	WFFunctionUtils_map(
		createRenditionRequestOrUpdatePriorityForLocalAssetIdIndex, 
			assetIdIndexesToCreateOrUpdate, 
				self, assetIds, cellSizeData, pixStreamId,
				firstCellBasedDefaultRenderSize, priorityName,
				maybeOverrideInitialState )

end

--------------------------------------------------------------------------------
-- createLoupeRenditionRequestOrUpdatePriority
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:createLoupeRenditionRequestOrUpdatePriority( 
												pixStreamId, 
												assetIdIndexesToCreateOrUpdate,
												assetIds, cellSizeData,
												priorityName )

	return createRenditionRequestOrUpdatePriorityImp( 
				self,
				pixStreamId, 
				assetIdIndexesToCreateOrUpdate, 
				assetIds, cellSizeData, 
				priorityName, 
				PixelPathProviderStateMachine.SendDataStates.MaybeSendLoupeRenditionPathFromInMemoryCache )

end

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

function PixelPathProviderSchedulerPrototype:createThumbnailRenditionRequestOrUpdatePriority(
					pixStreamId, 
					assetIdIndexesToCreateOrUpdate,
					assetIds, cellSizeData,
					priorityName )

	return createRenditionRequestOrUpdatePriorityImp(
				self,
				pixStreamId, 
				assetIdIndexesToCreateOrUpdate,
				assetIds, cellSizeData,
				priorityName )

end

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

PixelPathProviderSchedulerPrototype.createThumbnailRenditionRequestOrUpdatePriority =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"createThumbnailRenditionRequestOrUpdatePriority",
		PixelPathProviderSchedulerPrototype.createThumbnailRenditionRequestOrUpdatePriority )

--------------------------------------------------------------------------------
-- batchAddRequests
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:batchAddRequests( pixStreamId,
																localAssetIds,
																requestTypeName,
																longSide )

	WFArguments_check( "PixelPathProviderScheduler",
						"string", "table", "string", "number|string" )

	return WFFunctionUtils_map( addOrUpdateRequestForAssetId, 
								localAssetIds, self, pixStreamId,
								requestTypeName, longSide )

end

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

PixelPathProviderSchedulerPrototype.batchAddRequests =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"batchAddRequests",
		PixelPathProviderSchedulerPrototype.batchAddRequests )

--------------------------------------------------------------------------------
-- removeRequest
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.removeRequest =
	forwardToPriorityQueue( 'removeRequest' )

--------------------------------------------------------------------------------
-- batchRemoveRequests
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.priorityQueueBatchRemoveRequests =
	forwardToPriorityQueue( 'batchRemoveRequests' )

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

function PixelPathProviderSchedulerPrototype:batchRemoveRequests( 
												pixStreamId, localAssetIds )

	WFArguments_check( "PixelPathProviderScheduler", "string", "table" )

	local chunkSize = 64
	local maxIndex = #localAssetIds
	for chunkStart = 1, maxIndex, chunkSize do

		local batchOfIds = {}
		for i = chunkStart, math.min( maxIndex, chunkStart + chunkSize - 1 ) do
			batchOfIds[ #batchOfIds + 1 ] = localAssetIds[ i ]
		end

		self:priorityQueueBatchRemoveRequests( pixStreamId, batchOfIds )

	end

end

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

PixelPathProviderSchedulerPrototype.batchRemoveRequests =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"batchRemoveRequests",
		PixelPathProviderSchedulerPrototype.batchRemoveRequests )

--------------------------------------------------------------------------------
-- setRequestPriorityToNewPriority
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.setRequestPriorityToNewPriority =
	forwardToPriorityQueue( 'setRequestPriorityToNewPriority' )

--------------------------------------------------------------------------------
-- abandonAllPixStreamRequests
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.abandonAllPixStreamRequests =
	forwardToPriorityQueue( 'abandonAllPixStreamRequests' )

--------------------------------------------------------------------------------
-- abandonAllLocalAssetIdRequests
--------------------------------------------------------------------------------

PixelPathProviderSchedulerPrototype.abandonAllLocalAssetIdRequests =
	forwardToPriorityQueue( 'abandonAllLocalAssetIdRequests' )

--------------------------------------------------------------------------------
-- pixelPathStream is an Rx.Observable of:
--
-- 			pixStreamId, localAssetId, success, pixelPathOrError, maybeDigest,
-- 			maybeRenderSize, isCurrent
--

function PixelPathProviderSchedulerPrototype:pixelPathStream()

	WFArguments_check( "PixelPathProviderScheduler" )

	return self._anyPixStreamIdPixelPathStream

end

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

function PixelPathProviderSchedulerPrototype:pixStreamIdPixelPathStream( pixStreamId )

	WFArguments_check( "PixelPathProviderScheduler", "string" )

	return self
			:pixelPathStream()
			:filter( 
				function( thisPixStreamId )
					
					return thisPixStreamId == pixStreamId

				end )

end

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

function PixelPathProviderSchedulerPrototype:pixStreamIdsPixelPathStream( pixStreamIds )

	WFArguments_check( "PixelPathProviderScheduler", "table" )

	return self
			:pixelPathStream()
			:filter( 
				function( thisPixStreamId )
					
					return pixStreamIds[ thisPixStreamId ] ~= nil

				end )

end

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

function PixelPathProviderSchedulerPrototype:containsPixStreamIdSubstrPixelPathStream( pixStreamIdSubStr )

	WFArguments_check( "PixelPathProviderScheduler", "string" )

	return self
			:pixelPathStream()
			:filter( 
				function( thisPixStreamId )

					return string.find( thisPixStreamId, 
											pixStreamIdSubStr ) ~= nil

				end )

end

--------------------------------------------------------------------------------
-- bufferedPixelPathStream
--------------------------------------------------------------------------------

local function isValidFPS( fps )

	return type( fps ) == "number" and fps > 0 and 
				math.floor( fps ) == math.ceil( fps )

end

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

local function makeBufferedFPSStream( stream, fps )

	if not isValidFPS( fps ) then

		-- return a 1 element array
		return stream
				:map( 
					function( ... )

						return { { ... } }

					end )

	end

	local fpsAsSeconds = 1 / fps
	
	local scheduler = require 'WFRxScheduler'.sharedScheduler()

	local activeBuffer = {}

	return stream	
			:map(
				function( ... )

					activeBuffer[ #activeBuffer + 1 ] = { ... }

				end )
			:calm( fpsAsSeconds, scheduler )
			:map(
				function()

					local thisIntervalBuffer = activeBuffer

					activeBuffer = {}

					return thisIntervalBuffer

				end )

end

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

local function recordSentPathCount( weakSelf, bufferedPaths )

	local localSelf = derefWeakRef( weakSelf )

	if not localSelf then

		return bufferedPaths

	end

	rawset( localSelf, "_DEBUG_daemonBufferedPathsCallWorker",
				( rawget( localSelf, "_DEBUG_daemonBufferedPathsCallWorker" ) or 
					0 ) + 1 )

	rawset( localSelf, "_DEBUG_daemonBufferedPathsSentWorker",
				( rawget( localSelf, "_DEBUG_daemonBufferedPathsSentWorker" ) or 
					0 ) + #bufferedPaths )

	bufferedPaths._DEBUG_daemonBufferedPathsCallWorker =
					rawget( localSelf, "_DEBUG_daemonBufferedPathsCallWorker" )

	bufferedPaths._DEBUG_daemonBufferedPathsSentWorker =
					rawget( localSelf, "_DEBUG_daemonBufferedPathsSentWorker" )

	return bufferedPaths
	
end 

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

function PixelPathProviderSchedulerPrototype:bufferedPixelPathStream(
												fps )

	WFArguments_check( "PixelPathProviderScheduler" )

	local weakSelf = makeWeakRef( self )

	return makeBufferedFPSStream( self._anyPixStreamIdPixelPathStream, fps )
			:map( 
				function( bufferedPaths )
				
					return recordSentPathCount( weakSelf, bufferedPaths )

				end )

end

--------------------------------------------------------------------------------
-- _DEBUG_requestStatsStream
--------------------------------------------------------------------------------

local function initRequestStatsStream( self )
	
	self._DEBUG_requestStatsStreamSubscriberCount = 0

	self._DEBUG_requestStatsStreamData = Rx.Subject()

	return Rx.Observable.create(

			function( observer )

				self:_DEBUG_enableDetailedPixelProviderStats()

				self._DEBUG_requestStatsStreamSubscriberCount =
					 self._DEBUG_requestStatsStreamSubscriberCount + 1

				local subscription = 
					self._DEBUG_requestStatsStreamData:subscribe( observer )

				return Rx.Disposable.create(

					function()

						self._DEBUG_requestStatsStreamSubscriberCount =
							 self._DEBUG_requestStatsStreamSubscriberCount - 1

						subscription:dispose()

					end )
			
			end )

end

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

local function sortByCountDesc( throughput1, throughput2 )
	
	return throughput1[ 3 ] > throughput2[ 3 ]

end

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

function PixelPathProviderSchedulerPrototype:_maybeSendRequestStats()

	WFArguments_check( "PixelPathProviderScheduler" )

	if not self._DEBUG_requestStatsStreamSubscriberCount or
			self._DEBUG_requestStatsStreamSubscriberCount == 0 then

		return

	end

	local currChangeCounter = 
		PixelPathProviderSchedulerPrivateMethods.getStateChangeCounter( self )

	if self._DEBUG_requestStatsChangeCounter == currChangeCounter then

		-- no changes
		return

	end

	self._DEBUG_requestStatsChangeCounter = currChangeCounter


	local sentPerPixStream = rawget( self, "_DEBUG_pixelPathsSentPerPixStream" ) or {}

	local durations = self._requestPriorityQueue:_DEBUG_getPixStreamActiveRequestDurations()

	local pixStreamThroughput = {}

	for pixStreamId, count in pairs( sentPerPixStream ) do

		local duration = durations[ pixStreamId ]

		if duration then

			pixStreamThroughput[ #pixStreamThroughput + 1 ] =
				{ pixStreamId, count/duration, count, duration }

		end

	end

	table.sort( pixStreamThroughput, sortByCountDesc )

	local requestStateBenchmarks = 
			PixelPathProviderSchedulerDiagnosticLog.processRawRequestStateBenchmarks(
				self._requestPriorityQueue:_DEBUG_getRequestStateBenchmarks() )

	self._DEBUG_requestStatsStreamData:onNext(
		{ self._requestPriorityQueue:_DEBUG_getRequestCountsByState(),
			self._requestPriorityQueue:_DEBUG_getRequestCountsByPriority(),
			pixStreamThroughput, requestStateBenchmarks } )

end

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

function PixelPathProviderSchedulerPrototype:_DEBUG_enableDetailedPixelProviderStats()

	WFArguments_check( "PixelPathProviderScheduler" )

	self._requestPriorityQueue:_DEBUG_enableRequestStateBenchmarking(
		WFFiniteStateMachine.getStates( self._stateMachine ) )

end

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

function PixelPathProviderSchedulerPrototype:_DEBUG_requestStatsStream()

	WFArguments_check( "PixelPathProviderScheduler" )

	self._DEBUG_requestStatsObservable = 
		self._DEBUG_requestStatsObservable or initRequestStatsStream( self )

	return self._DEBUG_requestStatsObservable

end

--------------------------------------------------------------------------------
-- _DEBUG_localAssetIdStatesStream
--------------------------------------------------------------------------------

local function maybeSendLocalAssetIdStates( self, localAssetId )

	if not self._DEBUG_localAssetIdStatesStreams then
		
		return

	end

	local localAssetIdStatesStream = 
			self._DEBUG_localAssetIdStatesStreams[ localAssetId ]

	if not localAssetIdStatesStream then

		return

	end

	local localAssetIdStatesChangeCounter =
			self._DEBUG_localAssetIdStatesChangeCounters[ localAssetId ] or 0

	self._DEBUG_localAssetIdStatesChangeCounters[ localAssetId ] =
		PixelPathProviderSchedulerPrivateMethods.getStateChangeCounter( self )

	
	local pixStreamIdStatePairs = 
			self._requestPriorityQueue:_DEBUG_getPixStreamIdStatePairs( 
										localAssetId,
										localAssetIdStatesChangeCounter )

	if not next( pixStreamIdStatePairs ) then

		return

	end

	localAssetIdStatesStream:onNext( pixStreamIdStatePairs )

end

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

function PixelPathProviderSchedulerPrototype:_DEBUG_maybeSendLocalAssetIdsStates()

	for localAssetId in pairs( self._DEBUG_localAssetIdStatesStreams or {} ) do

		maybeSendLocalAssetIdStates( self, localAssetId )

	end

end

--------------------------------------------------------------------------------
-- _DEBUG_localAssetIdStatesStream
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:_DEBUG_localAssetIdStatesStream( 
															localAssetId )

	WFArguments_check( "PixelPathProviderScheduler", "number" )

	self._DEBUG_localAssetIdWatchCounts = self._DEBUG_localAssetIdWatchCounts or {}

	self._DEBUG_localAssetIdStatesStreams = self._DEBUG_localAssetIdStatesStreams or {}

	self._DEBUG_localAssetIdStatesChangeCounters = 
		self._DEBUG_localAssetIdStatesChangeCounters or {}

	return Rx.Observable.create(

			function( observer )

				self._DEBUG_localAssetIdWatchCounts[ localAssetId ] =
					( self._DEBUG_localAssetIdWatchCounts[ localAssetId ] or 0 ) + 1

				self._DEBUG_localAssetIdStatesStreams[ localAssetId ] =
					self._DEBUG_localAssetIdStatesStreams[ localAssetId ] or 
					Rx.Subject()

				self._DEBUG_localAssetIdStatesChangeCounters[ localAssetId ] = 0

				local subscription = 
					self._DEBUG_localAssetIdStatesStreams[ localAssetId ]:subscribe( observer )

				maybeSendLocalAssetIdStates( self, localAssetId )

				return Rx.Disposable.create(

					function()

						self._DEBUG_localAssetIdWatchCounts[ localAssetId ] =
							math.max( ( self._DEBUG_localAssetIdWatchCounts[ localAssetId ] or 0 ) - 1, 0 )

						if self._DEBUG_localAssetIdWatchCounts[ localAssetId ] == 0 then
						
							self._DEBUG_localAssetIdWatchCounts[ localAssetId ] = nil

							self._DEBUG_localAssetIdStatesStreams[ localAssetId ] = nil

							self._DEBUG_localAssetIdStatesChangeCounters[ localAssetId ] = nil

						end

						subscription:dispose()

					end )
			
			end )


end

--------------------------------------------------------------------------------
-- updateCellSizes
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:updateCellSizes( pixStreamId, 
																assetIdIndexes,
																cellSizeData )

	local firstCellBasedDefaultRenderSize = 
			getRenderSizeForCellNWithDefault( 
				cellSizeData, 1, _smallestDefaultRenderSize )

	local assetIds = 
			self._requestPriorityQueue:getLocalAssetIdsForPixstreamId(
											pixStreamId )

	for _, assetId in ipairs( assetIds ) do

		local assetIdIndex = assetIdIndexes[ assetId ] or 1

		local longSide = getNormalizedRenderSize( 
							cellSizeData, 
							assetIdIndex,
							firstCellBasedDefaultRenderSize )

		maybeUpdateLongSide( self, pixStreamId, assetId, longSide )

	end

end

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

PixelPathProviderSchedulerPrototype.updateCellSizes =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests( 
		"updateCellSizes",
		PixelPathProviderSchedulerPrototype.updateCellSizes )

--------------------------------------------------------------------------------
-- get/setDaemonDownloadStatus
--------------------------------------------------------------------------------

function PixelPathProviderSchedulerPrototype:setDaemonDownloadStatus( daemonDownloadStatus )

	self._daemonDownloadStatus = daemonDownloadStatus

end

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

function PixelPathProviderSchedulerPrototype:getDaemonDownloadStatus()

	return self._daemonDownloadStatus

end

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

PixelPathProviderSchedulerPrototype.class = {

		privateUnitTestExports = {

			getNormalizedRenderSize = getNormalizedRenderSize,

		}

}

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

local PixelPathProviderScheduler =
		Rx.Class( "PixelPathProviderScheduler",
					PixelPathProviderSchedulerPrototype )

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

return PixelPathProviderScheduler
