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

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

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

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable
local ReplaySubject = Rx.ReplaySubject

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

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

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

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

local check = assert( WFArguments.check )
KSData.registerArgumentChecker()

local logger = WFLogger 'WFRxContinuousTask'
--:setLevel( 'debug' )

local kTaskStopped = 'WFRxContinuousTask.taskStoppedError'


local kBackoffStrategy_Exponential = 'exponential'
local kBackoffStrategy_Linear = 'linear'

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

--[[!
	\class WFRxContinuousTask
	
	\brief This class provides basic functionality for continuous tasks.
]]--

local WFRxContinuousTask = {

	_taskName = false,
		-- (string) A name for the task. Used for debugging.

	_taskFunc = false,
		-- (function) A user-specified function used to perform the actual task.

	_stopContinuousTask = false,
		-- (ReplaySubject< boolean >) Indicates if we should stop scheduling the task.

	_taskDisposable = false,
		-- (Disposable) An Rx disposable object of the task.

	_taskRunning = false,
		-- (boolean) Indicates if the task is currently running.

	_taskDoingWork = false,
		-- (Rx ReplaySubject) A Rx.Subject used to indicate whether the task is performing work.

	_progressiveBackoff = false,
		-- (boolean) Indicates if we should use progressive backoff for failed tasks.
		
	_backoffStrategy = false,
		-- (string) Indicates the strategy used for progressive backoff. 
		-- kBackoffStrategy_Exponential or kBackoffStrategy_Linear
		
	_errorDelay = false,
		-- (number) Seconds to delay before restarting task in the event of an error.
		
	_noWorkDelay = false,
		-- (number) Seconds to delay before restarting task when we don't have work.
		
	_maxErrorDelay = false,
		-- (number) Maximum amount we can delay a task that is having errors.
		
	_maxNoWorkDelay = false,
		-- (number) Maximum amount we can delay a task when no work is available.
		
	_noWorkRetryCount = 0,
		-- (number) Number of times this task has been retried with success when no work occurs.
		
	_errorRetryCount = 0,
		-- (number) Number of times this task has been retried when errors are occuring.
		
	_hasClosed = false,
		-- (boolean) Indicates if this object has closed or not.

	_retryTimer = false,
 		-- (Subscription) An Rx subscription to a timer that will retry the task.
		
	_debug_isTaskRunningStream = false,
		-- (Rx.ReplaySubject) Used by unit tests.
		
	class = {
	
		BackoffStrategy_Linear = kBackoffStrategy_Linear,
		BackoffStrategy_Exponential = kBackoffStrategy_Exponential
		
	},
		
	WFCloseSupport.addCloseSupportMethods,

}

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

--[[!

	\brief Creates a new continuous task.

	\param taskName (WFCatalog) A valid name for the task.
	\param taskFunc (function) The task to perform.
	\param taskInfo (table) A table containing values used for delaying the task.
	
		{
			progressiveBackoff = <bool>,
			noWorkDelay = <number>,
			errorDelay = <number>,
			maxDelay = <number>,
			maxNoWorkDelay = <number>,
		}
	
--]]

function WFRxContinuousTask:__init( taskName, taskFunc, taskInfo )

	check( 'WFRxContinuousTask', 'string', 'function', '?table' )

	logger:infof( "Constructing WFRxContinuousTask: %s", taskName )

	taskInfo = taskInfo or {}

	self._taskName = taskName

	self._taskFunc = taskFunc

	self._stopContinuousTask = ReplaySubject( 1 )
	self._stopContinuousTask:onNext( false )
	
	self._progressiveBackoff = taskInfo.progressiveBackoff ~= false

	self._backoffStrategy = taskInfo.backoffStrategy or kBackoffStrategy_Linear

	self._noWorkDelay = taskInfo.noWorkDelay or 3
	
	self._errorDelay = taskInfo.errorDelay or 5

	self._maxErrorDelay = taskInfo.maxDelay or 60

	self._maxNoWorkDelay = taskInfo.maxNoWorkDelay or self._noWorkDelay

	self._taskDoingWork = ReplaySubject( 1 )
	self._taskDoingWork:onNext( false )

	self._debug_isTaskRunningStream = ReplaySubject( 1 )
	self._debug_isTaskRunningStream:onNext( false )

	WFDiagnosticLog.addLogProvider( self )

end

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

--[[!

	\brief Close and stop the task.

	\return (void)
	
--]]

