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

WFRequestDispatcher.lua

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

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

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

-- NOTE: This namespace is fundamental to the framework operation. Please avoid
-- introducing any other dependencies.

local assert = assert
local require = require
local type = type
local pairs = pairs

	-- RxLua (lazy load)
local RxObservable_lazy

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

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

	-- WF.core.reactive (lazy load)
local KSPromises_lazy
local KSReactive_lazy

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSPCallLoops = require 'KSPCallLoops'
local WFArguments = require 'WFArguments'
local WFCapabilities = require 'WFCapabilities'


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

local requestDispatcherLog = WFLogger( 'WFRequestDispatcher' )
--	:setLevel( 'DEBUG' )


--============================================================================--
--[[!
	\class WFRequestDispatcher

	\brief This namespace provides the mechanism by which worker threads can
		respond to messages from other threads.
--]]

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

local function RxObservable_isObservable_lazy()
	return false
end

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

local function KSPromises_isPromise_lazy()
	return false
end

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

local function KSReactive_isEventNode_lazy()
	return false
end

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

local KSReactive_isBehavior_lazy
	-- Can't be reached unless reactive system isn't available because
	-- KSReactive_isEventNode_lazy will always return false.

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

local KSReactive_lazy_SAME = 'same'

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

local function loadReactiveSystem()

	if not RxObservable_lazy then
		local Rx = require 'Rx'
		RxObservable_lazy = assert( Rx.Observable )
		RxObservable_isObservable_lazy = function( obj )
			return RxObservable_lazy:isClassOf( obj )
		end
	end

	if not KSPromises_lazy then
		KSPromises_lazy = require 'KSPromises'
		KSPromises_isPromise_lazy = assert( KSPromises_lazy.isPromise )
	end

	if not KSReactive_lazy then
		KSReactive_lazy = require 'KSReactive'
		KSReactive_isEventNode_lazy = assert( KSReactive_lazy.isEventNode )
		KSReactive_isBehavior_lazy = assert( KSReactive_lazy.isBehavior )
		KSReactive_lazy_SAME = assert( KSReactive_lazy.SAME )
	end

end


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

-- @mark WFRequestDispatcher namespace

local WFRequestDispatcher = {}

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

-- Data stored locally for requests. Indexed by raw request userdata.

local disposablesForRawRequests = {}
local eventNodesForRawRequests = {}
local promiseActionsForRawRequests = {}

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

local _requestStats -- (table) request stats, used for diagnostic log if requested
	-- key: (string) request name
	-- value: (table) info on request (count, min, max, total)

local _shouldBenchmark =
		WFCapabilities.getCapability( "WFWorkerThread_shouldBenchmark" )

local check = assert( WFArguments.check )

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

local stuffToClose

--------------------------------------------------------------------------------
--[[!
	\brief Closes a specific object after processing current request.

	The object is closed when the current request processing has finished.
	Typically, this is just after the request's data has been placed in
	transit. If the request has already been closed, it occurs when we
	detect that closure.

	\param object The object to close. Must be a \c WFCloseTarget or a native object
		or nil.

	\return (object) The object that was closed.
--]]

function WFRequestDispatcher.deferredClose( --[[WFCloseTarget]] object )

	check( '? userdata | table' )

	if object == nil then
		return
	end

	if not object.close then
		error( string.format( "WFRequestDispatcher.deferredClose: object %s (no close method)",
							   tostring( object ) ), 2 )
	end

	if not stuffToClose then
		stuffToClose = {}
	end

	stuffToClose[ object ] = true

	return object

end

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

local function sendClose( obj )

	obj:close()

end

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

local function processDeferredClosures()

	while stuffToClose do

		local currentCycle = stuffToClose

		stuffToClose = nil

		KSPCallLoops.pdoWithIterator( sendClose, nil, pairs( currentCycle ) )

	end

end

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

local function closeRawRequest( rawRequest )

	-- Unsubscribe from any associated Observable.

	local disposable = disposablesForRawRequests[ rawRequest ]

	if disposable then

		local messageId = disposable._messageId

		if messageId and _requestStats then

			local requestStatsForMessage = _requestStats[ messageId ]

			if requestStatsForMessage then

				requestStatsForMessage.earlyCloseCount = 
					( requestStatsForMessage.earlyCloseCount or 0 ) + 1

			end

		end

		requestDispatcherLog:debugf( "request closed before disposed. disposing now\n  "
									  .. "messageId = %s  requestId = %d",
									  messageId,
									  disposable._requestId )
		
		disposablesForRawRequests[ rawRequest ] = nil
		disposable:dispose()

		return

	end

	-- Close any associated event node.

	local eventNode = eventNodesForRawRequests[ rawRequest ]

	if eventNode then

		eventNodesForRawRequests[ rawRequest ] = nil
		eventNode:close()

		return

	end

	-- Close any associated promise action.

	local promiseAction = promiseActionsForRawRequests[ rawRequest ]

	if promiseAction then

		promiseActionsForRawRequests[ rawRequest ] = nil
		promiseAction:close()

		return

	end

