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

PixelProviderProcessor.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 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 WFFunctionUtils = require 'WFFunctionUtils'

	-- WF.core.lua.threading
local WFWorkerThread = require 'WFWorkerThread'

	-- WF.web.json
local KSJsonEncoder = require 'KSJsonEncoder'

	--
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- Rx
local Rx = require "Rx"

-- In case you need to turn off pixel provider processing
-- this is the config to look for. If you set this ( in
-- daemon config.lua ), then no pixel provider request 
-- will be processed. No rendition, original or proxy
-- will get downloaded. Even the idle requests wont 
-- be processed.
-- Grid will just show empty squares.
local DISABLE_PIXEL_PROVIDER = require "WFConfiguration".PixelProvider.disable

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

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

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

local WFFunctionUtils_map = WFFunctionUtils.map

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

local function makeWeakRef( v )

	local weakRef = KSTableUtils.makeWithMode( "kv" )

	weakRef[ v ] = v

	return weakRef

end

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

local function derefWeakRef( weakRef )

	return next( weakRef )

end

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

local function returnSame( ... )

	return ...

end

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

local function makeWorkerThreadRequestStreamSubject( self )

	local subject = Rx.Subject()

	self._workerThreadRelaySubjects[ subject ] = subject

	return subject

end

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

local function cleanupWorkerThreadRequestStreamSubject( self, subject )

	if not subject or 
		not self._workerThreadRelaySubjects[ subject ] then

		return

	end

	-- stop remembering the subject
	self._workerThreadRelaySubjects[ subject ] = nil

	-- Sending Rx.Observable.empty() ensures that any subscription to the 
	-- the worker thread request observable is disposed. In addition, it
	-- won't sent any further values.

	subject:onNext( Rx.Observable.empty() )

end

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

local function relayWorkerStream( self, workerThreadVarName,
									workerThreadMethodName, ... )

	local weakSelf = makeWeakRef( self )

	local theArgs = { ... }

	local nArgs = select( '#', ... )

	return Rx.Observable.create( 

		function( observer )
			
			local currSelf = derefWeakRef( weakSelf )

			if not currSelf then

				observer:onCompleted()

				return

			end

			-- We want to relay an 'inifinite' stream of values from a worker thread
			-- universe via a worker thread request observable.
			-- 
			-- As long as there is a subscriber to the worker thread observable,
			-- the underlying worker thread will not close.
			--
			-- When self closes in this universe, we want to close everything that
			-- keeps the worker thread alive. So we need to close all subscriptions
			-- to the worker thread requests and therefor need to retain control of
			-- the worker thread subscription in this object.
			--
			-- We do this via an Rx.Subject and flatMapLastest. If an observer
			-- subscribes to subject:flatMapLatest(), it will only subscribe to the
			-- most recent observable sent to the subject via :oNext. Each
			-- subsequent call to :onNext will start a new subsciption and dispose
			-- the previous subscription in the stream.
			--
			-- So if we call subject:onNext( Rx.Observable.empty() ), we'll clean
			-- up the subscription to the worker thread request.

			local workerThreadRequestStreamSubject = 
					makeWorkerThreadRequestStreamSubject( currSelf )

			local managedWorkerThreadRequestStream = 
					workerThreadRequestStreamSubject
						:flatMapLatest( returnSame )

			local subscription = managedWorkerThreadRequestStream:subscribe( observer )

			local workerThreadHandle = currSelf[ workerThreadVarName ]
			-- This call starts sending the worker thread request values via
			-- managedWorkerThreadRequestStream
			workerThreadRequestStreamSubject:onNext( 
				workerThreadHandle:request( workerThreadMethodName,
											unpack( theArgs, 1, nArgs ) ) )

			return Rx.Disposable.create( 

				function()

					subscription:dispose()

					local thisCurrSelf = derefWeakRef( weakSelf )

					if thisCurrSelf then

						cleanupWorkerThreadRequestStreamSubject( 
							thisCurrSelf, workerThreadRequestStreamSubject )

					end

				end )

		end

	)

end

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

local function logPurgeResult( purgeResult )

	local ok, path, maybeErr = unpack( type( purgeResult ) == "table" and
										purgeResult or {} )

	if not ok then

		logger:warnf( "Failed to deleted rendition path %s because: %s", path, maybeErr )

		return

	end

	logger:infof( "Successfully deleted rendition path: %s", path )

end

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

local function maybeLogPurgeResults( self )

	if not logger.will_warn then

		return

	end

	local purgeResultsStream =
		relayWorkerStream( self, "_renditionCachePurgeWorker", 
							"purgeResultsStream" )

	self._renditionCachePurgeWorkerResultsSubscription = 
		purgeResultsStream:
			subscribe(
				function( purgeResults )

					WFFunctionUtils_map( logPurgeResult, purgeResults )

				end )

end

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

local PixelProviderProcessor = {

		--( string )
	_schedulerWorkerThreadName = false,

	_schedulerWorker = false,
		-- (WFWorkerThread)
	
	_workerThreadRelaySubjects = false,

	_renditionCachePurgeWorkerName = false,

	_renditionCachePurgeWorker = false,
		-- (WFWorkerThread)

	_renditionCachePurgeWorkerResultsSubscription = false,

	WFCloseSupport.addCloseSupportMethods,

}

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

