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

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

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

WARNING: Rx documentation typically speaks of time in milliseconds, but there
is nowhere in the framework where actual milliseconds are coded in as a constant.

WF has a long history of working in seconds instead of milliseconds. For that
reason, it was deemed a lesser-of-evils to ignore Rx's definition of time and
implement the WF-Rx bridge using seconds throughout. (In other words, we're not
doing any seconds-to-milliseconds conversion.)

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

	-- RxLua
local Rx = require 'Rx'
local PriorityQueue = Rx.internals.PriorityQueue
local ScheduledItem = Rx.internals.ScheduledItem
local SchedulePeriodicRecursive = Rx.internals.SchedulePeriodicRecursive

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

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

	-- WF.core.os
local KSDateImp = require 'KSDateImp'

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


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

local check = assert( WFArguments.check )

local math_max = assert( math.max )
local math_ceil = assert( math.ceil )

local timeNowForAnimation = assert( KSDateImp.timeNowForAnimation )

local requestPulseAfterDelay = assert( WFCurrentThread.requestPulseAfterDelay )

local schedulerLog = WFLogger 'WFRxScheduler'

local kTimerTickPrecision = 16
	-- We are precise to a 16th of a second

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

-- @mark WFRxSchedulerImp class

local WFRxSchedulerImp = {

	Rx.Scheduler,

	_queue = false,
		-- (Rx.PriorityQueue)

	_upcomingTicks = false,
		-- (table) set of upcoming tick times

}

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

local function nextItemInQueue( self )

	local queue = self._queue

	while queue:length() > 0 do

		local nextItem = queue:peek()

		if not nextItem:isCancelled() then
			return nextItem
		end

		queue:dequeue()

	end

	return nil

end

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

local function scheduleNewPulseIfNeeded( self )

	local queue = self._queue

	local nextItem = queue:length() > 0 and queue:peek()

	if not nextItem then
		return
	end

	local currentTime = timeNowForAnimation()

	local targetTime = math_max( nextItem._dueTime, currentTime + .02 )
		-- Can not request a zero delay via requestPulseAfterDelay, so ensure
		-- that we're asking for a pulse at least slightly later than now.

	local targetTimerTicks = math_ceil( targetTime * kTimerTickPrecision )
								/ kTimerTickPrecision

	if self._upcomingTicks[ targetTimerTicks ] then
		return
	end

	self._upcomingTicks[ targetTimerTicks ] = true

	self:requestPulseAfterDelay( targetTimerTicks - currentTime )

end

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

local function runScheduledItems( self, shuttingDown )

	local currentTime = timeNowForAnimation()

	-- Remove upcomingTicks entries for any clock values
	-- that have now passed.

	if not shuttingDown then

		local currentTick = math_ceil( currentTime * kTimerTickPrecision ) / kTimerTickPrecision
		for tick in pairs( self._upcomingTicks ) do
			if tick <= currentTick then
				self._upcomingTicks[ tick ] = nil
			end
		end

	end

	-- Run queue of scheduled items.

	-- FIXME [scouten 2015-01-01]: Rework with pcall loop?

	local hasItemToProcess = true

	while hasItemToProcess do

		local nextItem = nextItemInQueue( self )

		if not nextItem then
			hasItemToProcess = false
			break
		end

		local success, message = pcall( function()

			if shuttingDown then

				-- We're shutting down. Close this item out of queue.

				nextItem._disposable:dispose()
				return true

			elseif nextItem._dueTime <= currentTime then

				-- Next item in queue is due now. Run it.

				nextItem:invoke()
				return true

			end

		end )

		if not success then
			schedulerLog:warnf( "scheduled item threw an error: %s", message )
		elseif not message then
			hasItemToProcess = false
			break
		end

	end

	if not shuttingDown then
		scheduleNewPulseIfNeeded( self )
	end

end

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

function WFRxSchedulerImp:__init()

	self._queue = PriorityQueue()

	self._upcomingTicks = {}
	self:registerPulseHandler( function( shuttingDown )
		runScheduledItems( self, shuttingDown )
	end )

end

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

function WFRxSchedulerImp:now()

	return timeNowForAnimation()

end

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

function WFRxSchedulerImp:schedule( taskRef, action )

	if action == nil then
		check( '_', 'function' )
	else
		check( '_', '_', 'function' )
	end

	if action == nil then
		action = taskRef
		taskRef = nil
	end

	return self:_scheduleFuture( taskRef, 0, action )

end

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

function WFRxSchedulerImp:_scheduleFuture( taskRef, dueTime, action )

	check( '_', '_', 'number', 'function' )

	local si

	local function run( scheduler, taskReference )
		self._queue:remove( si )
		return action( scheduler, taskReference )
	end

	si = ScheduledItem( self, taskRef, run, timeNowForAnimation() + dueTime )
	self._queue:enqueue( si )

	scheduleNewPulseIfNeeded( self )

	return si._disposable

end

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

function WFRxSchedulerImp:schedulePeriodic( taskRef, period, action )

	if action == nil then
		check( '_', 'number', 'function' )
	else
		check( '_', '_', 'number', 'function' )
	end

	if action == nil then
		action = period
		period = taskRef
		taskRef = nil
	end

	local s = SchedulePeriodicRecursive( self, taskRef, period, action )
	return s:start()

end

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

function WFRxSchedulerImp:requestPulseAfterDelay( delay )

	return requestPulseAfterDelay( delay )

end

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

function WFRxSchedulerImp:registerPulseHandler( handler )

	return WFCurrentThread.registerPulseHandler( handler )

end

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

WFRxSchedulerImp = Rx.Class( 'WFRxSchedulerImp', WFRxSchedulerImp )

local instance
	-- Lazy instantiated singleton instance of WFRxSchedulerImp for this thread

rawset( WFRxSchedulerImp, 'sharedScheduler', function()
	instance = instance or WFRxSchedulerImp()
	return instance
end )

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

return WFRxSchedulerImp