end

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

WFCurrentThread.registerPriorityChangeHandler( function( rawRequest, priority )

	-- We don't really support the notion of prioritization. (Never have.)
	-- We can, however, cancel requests.

	if priority <= -10000 then
		closeRawRequest( rawRequest )
	end

end )

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

local requestDispatcherLog_will_trace = requestDispatcherLog.will_trace

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

local requestDispatcherLog_will_warn = requestDispatcherLog.will_warn

--------------------------------------------------------------------------------
-- to make it easier to hook in benchmarking, intentionally implement
-- sendResponseWithData to have all capturable benchmarking data in first n
-- positions, followed by request, then the rest of the response data
--
-- note1: even though message appears to be used only for logging, benchmarking
-- hooks use it to map responses back to specific worker thread functions, so
-- it should not be removed from this function.
--

local function sendResponseWithData( message, maybeDoneStatus, dataType, 
										request, ... )
	
	if requestDispatcherLog_will_trace then

		requestDispatcherLog:tracef( "%s is sending from %s, %s, ...", 
										message, maybeDoneStatus, dataType )

	elseif requestDispatcherLog_will_warn and not dataType then

		requestDispatcherLog:warnf( "%s is using sendResponseWithData with done state %s, but no dataType!", 
										message, maybeDoneStatus, dataType )

	end
	
	return request:send( maybeDoneStatus, dataType, ... )

end

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

local function bindRequestToObservable( request, messageId, observable )

	local observableRequestId = KSIdGenerator.localCounter()

	requestDispatcherLog:debugf( "bind request to Observable\n  messageId = %s  requestId = %s",
								  messageId, observableRequestId )

	local rawRequest = request:getRawRequestRef()

	local disposable

	local didComplete = false

	local function unsubscribe()

		requestDispatcherLog:debugf( "request unsubscribing from Observable\n  "
									  .. "messageId = %s  requestId = %d",
									  messageId, observableRequestId )

		if disposable then
			disposable:dispose()
			disposable = nil
		end

	end

	local function sendData( ... )

		requestDispatcherLog:debugf( "sending Observable update\n  messageId = %s\n  "
									  .. "requestId = %d  values = %s, %s, %s",
									  messageId, observableRequestId, ... )

		if not request then

			unsubscribe()

		elseif not sendResponseWithData( messageId, nil, 'data', request, ... ) then

			request:close()
			request = nil

			closeRawRequest( rawRequest )
			unsubscribe()

		end

		processDeferredClosures()

	end

	local function sendCompleted()

		requestDispatcherLog:debugf( "sending Observable DONE message\n  "
									  .. "messageId = %s  requestId = %d",
									  messageId, observableRequestId )

		didComplete = true

		if request then
			request:send( 'done' )
		end

		closeRawRequest( rawRequest )

		unsubscribe()

		processDeferredClosures()
	end

	local function sendError( ... )

		requestDispatcherLog:debugf( "sending Observable ERROR message %s\n  "
									  .. "requestId = %d",
									  messageId, observableRequestId )

		if request then

			-- We can ignore status from :send call because the
			-- Observable is now in terminal state.

			sendResponseWithData( messageId, nil, 'error', request, ... )
			sendCompleted()

		end

	end

	disposable = observable:subscribe( sendData, sendError, sendCompleted )

	-- Some Observables, such as Observable.just(), complete on subscription.

	-- if a sendData occurs during subscribe and request:send fails the disposable
	-- will be disposed of so no need to save it as the array entry will never get cleared.

	if not didComplete and not disposable:isDisposed() then
		rawset( disposable, '_requestId', observableRequestId )
		rawset( disposable, '_messageId', messageId )
		disposablesForRawRequests[ rawRequest ] = disposable
	else
		disposable:dispose()
		disposable = nil
	end
	
end

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

