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

WFCatPixelFileCacheSaveImp.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.

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

	-- Rx
local Rx = require 'Rx'

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

	-- WF.core.lua
local KSData = require 'KSData'

	-- WF.core.rxlua
local WFRxScheduler = require 'WFRxScheduler'

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

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


local check = assert( WFArguments.check, "WFArguments.check not defined" )
local KSData_length = KSData.length
local Observable = Rx.Observable
local Observable_throw = Observable.throw

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

local WFCatPixelFileCacheSaveImp = {

	class = {},

	_name = false,
		-- (string) The name of this pixel file cache. If the name is "test"
		-- then file operations are mocked out.

	_test_fileOps = false,
		-- (Subject or false) A stream of file operations performed by the file
		-- cache when in "test" mode.

	_test_backlogTimer = false,
		-- (Subject or false) A stream that fires after a simulated backlog.
		-- used to allow "too many" requests to queue up for testing errors that
		-- are expected in that situation.

	_maxConcurrentCalls = false,
		-- (positive, non-zero integer) How many calls are permitted to be
		-- running at the same time.

	_maxBytesInFlight = false,
		-- (positive, non-zero integer) how many bytes of jpeg data are allowed
		-- to be in the act of saving at once.

	_numConcurrentCalls = false,
		-- (positive, non-zero integer) How many calls are currently running.

	_numBytesInFlight = false,
		-- (positive, non-zero integer) How many bytes of jpeg data are
		-- currently being saved to disk.

	WFCloseSupport.addCloseSupportMethods,

}

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

--[[!
	Create a new file cache. You may create as many of these objects as you
	want, and they will all use the same worker thread. However, some features,
	such as client-side rate-limiting, only work on a per-instance basis. (The
	limits are still applied on the worker thread, but there is a slight delay
	in doing so, and a system under heavy load might accumulate more memory
	especially if the worker thread is I/O bound and blocked on a write.)

	\param name (string) The name for this cache. Each unique name uses
		a different background thread for persistence. This might be useful
		for unit testing, or for dedicating a separate thread to caches
		located on different physical storage volumes.
	\param maxConcurrentCalls (non-zero positive integer) how many ongoing,
		simultaneous calls to :savePixels are allowed at once.
	\param maxBytesInFlight (non-zero positive integer) how many bytes of jpeg
		data are allowed to be in the act of saving at once.
--]]

function WFCatPixelFileCacheSaveImp:__init( name, maxConcurrentCalls, maxBytesInFlight )

	check( 'WFCatPixelFileCacheSaveImp', '+string', 'number', '?number' )

	assert( maxConcurrentCalls > 0, 'maxConcurrentCalls must be > 0' )
	assert( maxConcurrentCalls == math.floor( maxConcurrentCalls ),
		'maxConcurrentCalls must be an integer' )
	assert( maxBytesInFlight >= 0, 'maxBytesInFlight must be >= 0' )
	assert( maxBytesInFlight == math.floor( maxBytesInFlight ),
		'maxBytesInFlight must be an integer' )

	self._name = name

	if string.find( name, '^mockFileOps' ) then
		self._test_fileOps = Rx.Subject()
		self:addCloseAction( function()
			self._test_fileOps:onCompleted()
			self._test_fileOps:dispose()
		end )
	end

	self._maxConcurrentCalls = maxConcurrentCalls
	self._maxBytesInFlight = maxBytesInFlight
	self._numConcurrentCalls = 0
	self._numBytesInFlight = 0

	KSData.registerArgumentChecker()
		-- Ensure 'KSData' checks in our methods will succeed

	WFCloseSupport.mustCloseAtThreadTermination( self )

end

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