function WFRxContinuousTask:close()

	check( 'WFRxContinuousTask' )

	self._hasClosed = true

	local stopContinuousTask = self._stopContinuousTask
	self._stopContinuousTask = nil
	
	if stopContinuousTask then
		stopContinuousTask:onCompleted()
		stopContinuousTask:dispose()
	end
	
	local taskDisposable = self._taskDisposable
	self._taskDisposable = nil
	
	if taskDisposable then
		taskDisposable:dispose()
	end

	local debug_isTaskRunningStream = self._debug_isTaskRunningStream
	self._debug_isTaskRunningStream = nil
	
	if debug_isTaskRunningStream then
		debug_isTaskRunningStream:dispose()
	end

end

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

local function doTaskContinuously( self )

	if self._retryTimer then
		self._retryTimer:dispose()
		self._retryTimer = nil
	end

	local function retryAfterDelay( delay, retryCount, maxDelay )

		if not self._taskRunning then
			return
		end

		if self._progressiveBackoff then
		
			maxDelay = maxDelay or delay
			
			if self._backoffStrategy == kBackoffStrategy_Exponential then
				
				delay = math.min( delay * ( 2 ^ retryCount ), maxDelay )
				
			else -- kBackoffStrategy_Linear / Default
			
				delay = math.min( delay * retryCount, maxDelay )
				
			end
			
		end
		
		logger:infof( "Restarting %s task after delay %.2f", self._taskName, delay )
		
		if delay > 0 then
		
			 self._retryTimer = Observable.timer( delay, WFRxScheduler.sharedScheduler() )

				:subscribe( function()
				
					self._retryTimer = nil
					self._taskDisposable = doTaskContinuously( self )
				
				end )
				
		else
		
			self._taskDisposable = doTaskContinuously( self )
			
		end

	end

	if self._taskDisposable then
		self._taskDisposable:dispose()
		self._taskDisposable = nil
	end

	if self._hasClosed or WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		return
	end

	return self._stopContinuousTask

		:take( 1 ) -- just in case

		:filter( function( shouldStop )
			return not shouldStop
		end )

		:flatMapLatest( function()

			-- Increment _noWorkRetryCount here so taskDidWork can
			-- reset it to 0 after we increment it and cause the next
			-- call to occur immediately instead after 1 * noWorkdelay.
			self._noWorkRetryCount = self._noWorkRetryCount + 1

			return self._taskFunc( self )

		end )
		
		:subscribe(
			nil,
			function( errorMessage )

				self._taskDoingWork:onNext( false )

				self._errorRetryCount = self._errorRetryCount + 1

				logger:warnf( "Continuous task %s sent onError: %s ", self._taskName, errorMessage )
				retryAfterDelay( self._errorDelay, self._errorRetryCount, self._maxErrorDelay )
				
			end,
			function()

				if self._noWorkRetryCount > 0 or not self._taskRunning then
					-- taskDidWork wasn't called.
					self._taskDoingWork:onNext( false )
				end

				self._errorRetryCount = 0

				logger:infof( "Continuous task %s sent onCompleted", self._taskName )
				retryAfterDelay( self._noWorkDelay, self._noWorkRetryCount, self._maxNoWorkDelay )
				
			end )

end

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

--[[!

	\brief Notify that the task did work.

	\param optDidWork (boolean) Optional parameter to identify whether the task
	did work or not on this invocation. Defaults to true. You will only need to
	call with false in the event that the observable returned by your taskFunc
	doesn't send an onCompleted.
	
	\return (void)
	
--]]

function WFRxContinuousTask:taskDidWork( optDidWork )

	check( 'WFRxContinuousTask', '?boolean' )

	local didWork = optDidWork == nil or optDidWork

	-- If the task did work, we reset the retry count which is used in the
	-- progressive backoff logic.

	logger:infof( "Continuous task %s did work? %s", self._taskName, didWork )

	if didWork then

		if self._taskRunning then
			self._noWorkRetryCount = 0
		end

	else

		self._noWorkRetryCount = self._noWorkRetryCount + 1

	end

	self._taskDoingWork:onNext( didWork )

end

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

--[[!

	\brief Reports whether the continuous task is actually doing any work.

	\return (boolean) true if doing work, false if last run of task did nothing.

--]]

function WFRxContinuousTask:isTaskDoingWork()

	return self._taskDoingWork
		:distinctUntilChanged()
		:tap( function( isTaskWorking )
			logger:infof( "Is %s working? %s", self._taskName, isTaskWorking )
		end )
		