local function bindRequestToEventNode( request, messageId, eventNode )

	local rawRequest = request:getRawRequestRef()

	eventNodesForRawRequests[ rawRequest ] = eventNode

	local function closeRequest()
		if request then
			request:close()
			request = nil
			closeRawRequest( rawRequest )
		end
	end

	local function closeEventNode()
		eventNode:close()
		eventNodesForRawRequests[ rawRequest ] = nil
	end

	local function sendNonDone( nonDoneStatus, ... )

		if not request then

			closeEventNode()

		elseif not sendResponseWithData( messageId, nil, nonDoneStatus, request, ... ) then

			closeRequest( true )

			closeEventNode()

		end

	end

	local function sendData( ... )

		requestDispatcherLog:debugf( "sending event node update\n  messageId = %s\n  values = %s, %s, %s",
									  messageId, ... )

		sendNonDone( 'data', ... )

		processDeferredClosures()

	end

	local function sendError( ... )

		requestDispatcherLog:debugf( "sending event node ERROR\n  messageId = %s\n  values = %s, %s, %s",
									  messageId, ... )

		sendNonDone( 'error', ... )

		processDeferredClosures()

	end

	local function sendDone()

		requestDispatcherLog:debugf( "sending event node CLOSED\n  messageId = %s", messageId )

		if request then
			request:send( 'done' )
		end

		closeEventNode()

		closeRequest( false )

		processDeferredClosures()

	end

	if KSReactive_isBehavior_lazy( eventNode ) then
		local isError, message = eventNode:isErrorNow()
		if isError then
			sendError( message )
		else
			sendData( eventNode:valueNow() )
		end
	end

	if eventNode:isDone() then
		sendDone()
		return
	end

	-- Use closingRawMapE to avoid using onDataE and onErrorE. This changes which
	-- event node we're actually bound to.

	eventNode = eventNode:closingRawMapE( function( status, ... )

		if status == 'data' then
			sendData( ... )
		elseif status == 'error' then
			sendError( ... )
		end

		return KSReactive_lazy_SAME

	end )

	-- Then attach a detector for the done state.

	eventNode = eventNode:closingWeakOnDoneE( sendDone ):protectFromAutoCloseE()

	eventNodesForRawRequests[ rawRequest ] = eventNode

end

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

local function bindRequestToPromise( request, messageId, promiseP )

	local rawRequest = request:getRawRequestRef()

	local function handlePromiseResult( status, ... )

		promiseActionsForRawRequests[ rawRequest ] = nil

		requestDispatcherLog:debugf( "sending promise result\n  messageId = %s\n  status = %s\n  values = %s, %s, %s",
									  messageId, status, ... )

		if status == 'data' then
			sendResponseWithData( messageId, 'done', 'data', request, ... )
		else
			sendResponseWithData( messageId, 'done', 'error', request, ... )
		end

		request:close()

		processDeferredClosures()

	end

	local promiseAction = promiseP:whenResolvedDo( handlePromiseResult )

	promiseActionsForRawRequests[ rawRequest ] = promiseAction

	promiseAction:runUntilDone()
		-- Don't start it until we've put it in the pool otherwise it
		-- might clear the entry before we put it in.

end

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

local function finish_sendResponse( messageId, startTime, request, success, ... )

	requestDispatcherLog:debugf( "finish_sendResponse\n  messageId = %s\n  success = %s",
								 messageId, success )

	if success then

		-- If first return value is an event node, then we bind the request
		-- to the node and keep it running.

		if RxObservable_isObservable_lazy( ... ) then

			bindRequestToObservable( request, messageId, ( ... ) )

		elseif KSReactive_isEventNode_lazy( ... ) then

			bindRequestToEventNode( request, messageId, ( ... ) )

		elseif KSPromises_isPromise_lazy( ... ) then

			bindRequestToPromise( request, messageId, ( ... ) )

		else

			-- If not an Observable, reactive node or promise, then treat
			-- this as a one-off request: Send back whatever values were returned
			-- to us and close the callback.

			requestDispatcherLog:debugf( "sending completed request results\n  messageId = %s\n  values = %s, %s, %s",
										 messageId, ... )

			sendResponseWithData( messageId, 'done', 'data', request, ... )
			request:close()

		end

	else

		requestDispatcherLog:debugf( "sending completed request results WITH ERROR\n  messageId = %s\n  values = %s, %s, %s",
									 messageId, ... )

		sendResponseWithData( messageId, 'done', 'error', request, ... )
		request:close()

	end

	if _requestStats then

		local elapsedTime = debug.timeNow() - startTime

		local stat = _requestStats[ messageId ]

		if stat then
			stat.count = stat.count + 1
			stat.min = math.min( stat.min, elapsedTime )
			stat.max = math.max( stat.max, elapsedTime )
			stat.total = stat.total + elapsedTime
			stat.errorCount = stat.errorCount + ( success and 0 or 1 )
		else
			_requestStats[ messageId ] = {
				messageId = messageId,
				count = 1,
				errorCount = success and 0 or 1,
				earlyCloseCount = 0,
				min = elapsedTime,
				max = elapsedTime,
				total = elapsedTime,
			}
		end

	end

	processDeferredClosures()