function PixelProviderProcessor:__init( wfCatUser, wfCatSpec,
											rootCatalogDataDir, 
											catalogOptions,
											pixelProviderSettings )

	local workerThreadArgs = KSJsonEncoder.encodeOzUploadJson( 
								{ wfCatUser:serverBaseUrl(),
									wfCatUser:userId(),
									wfCatUser:userInfoCacheFolder(),
									wfCatSpec:catalogDoc() or wfCatSpec:catalogId(),
									rootCatalogDataDir, 
									catalogOptions,
									pixelProviderSettings
								} )

	self._schedulerWorkerThreadName = "PixelProviderSchedulerWorker.lua"

	self._workerThreadRelaySubjects = {}

	self._renditionCachePurgeWorkerName = "PixelProviderFileCachePurgeWorker.lua"

	self:setOwned {

		_schedulerWorker = WFWorkerThread( self._schedulerWorkerThreadName,
											workerThreadArgs ),

		_renditionCachePurgeWorker = 
			WFWorkerThread( self._renditionCachePurgeWorkerName,
								workerThreadArgs )

	}

	maybeLogPurgeResults( self )

end

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

function PixelProviderProcessor:close()

	logger:infof( "PixelProviderProcessor named '%s' is closing", self._schedulerWorkerThreadName )

	for subject in pairs( self._workerThreadRelaySubjects ) do

		logger:infof( "PixelProviderProcessor named '%s' is closing a relayed stream", self._schedulerWorkerThreadName )

		cleanupWorkerThreadRequestStreamSubject( self, subject )

	end

	if self._renditionCachePurgeWorkerResultsSubscription then

		self._renditionCachePurgeWorkerResultsSubscription:dispose()

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes the given PixelProvider function
	in a worker thread. This function should be used
	when you don't care about the return value.
	The stream will be alive till it is done.

	\param name (string) Name of the method to be executed
	\param Arguments to be passed to the function with
			the name `Name`

	\return Observable representing the return value
]]--
function PixelProviderProcessor:runUntilDone( name, ... )

	WFArguments_check( 'PixelProviderProcessor', 'string' )
	
	if WFCloseSupport.hasBeenOrWillBeClosed( self._schedulerWorker ) then
		return Rx.Observable.just()
	end
	
	return WFRxRunUntilDone( self._schedulerWorker:request( 'dispatch', name, ... ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes the given PixelPathProviderScheduler method in a worker
	thread. This function should be used when you want to manage the lifetime
	of the worker thread request

	\param name (string) Name of the method to be executed
	\param Arguments to be passed to the function with
			the name `Name`

	\return Observable representing a worker thread request
]]--
function PixelProviderProcessor:dispatch( name, ... )

	WFArguments_check( 'PixelProviderProcessor', 'string' )
	
	if WFCloseSupport.hasBeenOrWillBeClosed( self._schedulerWorker ) then

		return Rx.Observable.empty()

	end
	
	return self._schedulerWorker:request( 'dispatch', name, ... )

end

--------------------------------------------------------------------------------
--[[!
	\brief Executes the given PixelProvider function
	in a worker thread. This function should be used
	when you don't care about the return value.
	The stream will be alive till it is done.

	\param name (string) Name of the method to be executed
	\param Arguments to be passed to the function with
			the name `Name`

	\return Observable representing the return value
]]--
function PixelProviderProcessor:scanAndPurge()

	WFArguments_check( 'PixelProviderProcessor' )

	return self._renditionCachePurgeWorker:request( 'scanAndPurge' )

end

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

function PixelProviderProcessor:purgeRenditionsForDeletedAssets( assetIds )

	WFArguments_check( 'PixelProviderProcessor', 'table' )

	return self._renditionCachePurgeWorker:request( 'purgeRenditionsForDeletedAssets', assetIds )

end

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

function PixelProviderProcessor:updateAssetIdList( pixStreamId, prevAssetIds,
													newAssetIds )

	WFArguments_check( 'PixelProviderProcessor' )

	return WFRxRunUntilDone( self._schedulerWorker:request( 
								'updateAssetIdList', 
									pixStreamId, prevAssetIds, newAssetIds ) )

end

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

function PixelProviderProcessor:updateCellSizes( pixStreamId, assetIds,
													cellSizeData )

	WFArguments_check( 'PixelProviderProcessor' )

	return WFRxRunUntilDone( self._schedulerWorker:request( 
								'updateCellSizes', 
									pixStreamId, assetIds, cellSizeData ) )

end

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

function PixelProviderProcessor:updateForGridScrolling( 
				pixStreamId, assetIds, cellSizeData,
				prevOnScreenRange, onScreenRange,
				prevBufferedRange, bufferedRange,
				scrollDirection, 
				prevFetchAheadRange, fetchAheadRange,
				prevFetchBehindRange, fetchBehindRange)


	WFArguments_check( 'PixelProviderProcessor' )

	return WFRxRunUntilDone( self._schedulerWorker:request( 
								'updateForGridScrolling', 
									pixStreamId, assetIds,
									cellSizeData,
									prevOnScreenRange, onScreenRange,
									prevBufferedRange, bufferedRange,
									scrollDirection, 
									prevFetchAheadRange, fetchAheadRange,
									prevFetchBehindRange, fetchBehindRange ) )