end


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

--[[!

	\brief Stop the continuous task.

	\return (void)
	
--]]

function WFRxContinuousTask:stopContinuousTask()

	self._taskRunning = false
	
	if self._stopContinuousTask then
		self._stopContinuousTask:onNext( true )
	end

	if self._debug_isTaskRunningStream then
		self._debug_isTaskRunningStream:onNext( false )
			-- Used by unit tests.
	end
	
end

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

--[[!

	\brief Start the continuous task.

	\return (void)
	
--]]

function WFRxContinuousTask:startContinuousTask()

	if not self._taskRunning then
		self._taskRunning = true
		self._noWorkRetryCount = 0
		self._errorRetryCount = 0
		self._stopContinuousTask:onNext( false )
		self._taskDisposable = doTaskContinuously( self )
	end

	if self._debug_isTaskRunningStream then
		self._debug_isTaskRunningStream:onNext( true )
			-- Used by unit tests.
	end
	
end

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

-- SPAM [jandrew 2019-08-29]: This function is only used by unit tests.

function WFRxContinuousTask:_testOnly_getIsTaskRunningStream()

	return self._debug_isTaskRunningStream

end

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

--[[!

	\brief Triggers the continuous task and resets the progressive backoff.

	\return (void)
	
--]]

function WFRxContinuousTask:triggerTask()

	self._noWorkRetryCount = 0

	if self._taskRunning and self._retryTimer then
		
		-- If _retryTimer is set, then we know the task is currently waiting
		-- to run again. In this case, we can immeadiately call doTaskContinuously()
		-- to trigger the task. If _retryTimer is not set, then the task is already
		-- running and there's no need to trigger it.
		
		self._taskDisposable = doTaskContinuously( self )
		
	end

end

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

--[[!

	\brief Reports whether the continuous task is running or not.

	\return (boolean) true if running, false if stopped.

--]]

function WFRxContinuousTask:isRunning()

	return self._taskRunning

end

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

--[[!

	\brief Helper function to be used within a flatMapLatest operator by a
	taskFunc that doesn't typically complete.  This will throw a kTaskStopped
	error if the task has been aborted.  There is another helper function
	called throwIfNotStopped that you can place in a catch handler if you
	do not want kTaskStopped errors returned from your observable as errors.

	\return (Observable) observable throws with kTaskStopped as error value
	if this WFRxContinuousTask is not running, otherwise it relays the value.

--]]

function WFRxContinuousTask:throwIfNotRunning( ... )

	if not self._taskRunning then
		return Observable.throw( kTaskStopped )
	end

	return Observable.just( ... )

end

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

--[[!

	\brief Helper function to be used within a :catch operator that is used
	in conjunction with throwIfNotRunning.

	\return (Observable) observable completes if errorMessage was caused by
	calling WFRxContinuousTask:throwIfNotRunning.

--]]

function WFRxContinuousTask:throwIfNotStopped( errorMessage )

	self:taskDidWork( false )

	if errorMessage == kTaskStopped and not self._taskRunning then

		-- Complete the observable without error
		return Observable.empty()

	end

	return Observable.throw( errorMessage )

end

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

function WFRxContinuousTask:sendDiagnosticLogData()

	local body = {}
	for k, v in pairs( self ) do
		if WFDiagnosticLog.isSimpleValue( v ) then
			body[ k ] = v
		end
	end

	local function addCurrentRetryDelay( delayName, delay, retryCount, maxDelay )

		if not self._taskRunning then
			return
		end

		if self._progressiveBackoff then
			maxDelay = maxDelay or delay
			delay = math.min( delay * retryCount, maxDelay )
		end

		body[ delayName ] = delay

	end

	addCurrentRetryDelay( "currentNoWorkDelay", self._noWorkDelay, self._noWorkRetryCount, self._maxNoWorkDelay )
	addCurrentRetryDelay( "currentErrorDelay", self._errorDelay, self._errorRetryCount, self._maxErrorDelay )

	body.taskDoingWork = self._taskDoingWork

	return {

		name = "WFRxContinuousTask: " .. self._taskName,
		id = 'WFRxContinuousTask-' .. self._taskName,
		weight = 530,

		body = body,

	}

end

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

return Rx.Class 'WFRxContinuousTask'( WFRxContinuousTask )