end

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

local function sendResponseOrErrorToCallback( messageId, request, func, ... )

	local startTime = _requestStats and debug.timeNow()

	finish_sendResponse( messageId, startTime, request, pcall( func, ... ) )

end

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

local _1MbAsK = 1024

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

local _1GbAsK = 1024 * _1MbAsK

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

local function formatKilobytesAsReadableUnit( kb )

	local unit = kb < _1MbAsK and 1 or ( kb < _1GbAsK and _1MbAsK or _1GbAsK )

	local unitStr = kb < _1MbAsK and "kb" or ( kb < _1GbAsK and "Mb" or "Gb" )

	return string.format( "%.2f %s", kb / unit, unitStr )

end

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

local function safeToNumber( n )

	return tonumber( n ) or 0

end

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

local function makeHTMLTableRow( cellDataArray, nCells, cellType )

	local cells = {}

	for i = 1, nCells do

		cells[ i ] = string.format( "<%s>%s</%s>", 
										cellType, cellDataArray[ i], cellType )

	end

	return string.format( "<tr>%s</tr>", table.concat( cells, "" ) )

end

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

local function mapA( f, t, ... )

	local ret = {}

	for i = 1, #t do

		ret[ i ] = f( t[ i ], ... )

	end

	return ret

end

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

local function makeHTMLTableRows( rowDataArray, nCells, cellType )

	-- if column nCells + 1 is non nil, it is a non displayed sort criteria

	if #rowDataArray > 0 and rowDataArray[ 1 ][ nCells + 1 ] then

		table.sort(
			rowDataArray,
			function( rowDataA, rowDataB )

				local aVal = rowDataA[ nCells + 1 ]

				local bVal = rowDataB[ nCells + 1 ]

				if type( aVal ) == type( bVal ) then

					return aVal > bVal

				end

				return tostring( aVal ) > tostring( bVal )

			end )

	end

	return table.concat( 
			mapA( makeHTMLTableRow, rowDataArray, nCells, cellType ), 
			"" )

end

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