--[[!
	\brief Save the blob to disk. This is done on a dedicated WFWorkerThread.
	Should more calls to savePixels accumulate than maxConcurrentCalls (10) or
	maxBytesInFlight (128 MB) the call will result in a "too many requests"
	error.

	If the parent directory does not exist this call will resolve to an error
	"no such directory". If the file already exists at that location, that file
	will be overwritten if possible, otherwise will result in an "access denied"
	error. An "access denied" error may also occur if there is insufficient
	permissions to write to the directory, or if any of the directories or
	parent directories are not directories but are, for example, files instead.

	The jpegBlob KSData will be closed when this operation completes (or fails).
	If the caller wishes keep a copy of the data for other operations, they must
	construct a copy by calling `KSData.new( jpegBlob )`

	Note that limit testing is done in two passes: Each client thread that saves
	will enforce the limit for that calling thread. Additionally, the worker
	thread will also keep track of the number of pending incoming requests and
	will short-circuit (and fail) if the input queue grows beyond the limits. In
	either case, the error is signaled as an observable error: This function
	will never throw a lua error directly.

	\param filePath (string, file path) must be a path to the location to save
		the bytes.
	\param jpegBlob (KSData) the data to write to the file.

	\return (Observable) Completes when operation finishes successfully, or
		results in onError with the error message.
--]]

function WFCatPixelFileCacheSaveImp:savePixels( filePath, jpegBlob )

	check( 'WFCatPixelFileCacheSaveImp', '+string', 'KSData' )

	local numBytes = KSData_length( jpegBlob )
	local totalBytes = self._numBytesInFlight + numBytes
	local numCalls = self._numConcurrentCalls + 1

	if self._maxBytesInFlight > 0 and 
			totalBytes > self._maxBytesInFlight then
		jpegBlob:close()
		return Observable_throw( "too much memory used by unsaved blobs" )
	end

	if  numCalls > self._maxConcurrentCalls then
		jpegBlob:close()
		return Observable_throw( "too many requests" )
	end

	self._numBytesInFlight, self._numConcurrentCalls = totalBytes, numCalls

	local function decrementAndRelease()
		self._numBytesInFlight = self._numBytesInFlight - numBytes
		self._numConcurrentCalls = self._numConcurrentCalls - 1
		jpegBlob:close()
	end

	return self:fireAsap()
		:map( function()
			self:doSave( filePath, jpegBlob )
			return true
		end )
		:takeLast( 0 )
		:finally( decrementAndRelease )

end

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

function WFCatPixelFileCacheSaveImp:doSave( filePath, jpegBlob )

	check( 'WFCatPixelFileCacheSaveImp', '+string', 'KSData' )

	if self._test_fileOps then
			-- self._test_fileOps is only defined if we are the special 'test' instance
			-- that never actually writes to the file system.
		local md5 = jpegBlob:digest( 'MD5' )
		self._test_fileOps:onNext( 'write', filePath, KSData.length( jpegBlob ), md5 )
	else
		KSData.saveToFile( jpegBlob, filePath )
	end
end

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

function WFCatPixelFileCacheSaveImp:fireAsap()
	if self._test_backlogTimer then
		return self._test_backlogTimer
	end
	return Observable.timer( 0, WFRxScheduler.sharedScheduler() )
end

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

function WFCatPixelFileCacheSaveImp:test_getFileOps( )
	return self._test_fileOps
end

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

function WFCatPixelFileCacheSaveImp:test_startBacklog( )
	self._test_backlogTimer = Rx.Subject()
	return Observable.empty()
end

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

function WFCatPixelFileCacheSaveImp:test_endBacklog( )
	local oldTimer = self._test_backlogTimer
	self._test_backlogTimer = false
	oldTimer:onNext( 1 )
	oldTimer:onCompleted()
	oldTimer:dispose()
	return Observable.empty()
end

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

function WFCatPixelFileCacheSaveImp.class.initMessageProcessor( Class, id )

	local args = KSJsonDecoder.decodeWebSafeJson( id )
	local imp = Class( unpack( args ) )

	WFRequestDispatcher.onMessageDo {
		message = 'savePixels',
		target = imp,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'test_getFileOps',
		target = imp,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'test_startBacklog',
		target = imp,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'test_endBacklog',
		target = imp,
	}

end

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

return Rx.Class( 'WFCatPixelFileCacheSaveImp', WFCatPixelFileCacheSaveImp )