end

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

function PixelProviderProcessor:updateForLoupeScrolling( 
				pixStreamId, assetIds, defaultLoupeRenditionSizeData,
				prevOnScreenRange, onScreenRange,
				prevBufferedRange, bufferedRange,
				scrollDirection, 
				prevFetchAheadRange, fetchAheadRange,
				prevFetchBehindRange, fetchBehindRange)

	WFArguments_check( 'PixelProviderProcessor' )

	defaultLoupeRenditionSizeData =
		( type( defaultLoupeRenditionSizeData ) == "number" and
			{ defaultLoupeRenditionSizeData } ) or
		( type( defaultLoupeRenditionSizeData ) == "table" and
			defaultLoupeRenditionSizeData ) or
			error( "defaultLoupeRenditionSizeData has wrong type" )

	return WFRxRunUntilDone( self._schedulerWorker:request( 
								'updateForLoupeScrolling', 
									pixStreamId, assetIds,
									defaultLoupeRenditionSizeData,
									prevOnScreenRange, onScreenRange,
									prevBufferedRange, bufferedRange,
									scrollDirection, 
									prevFetchAheadRange, fetchAheadRange,
									prevFetchBehindRange, fetchBehindRange ) )

end

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

local _sharedStreams = {}

--------------------------------------------------------------------------------
-- This function calls 
-- 
-- 	PixelProviderScheduler:streamApiName( ... )
-- 	
-- on the pixel provider worker thread.
-- 
-- If it is called multiple times for ( ... ), the callers will share an
-- underlying worker thread request. This limits the amount of callbacks made
-- to this thread (i.e. the main daemon thread)
--------------------------------------------------------------------------------

local function makeSharedStream( self, streamApiName, ... )

	local streamKeyFmtStr = "%s" .. string.rep( "%s", select( '#', ... ) )

	local streamKey = string.format( streamKeyFmtStr, streamApiName, ... )

	local sharedStreamArgs = { ... }
	local sharedStreamArgsN = select( '#', ... )

	if not _sharedStreams[ streamKey ] then

		local nonRefCountedStream = 
				Rx.Observable.create(

					function( observer )

						local subscription = 
							self._schedulerWorker
								:request( streamApiName, 
											unpack( sharedStreamArgs, 1,
													sharedStreamArgsN ) )
								:subscribe( observer )


						return Rx.Disposable.create(
							function()
						
								_sharedStreams[ streamKey ] = nil

								return subscription:dispose()

							end )

					end )

		local sharedStream =
				nonRefCountedStream
					:publish()
					:refCount()

		_sharedStreams[ streamKey ] = sharedStream

	end

	return _sharedStreams[ streamKey ]

end

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

function PixelProviderProcessor:pixStreamIdPixelPathStream( pixStreamId )

	WFArguments_check( 'PixelProviderProcessor', 'string' )

	return makeSharedStream( self, 'pixStreamIdPixelPathStream', pixStreamId )

end

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

function PixelProviderProcessor:pixStreamIdsPixelPathStream( pixStreamIds )

	WFArguments_check( 'PixelProviderProcessor', 'table' )

	return makeSharedStream( self, 'pixStreamIdsPixelPathStream', pixStreamIds )

end

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

function PixelProviderProcessor:containsPixStreamIdSubstrPixelPathStream( pixStreamIdSubStr )

	WFArguments_check( 'PixelProviderProcessor', 'string' )

	return makeSharedStream( self, 'containsPixStreamIdSubstrPixelPathStream', 
								pixStreamIdSubStr )

end

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

function PixelProviderProcessor:bufferedPixelPathStream( fps )

	return makeSharedStream( self, 'bufferedPixelPathStream', fps )

end

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

function PixelProviderProcessor:_DEBUG_requestStatsStream()

	return makeSharedStream( self, '_DEBUG_requestStatsStream' )

end

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

function PixelProviderProcessor:_DEBUG_localAssetIdStatesStream( localAssetId )

	return makeSharedStream( self, '_DEBUG_localAssetIdStatesStream',
								localAssetId )

end

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

if DISABLE_PIXEL_PROVIDER then

	local function isOwnedFunction( methName )
		return type( methName ) ~= "number"
	end

    local function noOp() end

    local function returnEmpty() return Rx.Observable.empty() end

    for methName, maybeMethod in pairs( PixelProviderProcessor ) do

		-- only override the named methods we explicitly set in this file.
        if type( maybeMethod ) == "function" and isOwnedFunction( methName ) then

            PixelProviderProcessor[ methName ] = 
                ( methName == "__init" or methName == "close" ) and
                    noOp or
                    returnEmpty

        end

    end

end

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

return Rx.Class( 'PixelProviderProcessor', PixelProviderProcessor )