local function makeHTMLTable( headings, rowDataArray )

	return string.format( "<table>%s%s</table>", 
							makeHTMLTableRow( headings, #headings, "th" ),
							makeHTMLTableRows( rowDataArray, #headings, "td" ) )

end

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

local function wrapInTag( tag, str )

	return string.format( "<%s>%s</%s>", tag, str, tag )

end

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

local function makeClientIdMessageIdHTML( 
				perClientIdRow, clientIdMessageIdRows, perClientMessageIdCols )

	local clientId = perClientIdRow[ 1 ]

	local messageIdStats = clientId and clientIdMessageIdRows[ clientId ]

	if not messageIdStats then

		return ""

	end

	return wrapInTag( "h4", string.format( "calls from %s", clientId ) ) ..
			makeHTMLTable( perClientMessageIdCols, messageIdStats )

end

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

local _lastStarted = 0

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

local _lastFinished = 0

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

local _clientThreadIdNames = {}

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

local _perClientExecStats = {}

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

local _totalExecStats

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

local function maybePerClientStatsInHTML()

	if type( _totalExecStats ) ~= "table" or not next( _perClientExecStats ) then

		return ""

	end

	local perClientIdRows = {}

	local clientIdMessageIdRows = {}

	for clientId, perClientStats in pairs( _perClientExecStats ) do

			local totalPerClientStat = perClientStats.total

			local clientIdName =
						 _clientThreadIdNames[ clientId ] or clientId

			perClientIdRows[ #perClientIdRows + 1 ] = 
					{
						 clientIdName,
						string.format( "%.2fs", totalPerClientStat.elapsed ),
						 string.format( "%.2fs", 
										totalPerClientStat.count > 0 and 
											totalPerClientStat.elapsed /
												totalPerClientStat.count or
											0 ),
						string.format( "%.2fs", totalPerClientStat.maxElapsed ),
						 string.format( "%s", totalPerClientStat.maxBacklog ),
						 string.format( "%s", totalPerClientStat.count ),
						 string.format( "%.2fs", totalPerClientStat.totalWaitTime ),
						 string.format( "%s", totalPerClientStat.waitCount ),
						 string.format( "%.2fs", 
										totalPerClientStat.waitCount > 0 and 
											totalPerClientStat.totalWaitTime /
												totalPerClientStat.waitCount or
											0 ),
						string.format( "%.2fs", totalPerClientStat.maxWait ),
						totalPerClientStat.elapsed -- extra column for sorting

					}

		local messageIdRows = {}

		for messageId, perClientMessageIdStat in pairs( perClientStats ) do

			messageIdRows[ #messageIdRows + 1 ] =
					{
						 messageId,
						string.format( "%.2fs", perClientMessageIdStat.elapsed ),
						 string.format( "%.2fs", 
										perClientMessageIdStat.count > 0 and 
											perClientMessageIdStat.elapsed /
												perClientMessageIdStat.count or
											0 ),
						string.format( "%.2fs", perClientMessageIdStat.maxElapsed ),
						 string.format( "%s", perClientMessageIdStat.maxBacklog ),
						 string.format( "%s", perClientMessageIdStat.count ),
						 string.format( "%.2fs", perClientMessageIdStat.totalWaitTime ),
						 string.format( "%s", perClientMessageIdStat.waitCount ),
						 string.format( "%.2fs", 
										perClientMessageIdStat.waitCount > 0 and 
											perClientMessageIdStat.totalWaitTime /
												perClientMessageIdStat.waitCount or
											0 ),
						string.format( "%.2fs", perClientMessageIdStat.maxWait ),
						perClientMessageIdStat.elapsed -- extra column for sorting
					}

		end

		clientIdMessageIdRows[ clientIdName ] = messageIdRows

	end

	local perClientIdCols = { "clientId", "elapsed", "aveElapsed",
								"maxElapsed", "maxBacklog", "count",
								"totalWaitTime", "waitCount", "aveWait",
								"maxWait" }

	local perClientIdHTMLTable =
			makeHTMLTable( perClientIdCols, perClientIdRows )

	local perClientMessageIdCols = { "messageId", "elapsed", "aveElapsed",
										"maxElapsed", "maxBacklog", "count",
										"totalWaitTime", "waitCount",
										"aveWait", "maxWait" }
		
	local clientIdMessageIdHTMLTables =
			mapA( makeClientIdMessageIdHTML,
					perClientIdRows, clientIdMessageIdRows,
					perClientMessageIdCols )

	return wrapInTag( 
				"h4", 
				"backlog stats" ) ..
			wrapInTag( 
				"h5", 
				string.format( "curr backlog: %s",
								_totalExecStats.currBacklog ) ) ..
			wrapInTag( 
				"h5", 
				string.format( "max backlog: %s", 
								_totalExecStats.maxBacklog ) ) ..
			wrapInTag( 
				"h5", 
				string.format( "max wait time: %.2fs", 
								_totalExecStats.maxWait ) ) ..
			wrapInTag( 
				"h5", 
				string.format( "total wait time: %.2fs",
								_totalExecStats.totalWaitTime ) ) ..
			wrapInTag( 
				"h4", 
				"calling thread stats" ) ..
			perClientIdHTMLTable ..
			table.concat( clientIdMessageIdHTMLTables, "" )

end

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

local function enableDiagnosticLogForRequestStats()

	if _requestStats then return end

	_requestStats = {}

	-- Don't move the require 'WFDiagnosticLog' out of this function.
	-- 
	-- WFDiagnosticLog spins up promises and reactive, which are both fairly
	-- heavyweight.  We don't want to add them unless the client code has
	-- called enableDiagnosticLogForRequestStats() 

	local WFDiagnosticLog = require 'WFDiagnosticLog'

	WFDiagnosticLog.addLogSectionFunc( function()

		local sortedRequestStats = {}
		local n = 0

		local cols = {
				"messageId",
				"min",
				"max",
				"avg",
				"errorCount",
				"earlyCloseCount",
				"count",
				"total", }

		for _, v in pairs( _requestStats ) do
			n = n + 1
			sortedRequestStats[ n ] = {
				v.messageId,
				string.format( "%.3f", v.min ),
				string.format( "%.3f", v.max ),
				string.format( "%.3f", v.total / v.count ),
				v.errorCount,
				v.earlyCloseCount,
				v.count,
				v.total, -- will format after sort
			}
		end

		table.sort( sortedRequestStats, function( a, b )
			return safeToNumber( a[ #a ] )  > safeToNumber( b[ #b ] ) 
		end )

		for _, v in pairs( sortedRequestStats ) do
			v[ #v ]  = string.format( "%.3f", safeToNumber( v[ #v ] ) )
		end

		local countDisposablesForRawRequests = 0
		for _, _ in pairs( disposablesForRawRequests ) do
			countDisposablesForRawRequests = countDisposablesForRawRequests + 1
		end

		local body =
				string.format( "<h4>GC size: %s</h4>\n<h4>disposablesForRawRequests: %s</h4>\n",
									formatKilobytesAsReadableUnit( collectgarbage( 'count' ) ),
									countDisposablesForRawRequests ) ..
				makeHTMLTable( cols, sortedRequestStats ) ..
				maybePerClientStatsInHTML()

		return {
			name = "Messaging Stats :: " .. WFCurrentThread.DEBUG_getThreadName(),
			id = 'WFRequestDispatcher-' .. WFCurrentThread.DEBUG_getThreadName():gsub( ' :: ', '+' ),
			weight = 500,
			parentSection = WFDiagnosticLog.Sections.PERFORMANCE,
			
			body = body,
		}

	end )

end

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

local _messageHandlers = {}
	-- key (string): message name
	-- value (function): function to be called when this message is received

local benchmarkAndDispatchMessage -- fwd declare

--------------------------------------------------------------------------------
-- dispatchThreadMessage works by looking up the method name that is in the
-- selector string and then running it.
--------------------------------------------------------------------------------

local function dispatchThreadMessage( requestRef,
														  priority,
														  selector,
														  ... )

	if _shouldBenchmark and not _requestStats then

		enableDiagnosticLogForRequestStats()

	end

	requestDispatcherLog:debugf( "receiving request\n  messageId = %s\n  priority = %s\n  values = %s, %s, %s",
								  selector, priority, ... )

	-- Special selector for benchmarking threads
	if selector == '_benchmarkAndDispatchMessage' then

		-- the ... passed to _benchmarkAndDispatchMessage are
		-- 
		-- 		clientThreadId, clientRequestStartTime, targetSelector, ... 
		-- 		
		-- the vararg following targetSelector are arguments to pass to
		-- the targetSelector function

		benchmarkAndDispatchMessage( requestRef, priority, ... )

		return

	end

	local messageHandler = _messageHandlers[ selector ]

	if messageHandler then

		return messageHandler( requestRef, ... )

	elseif selector == '_notifyWhenReady' then

		requestRef:send( 'done', nil, true )
		requestRef:close()

	else

		requestDispatcherLog:errorf( "No message handler defined for request\n  messageId = %s\n  priority = %s\n  values = " .. string.rep( "%s, ", select( '#', ... ) ),
									 selector, priority, ... )

		requestRef:send( 'done', 'error',
						 "No message handler defined for message: " .. selector )

		requestRef:close()

	end

end 

--==============================================================================
-- WFCurrentThread.registerRequestDispatchHandler is called by the lua wrapped
-- native message processor code when it pops a unit of work from its private
-- request queue.
-- 
-- In other words, WFWorkerThread bottlenecks all of its calls through the
-- dispatchThreadMessage function we are registering as the handler
--==============================================================================

WFCurrentThread.registerRequestDispatchHandler( dispatchThreadMessage )

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

local function getNewExecStatEntry()

	return {

			currBacklog = 0,

			maxBacklog = 0,

			count = 0,

			totalWaitTime = 0,

			waitCount = 0,

			maxWait = 0,

			elapsed = 0,

			maxElapsed = 0

		}

end

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

local function incrementStatEntry( statEntry, clientRequestStartTime,
									workerStartTime, workerFinishTime )

	statEntry.count = statEntry.count + 1

	local elapsedTime = workerFinishTime - workerStartTime

	statEntry.maxElapsed = math.max( statEntry.maxElapsed, elapsedTime )

	statEntry.elapsed = statEntry.elapsed + elapsedTime

	-- we can always assume workerStartTime - clientRequestStartTime > 0
	local waitTime = workerStartTime - clientRequestStartTime

	statEntry.totalWaitTime = statEntry.totalWaitTime + waitTime

	statEntry.maxWait = math.max( statEntry.maxWait, waitTime )

	if _lastStarted > 0 then

		if clientRequestStartTime < _lastStarted then
	
			statEntry.waitCount = statEntry.waitCount + 1

			statEntry.currBacklog = statEntry.currBacklog + 1

		elseif clientRequestStartTime < _lastFinished then

			statEntry.waitCount = statEntry.waitCount + 1

			statEntry.currBacklog = 1
			
		else

			statEntry.currBacklog = 0

		end

	end

	statEntry.maxBacklog = 
		math.max( statEntry.currBacklog, statEntry.maxBacklog )

	return statEntry

end

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

local function maybeLogExecStats( clientThreadId, clientThreadName, messageId,
											clientRequestStartTime,
											workerStartTime, workerFinishTime )

	if not clientThreadId then

		return

	end

	if workerStartTime < _lastFinished then

		-- This unexpected and will mess up stats. Ignore.

		return

	end

	if not _clientThreadIdNames[ clientThreadId ] then

		_clientThreadIdNames[ clientThreadId ] = clientThreadName

	end

	local perClientExecStats = _perClientExecStats[ clientThreadId ]

	if not perClientExecStats then

		perClientExecStats = {}

		_perClientExecStats[ clientThreadId ] = perClientExecStats

	end
		
	perClientExecStats[ messageId ] = 
		incrementStatEntry( 
			perClientExecStats[ messageId ] or getNewExecStatEntry(),
			clientRequestStartTime, workerStartTime, workerFinishTime )
											
	perClientExecStats.total =
		incrementStatEntry( 
			perClientExecStats.total or getNewExecStatEntry(),
			clientRequestStartTime, workerStartTime, workerFinishTime )


	_totalExecStats = 
		incrementStatEntry( 
			_totalExecStats or getNewExecStatEntry(),
			clientRequestStartTime, workerStartTime, workerFinishTime )

	_lastStarted = workerStartTime

	_lastFinished = workerFinishTime

end

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

benchmarkAndDispatchMessage = function ( requestRef, priority, clientThreadId,
											clientThreadName,
											clientRequestStartTime, selector,
											... )

	local workerStartTime = debug.timeNow()

	-- call the actual target method
	dispatchThreadMessage( requestRef, priority, selector, ... )

	-- now log some detailed stats about thread execution
	local workerFinishTime = debug.timeNow()

	maybeLogExecStats( clientThreadId, clientThreadName, selector,
						clientRequestStartTime, workerStartTime,
						workerFinishTime )

end

--------------------------------------------------------------------------------
--[[!
	\brief Registers a handler for a specific incoming message.

	\param args (table) A table with these keys:
		- \c message: (string) The name of the message to be handled.
		- \c handler: (function) A callback function to handle the message. You
			must provide either this handler function OR a target object
			containing a handler method, not both.
		- \c target: (object) An object to handle the message. This object must
			contain a method with the same name as the message, to be called
			if the message is received. Provide either this object OR the handler
			function, not both.

	The message handler function or method can return any number of results, which
	are interpreted as follows:

		- If the first result is an Observable, it is kept alive as long as both
			(1) the client is interested and (2) the Observable has not signaled
			error or done state. Any data posted to the observable is passed along
			to the client via the request object.
		- If the first result is a Behavior, it is kept alive as long as the client
			is interested, and any data posted to the node is passed
			to the client via the request object. Any subsequent results
			are ignored.
		- If the first result is a Promise, it is kept alive as long as the client
			is interested. When the promise resolves, the
			status (\c data or \c error) and any additional data is passed
			to the client via the request object. Any subsequent results
			are ignored.
		- If the first result is NOT a Reactive node, all results are passed
			back to the client via the request object, and the request is marked as done.
			All of the results passed through must be transitable (that is, scalar
			values, \c KSData, certain native object references, or tables
			containing only transitable values).
		- If the message handler fails with an error, the error status is
			passed to the client via the request object.
		- If the message handler returns no result, the request is marked as done.
]]--

function WFRequestDispatcher.onMessageDo( args )

	check( 'table' )

	local message = args.message
	assert( type( message ) == 'string' and #message > 0,
			"WFRequestDispatcher.onMessageDo: missing required argument 'message' (must be non-empty string)" )

	local target = args.target
	local func = assert( args.handler or args.target[ args.method or message ],
						  "WFRequestDispatcher:onMessageDo: must specify either handler or target" )

	if not args.simpleResponses then
		loadReactiveSystem()
	end

	_messageHandlers[ message ] = function( request, ... )

		-- This code is called more or less directly by the "we've received a
		-- request" code so it can rely on the fact that we've filtered out
		-- canceled requests at the level above.

		if target then
			sendResponseOrErrorToCallback( args.message, request, func, target, ... )
		else
			sendResponseOrErrorToCallback( args.message, request, func, ... )
		end

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Registers a handler with a reference to the request for a specific incoming message.

	See the documentation of onMessageDo above for documentation of arguments.

	The installed handler/target will receive a function( progressData ) as the first argument.
	This function should be called approximately 100 - 200 times, to enable the display of a smooth
	progress bar with percent resolution in the receiver.

	The values passed into progressData will be emitted as an observable sequence of progressData to the consumer.

	If it makes sense, use a table like { processedCount = i, totalCount = n }, which is the common format
	used by WFProgressReporter. If the totalCount is not meaningful, a format like { percentComplete = x ) is recommended.

	It is recommended to return the last progress value as a return value from the handler or target.

	If necessary, it's also possible to return an Observable from the handler/target, but keep in mind that the
	values emitted from that observable will be concatenated to the progress values emitted previously.
]]--

function WFRequestDispatcher.onMessageWithProgress( args )

	check( 'table' )

	local messageId = args.message
	assert( type( messageId ) == 'string' and #messageId > 0,
			"WFRequestDispatcher.onMessageDo: missing required argument 'message' (must be non-empty string)" )

	local target = args.target
	local func = assert( args.handler or args.target[ args.method or messageId ],
						  "WFRequestDispatcher:onMessageDo: must specify either handler or target" )

	if not args.simpleResponses then
		loadReactiveSystem()
	end

	_messageHandlers[ messageId ] = function( request, ... )

		local progress = function( progressData )
			sendResponseWithData( messageId, nil, 'data', request, progressData )
		end

		-- This code is called more or less directly by the "we've received a
		-- request" code so it can rely on the fact that we've filtered out
		-- canceled requests at the level above.

		if target then
			sendResponseOrErrorToCallback( messageId, request, func, target, progress, ... )
		else
			sendResponseOrErrorToCallback( messageId, request, func, progress, ... )
		end

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Registers a handler for a specific incoming message. (Simple version.)

	\param message (string) The name of the message to be handled.
	\param handler (function) A function to handle the message.

	The message handler function can return any number of results, which
	are interpreted as follows:

		- If the first result is an Observable, it is kept alive as long as both
			(1) the client is interested and (2) the Observable has not signaled
			error or done state. Any data posted to the observable is passed along
			to the client via the request object.
		- If the first result is a Behavior, it is kept alive as long as the client
			is interested, and any data posted to the node is passed
			to the client via the request object. Any subsequent results
			are ignored.
		- If the first result is a Promise, it is kept alive as long as the client
			is interested. When the promise resolves, the
			status (\c data or \c error) and any additional data is passed
			to the client via the request object. Any subsequent results
			are ignored.
		- If the first result is NOT a Reactive node, all results are passed
			back to the client via the request object, and the request is marked as done.
			All of the results passed through must be transitable (that is, scalar
			values, \c KSData, certain native object references, or tables
			containing only transitable values).
		- If the message handler fails with an error, the error status is
			passed to the client via the request object.
		- If the message handler returns no result, the request is marked as done.
]]--

function WFRequestDispatcher.onMessage( message, handler )

	check( 'string', 'function' )

	return WFRequestDispatcher.onMessageDo {
		message = message,
		handler = handler,
	}

end

--------------------------------------------------------------------------------
--[[!
	\brief Enables logging for request operations.
]]--

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

WFRequestDispatcher.enableDiagnosticLogForRequestStats = 
	enableDiagnosticLogForRequestStats

--------------------------------------------------------------------------------
--[[!
	\brief Return a list of the message handlers that have been installed so far.

	\return (table) Array of message handler names.
]]--

function WFRequestDispatcher.getInstalledMessages()

	local result = {}

	for k in pairs( _messageHandlers ) do
		result[ #result + 1 ] = k
	end

	return result

end


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

-- @mark Support for memory leak tool

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

WFRequestDispatcher.onMessageDo {

	message = 'debug_dumpUniverse',
	simpleResponses = true,

	handler = function( options )

		local WFLeakCheck = require 'WFLeakCheck'

		if not options.containers then
			options.containers = {}
		end

		for k, v in pairs( _messageHandlers ) do
			if type( k ) == 'string' then
				k = string.gsub( k, "%..+", "" )
				options.containers[ "MH_" .. k ] = v
			end
		end

		WFLeakCheck.dumpThisUniverseWithOptions( options )

	end,

}

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

return WFRequestDispatcher
