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

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

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

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

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

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

	-- WF.core.rxlua
local WFRxEventBridge = require 'WFRxEventBridge'
local WFRxPromiseBridge = require 'WFRxPromiseBridge'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCapabilities = require 'WFCapabilities'
local WFCloseSupport = require 'WFCloseSupport'
local WFDebugTraceback = require 'WFDebugTraceback'
local WFObjectProxy = require 'WFObjectProxy'


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

local logger = WFLogger 'WFWorkerThread'
--	:setLevel 'debug'

local check = assert( WFArguments.check )

local asKSEvent = WFRxEventBridge
local asKSPromise = WFRxPromiseBridge

local WFCloseSupport_hasBeenOrWillBeClosed = WFCloseSupport.hasBeenOrWillBeClosed
local WFCloseSupport_makeCloser = assert( WFCloseSupport.makeCloser )

local assert = assert
local select = select
local error = error

local observerForRawRequestRef = {}

local string_format = assert( string.format )

local plusByte = string.byte('+')
local minusByte = string.byte('-')

local WFWorkerThreadClass

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

local CHECK_REQUEST_CLOSURE = _G.DEBUG

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

local GCMonitor = Rx.Class 'WFWorkerThreads_GCMonitor' {

	_case = false,
	_worker = false,
	_message = false,

	__tostring = function( self )
		return string_format( "%s( %s : %s )",
				tostring( self._case or nil ),
				tostring( self._worker or nil ),
				tostring( self._message or nil ) )
	end,
		
	__init = function( self, case, worker, message )
		self._case = case
		self._worker = worker
		self._message = message
	end,
		
	WFCloseSupport.addCloseSupportMethods

}

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

local function makeGCMonitor( case, worker, message, backtrace )

	-- The _notifyWhenReady and close messages can trigger a GC warning if the worker
	-- thread object gets created, but then the client thread (i.e. this one) gets closed
	-- down quickly itself. I'll give these messages a pass, but I don't want to set
	-- that precedent globally.

	if not CHECK_REQUEST_CLOSURE
		or message == '_notifyWhenReady'
		or message == 'close'
	then
		return
	end
	
	local obj = GCMonitor( case, worker, message )
	
	return WFCloseSupport_makeCloser( obj, nil, backtrace )

end


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

-- @mark raw observer request processing

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

local _clientThreadName

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

local function niceThreadName()

	local rawThreadName = WFCurrentThread.DEBUG_getThreadName()

	if type( rawThreadName ) ~= "string" then

		return '( unknown )'

	end

	local shortThreadId = rawThreadName:match( 'id = *"(%x+)"' )

	if shortThreadId == nil then

		shortThreadId = rawThreadName:match( '"id":"(%x+)"' )

	end

	if not shortThreadId then

		return rawThreadName

	end

	local threadPrefix = rawThreadName:match( '([%a. :]+)' )

	if not threadPrefix then

		return rawThreadName

	end

	return string.format( "%s%s", threadPrefix, shortThreadId )

end

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

local _isThisThreadTheMainMessageProcessor

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

local _responsePerMessageDataCounts = {}

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

local _responsePerMessageDoneCounts = {}

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

local _responseCount = 0

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

local function maybeLogPerThreadResponseCounts( responseType, message )

	_clientThreadName = _clientThreadName or niceThreadName()

	if _isThisThreadTheMainMessageProcessor == nil then

		_isThisThreadTheMainMessageProcessor = 
			string.find( _clientThreadName, "NmbDaemonSideRtmfpEndpoint" )

	end

	if not _isThisThreadTheMainMessageProcessor then

		return

	end

	_responseCount = _responseCount + 1

	local countMap = responseType == "data" and 
						_responsePerMessageDataCounts or
						_responsePerMessageDoneCounts

	countMap[ message ] = 
		( countMap[ message ] or 0 ) + 1

	if _responseCount % 50 == 0 then

		for _, respType in ipairs( { "data", "done" } ) do

			local cnts 

			local countMap2 = respType == "data" and 
								_responsePerMessageDataCounts or
								_responsePerMessageDoneCounts

			for m, c in pairs( countMap2 ) do

				cnts = 
					cnts and 
						string.format( "%s,\n%s = %s", cnts, m, c ) or
						string.format( "%s = %s", m, c )

			end

			logger:infof( "%s response counts:\n%s", respType, cnts )

		end

	end

end

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

local function forwardResponseToObserver( observer, rawRequestRef, requestStatus,
										  statusSelector, ... )

	logger:debugf( "processing observer response:\n  message %s\n  "
				   .. "requestStatus = %s\n  statusSelector = %s\n  "
				   .. "params = %s, %s, %s, %s",
				   observer._message,
				   requestStatus,
				   statusSelector,
				   ... )

	-- If request was orphaned, signal that as an error and close event stream.
	
	if requestStatus == 'orphaned' then
		observer:onError( "Request was orphaned by worker thread" )
		return
	end
	
	local nArgs = select( '#', ... )

	-- Post results from this response. Special case: If no extra values (... is empty)
	-- and the worker thread is saying it's done, don't post an empty result before closing.
	
	if statusSelector == 'error' then

		if nArgs > 0 then
			observer:onError( ... )
		else
			observer:onError( "< Unknown Error >" )
		end
		
		observerForRawRequestRef[ rawRequestRef ] = nil

	elseif requestStatus ~= 'done' or nArgs > 0 then

		if logger.will_info then

			maybeLogPerThreadResponseCounts( "data", observer._message )
						
		end

		observer:onNext( ... )

	end

	-- If worker thread is done, signal Observer is done.

	if requestStatus == 'done' then

		if logger.will_info then

			maybeLogPerThreadResponseCounts( "done", observer._message )
						
		end

		observer:onCompleted()
		observerForRawRequestRef[ rawRequestRef ] = nil

	end
	
end


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

local function createSingleSubscriberObservable( subscribeFn, disposeFn, message )

	local didSubscribe = false

	return Observable.create( function( observer )

		if didSubscribe then

			local errStr =  string.format( "Canceled worker thread requests cannot be restarted: %s",
											  message )

			logger:error( errStr )

			error( errStr )

		end

		didSubscribe = true

		if subscribeFn then
			subscribeFn( observer )
		else
			observer:onNext()
			observer:onCompleted()
		end

		return disposeFn

	end )

end

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

local function maybeCreateMessageProcessorRequestImp( messagingProcessor, ... )

	return WFMessagingRequest.create( messagingProcessor, ... )

end

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

local _benchmarkAndDispatchMessage = "_benchmarkAndDispatchMessage"

--------------------------------------------------------------------------------
											   
local _plus_benchmarkAndDispatchMessage = "+_benchmarkAndDispatchMessage"

--------------------------------------------------------------------------------
											   
local _minus_benchmarkAndDispatchMessage = "-_benchmarkAndDispatchMessage"

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

local _clientThreadId

--------------------------------------------------------------------------------
											   
local function maybeCreateBenchmarkedMessageProcessorRequest( 
				messagingProcessor, message, args )

	_clientThreadName = _clientThreadName or niceThreadName()

	-- clientThreadId is optimized for quick lookup in lua tables
	_clientThreadId = _clientThreadId or 
						( #_clientThreadName <= 32 and
							_clientThreadName or
							require "KSIdGenerator".uuidInLowercaseHex() )

	local messageFirstChar = message:sub( 1, 1 )

	local messageWithoutSpecialFirstChar = 
			( messageFirstChar == '+' or messageFirstChar == '-' ) and
			message:sub( 2 ) or message

	local benchmarkAndDispatchMessage =
			( messageFirstChar == '+' and 
				_plus_benchmarkAndDispatchMessage ) or
			( messageFirstChar == '-' and
				_minus_benchmarkAndDispatchMessage ) or
			_benchmarkAndDispatchMessage

	-- this sends the threadName and startTime to the worker thread universe
	-- which uses those values to benchmark before forwarding the call to the
	-- desired function.
	return maybeCreateMessageProcessorRequestImp( 
				messagingProcessor, 
				benchmarkAndDispatchMessage,
				_clientThreadId,
				_clientThreadName,
				debug.timeNow(),
				messageWithoutSpecialFirstChar,
				unpack( args, 1, args.n ) )

end

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

local function maybeCreateMessageProcessorRequest( messagingProcessor,
											  message, args )

	return maybeCreateMessageProcessorRequestImp( 
				messagingProcessor, 
				message,
				unpack( args, 1, args.n ) )

end

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

local _shouldBenchmark = 
		WFCapabilities.getCapability( "WFWorkerThread_shouldBenchmark" )

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

-- Important: This private function assumes that mySelf is already a
-- _relay of the worker thread.

local function request( mySelf, messagingProcessor, message, args )

	assert( messagingProcessor )

	args = args or {}
	
	local didDispose = false
	local request, rawRequestRef

	local backtrace

	if CHECK_REQUEST_CLOSURE then
		backtrace = message .. tostring( WFDebugTraceback.traceback { 'WFCloseSupport.lua', 'KSPromises.lua', 'Class.lua' } )
	end

	local monitor

	local function dispose()

		logger:debugf( "DISPOSING observable request( %s, %s, %s, %s ) via %s",
					   messagingProcessor, message, args[ 1 ], args[ 2 ],
					   ':traceback:' )

		didDispose = true
		
		if request then
			request:close()
			request = nil
		end
		
		if rawRequestRef then
			observerForRawRequestRef[ rawRequestRef ] = nil
			rawRequestRef = nil
		end

		if monitor then
			monitor:close()
			monitor = nil
		end
		
		if mySelf then
			mySelf:close()
			mySelf = nil
		end
		
		messagingProcessor = nil

	end

	local function wireUpObserver( observer )
	
		if didDispose then
			return
		end

		monitor = makeGCMonitor(
			'request',
			mySelf._initScriptName,
			message,
			backtrace )

		logger:debugf( "sending observable request( %s, %s, %s, %s )",
					   messagingProcessor, message, unpack( args, 1, args.n ) )


		local maybeCreateMessageProcessorRequestFunc =
				( _shouldBenchmark or
					mySelf._shouldBenchmarkPerClient ) and
					maybeCreateBenchmarkedMessageProcessorRequest or
					maybeCreateMessageProcessorRequest
	
		local success, maybeMessageProcessorRequest = 
					pcall( maybeCreateMessageProcessorRequestFunc,
							messagingProcessor, message, args )
		
		if not success then
	
			local errmessage = string_format( "WFWorkerThread:request: Attempting to send a request "
											.. "to a closed messaging processor?"
											.. "\n\n   messagingProcessor = %s\n\n   "
											.. "messageId = %s\n\n   error = %s",
											messagingProcessor, message, maybeMessageProcessorRequest )
			
			logger:error( errmessage )
			
			observer:onError( errmessage )
			
			return
	
		end
	
		request = maybeMessageProcessorRequest
		rawRequestRef = request:getRawRequestRef()

		rawset( observer, '_message', message )

		observerForRawRequestRef[ rawRequestRef ] = observer
		
	end

	return createSingleSubscriberObservable( wireUpObserver, dispose, message )
	
end


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

-- @mark raw message response handling

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

local function handleResponse( rawRequestRef, requestStatus, statusSelector, ... )

	local observer = observerForRawRequestRef[ rawRequestRef ]
	
	if observer then
		forwardResponseToObserver( observer, rawRequestRef, requestStatus, statusSelector, ... )
		return
	end
	
end

WFCurrentThread.registerResponseDispatchHandler( handleResponse )
 
--============================================================================--

-- @mark raw event processor loading

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

local function assertNotClosed( self, methodName )

	if WFCloseSupport_hasBeenOrWillBeClosed( self ) then
		error( string.format( "WFWorkerThread:%s: thread %s %s has been closed\n\n"
							  .. "created in thread %s\n\n%s",
							  methodName, self._initScriptName,
							  self._optArg,
							  WFCurrentThread.DEBUG_getThreadName(),
							  self._creationTraceback ), 3 )
	end

end


--============================================================================--
--[[!
	\brief An object of this type acquires a worker thread instance and releases it
	when no longer needed.
	
	This class wraps the underlying native APIs and is the preferred mechanism
	for accessing worker threads (also called event processors).
--]]

-- @mark WFWorkerThread class

local workerThreadMethods = {

	_threadIsReady = false,
		-- (Observable replay boolean) set to true when thread is ready

	_messagingProcessor = false,
		-- (WFMessagingProcessor) filled in once MP has confirmed that is has
		-- loaded.
	
	_initScriptName = false,
		-- (string) path by which this MP was initialized
	
	_optArg = false,
		-- (string) optional specialization argument for MP
	
	_creationTraceback = false,
		-- (string) in debug builds, traceback from which this object was created
	
	_shouldBenchmarkPerClient = false,
		-- (boolean) when true, all requests sent to the worker thread will
		-- benchmark backpressure in WFRequestDispatcher
	
	WFCloseSupport.addCloseSupportMethods,

}

--------------------------------------------------------------------------------
--[[!
	\brief Constructs a worker thread for a named initialization script and
		optional specialization.

	Worker thread instances are shared and reference-counted. If two client
	threads specify a worker thread using this name and specialization pair,
	they address the same actual worker thread. When all client references
	are closed or garbage-collected, the worker thread terminates.
 
 	\param initScriptName The name of a Lua script.

	\param optArg A specialization that distinguishes this thread from any other
		that is initialized with the same script. Must be \c nil or a string.

	\return A reference to the new worker thread.
]]--

function workerThreadMethods:__init( initScriptName, optArg )

	check( 'WFWorkerThread', 'string', '?string' )

	-- FIXME [scouten 2014-07-09 via jandrew]: This causes a crash in
	-- the WFModelsTest test suite.
	--WFCloseSupport.mustClose( self )

	self._initScriptName = initScriptName
	self._optArg = optArg
	
	if initScriptName == ':self:' then
		error( "WFWorkerThread: using ':self:' to reference the current message processor is deprecated", 2 )
	end
	
	if _G.DEBUG then
		self._creationTraceback = debug.traceback( "created via", 4 )
	end

end

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

function workerThreadMethods:__tostring()

	local desc = self._initScriptName

	if self._optArg then

		desc = string.format( "%s :: %s", desc, self._optArg )

	end

	return string.format( "WFWorkerThread( %s )", desc )

end

--------------------------------------------------------------------------------
-- threadIsReady() is a replay stream that sends a `_notifyWhenReady` message to
-- the worker thread once per this thread. 
--
-- `_notifyWhenReady` will throw if there are errors raised from the worker
-- thread's `initMessageProcessor` method.
--
-- Because this is a replay stream, if `_notifyWhenReady` throws, all subsequent
-- calls to `threadIsReady` will throw without sending a message to the worker
-- thread.
--
-- threadIsReady() is called before sending any `self:request( apiName, ...)`
-- calls to the worker thread, so the combination of replay and "throw when
-- worker thread init fails" means a failed init will stop this thread from
-- sending any more messages to the broken worker thread.
--------------------------------------------------------------------------------

local function threadIsReady( self )
	
	if not self._threadIsReady then

		local selector = assert( self._initScriptName )
	
		if self._optArg then
			selector = string.format( '%s :: %s', self._initScriptName, self._optArg )
		end
	
		local resultMP = WFMessagingProcessor.createForId( selector )
	
		self._threadIsReady = 
			request( self:_relay(), resultMP, '_notifyWhenReady', { true } )
				:catch(
					function( errStr )

						-- This error will be replayed on all subsequent
						-- `self:request( apiName, ... )` calls, so we want to
						-- include "failed to init" messaging in the error to
						-- make it clear there error happened on init, not when
						-- executing the the `apiName` method.
						--
						-- These errors tend to cascade and be repeated, so we add extra
						-- newlines so they stand out in the log
						local failedToInitErrStr =
							string.format(
								"\n\nFailed to initialize message processor %s:\n\n\t%s\n\n",
								selector, errStr )

						-- this reraises the error.
						return Rx.Observable.throw( failedToInitErrStr )

					end )
				:map( 
					function()
						if WFCloseSupport_hasBeenOrWillBeClosed( self ) then
							resultMP:close()
							return false
						else
							logger:debugf( "Thread is now ready :: %s", selector )
							self._messagingProcessor = resultMP
							return true
						end
					end )
				:replay( nil, 1 )

		-- `:connect()` fires off the `_notifyWhenReady` message immediately
		-- before downstream subscription and stashes the result in the replay
		self._threadIsReady:connect()

	end
	
	return self._threadIsReady

end

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

function workerThreadMethods:close()

	check( 'WFWorkerThread' )

	if self._messagingProcessor then
		self._messagingProcessor:close()
		self._messagingProcessor = " < CLOSED > "
	end

end

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

function workerThreadMethods:_relay()

	check( 'WFWorkerThread' )

	-- Because WFWorkerThread is a WFObjectProxy, creating a new one here
	-- will actually just give us another proxy to this

	return WFWorkerThreadClass( self._initScriptName, self._optArg )

end

--------------------------------------------------------------------------------
--[[!
	\brief This Observable notifies you when the worker thread has finished
		loading and is ready to accept requests.

	You can attach actions to this if you need specialized load success/failure
	notification. By default, the class logs a warning in the event of a load failure.
	
	\return <code>Observable< true ></code> fires when the worker thread has loaded
		successfully. If the worker thread fails to load, sends the error message
		from the thread.
]]--

function workerThreadMethods:threadIsReady()

	check( 'WFWorkerThread' )

	assertNotClosed( self, 'threadIsReady' )

	return threadIsReady( self )

end

--------------------------------------------------------------------------------
--[[!
	\brief This Promise notifies you when the worker thread has finished
		loading and is ready to accept requests.

	You can attach actions to this if you need specialized load success/failure
	notification. By default, the class logs a warning in the event of a load failure.
	
	\return Resolves to \c true when the worker thread loads successfully.
		If the worker thread fails to load, resolves with the error message
		from the thread.
]]--

function workerThreadMethods:threadIsReadyP()

	check( 'WFWorkerThread' )
	
	return self:threadIsReady():call( asKSPromise )

end

--------------------------------------------------------------------------------
--[[!
	\brief Initiates a request to this worker thread and returns the result(s)
		via an Observable.
	
	If the worker thread has not finished loading, the request waits until
	loading is complete. If the worker thread fails to load, the Observable
	will terminate with the error generated by the load attempt.
	
	\warning You must ensure that there is exactly one subscriber to this
		object. For requests that have side effects, you must ensure that
		the subscriber isn't disposed prematurely.
	
	\param message The name of the message to send to this worker thread.
	\param ... Other arguments to pass to the message.
	
	\return An \c Rx.Observable object with the result(s) of the message.
]]--

function workerThreadMethods:_request( message, ... )

	check( 'WFWorkerThread', 'string' )

	assertNotClosed( self, 'request' )
	
	local messagingProcessor = self._messagingProcessor
	local mySelf = self:_relay()
	local args = { n = select( '#', ... ), ... }
	
	if messagingProcessor then

		return request( mySelf, messagingProcessor, message, args )
			:replay( nil, 1 )
			:refCount()

	else

		logger:debug( "Thread not ready, waiting" )

		return createSingleSubscriberObservable( nil, nil, message )
			:flatMapLatest( function()
				return threadIsReady( self )
			end )
			:flatMapLatest( function()
				logger:debug( "Thread now ready, resuming request" )
				return request( mySelf, self._messagingProcessor, message, args )
			end )
			:replay( nil, 1 )
			:refCount()

	end

end

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

--[[!
	\brief Initiates a request to this worker thread and returns the result(s)
		via an Observable.
	
	\warning Unless the priority (+, -) is already specified in the message,
	this function will add a - to it. Not doing so can result in some unexpected 
	LIFO behavior. - will enforce FIFO.
	
	If the worker thread has not finished loading, the request waits until
	loading is complete. If the worker thread fails to load, the Observable
	will terminate with the error generated by the load attempt.
	
	\warning You must ensure that there is exactly one subscriber to this
		object. For requests that have side effects, you must ensure that
		the subscriber isn't disposed prematurely.
	
	\param message The name of the message to send to this worker thread.
	\param ... Other arguments to pass to the message.
	
	\return An \c Rx.Observable object with the result(s) of the message.
]]--
function workerThreadMethods:request( message, ... )
	
	check( 'WFWorkerThread', 'string' )
	-- SPAM [matthewj 2016-01-14]: The underlying WFMessaging calls default
	-- to useFastEnqueue = false, useFastInitialResponse = true. The latter 
	-- will cause the worker thread to process LIFO which can result in the 
	-- inversion of the order of responses, producing unexpected ordering of
	-- the observable results. In order to avoid that, by default we add 
	-- the - selector modifier which causes useFastInitialResponse = false 
	-- and ensures FIFO processing of the request. Specifying a message with 
	-- a selector modifier of + will still produce useFastEnqueue = true, 
	-- useFastInitialResponse = true and if the previous false, true behavior 
	-- REALLY is necessary you can call :_request directly instead.
	local firstByte = message:byte( 1 )
	if firstByte ~= plusByte and firstByte ~= minusByte then
		message = '-' .. message
	end

	return self:_request( message, ... )
end


--------------------------------------------------------------------------------
--[[!
	\brief Initiates a message request to this worker thread and returns the result
		through a promise.
	
	If the worker thread has not finished loading, the request waits until
	loading is complete. If the worker thread fails to load, the result of
	the promise is the failure message generated by the load attempt.
	
	\param message The name of the message to send to this worker thread.
	\param ... Other arguments to pass to the message.

	\return A \c KSPromise object with the result of the message.
]]--

function workerThreadMethods:requestP( message, ... )

	return self:_request( message, ... )
		:call( asKSPromise )

end

--------------------------------------------------------------------------------
--[[!
	\brief Initiates a request to this worker thread and returns the result
	or results through an event stream.
	
	If the worker thread has not finished loading, the request waits until
	loading is complete. If the worker thread fails to load, the failure message
	generated by the load attempt is posted as an error event on this stream,
	and the stream is then closed.
	
	\warning Do <b>NOT</b> use this function for requests that have side effects.
	The lifetime of such requests is unpredictable and can led to subtle
	silent-failure bugs. Use \c :requestP() instead.
	
	\param message The name of the message to send to this worker thread.
	\param ... Other arguments to pass to the message.
	
	\return A \c KSEventStream object with the result or results of the message.
]]--

function workerThreadMethods:requestE( message, ... )

	return self:_request( message, ... )
		:call( asKSEvent )

end

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

local function collectArgs( t, field )

	local v = t[ field ]

	if type( v ) == 'table' then
		return v, v.n or #v
	elseif v == nil then
		return {}, 0
	else
		return { v }, 1
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Like \c requestE, but starts with an initial value.
	
	\param params A table of parameters
		* message: The name of the message to send to this worker thread.
		* arguments: (number, string, boolean, nil, kNilValue or array) Other arguments to pass,
			packed as an array.
			If it is a sparse array, include a field 'n' to indicate the total # of arguments.
			As a convenience, if it is a value (not a table) then it will be treated as the
			single argument (and wrapped in an array for you).
			If it is nil then no arguments are sent.  To use nil (rather than
			then empty tuple) use WFWorkerThread.kNilValue.
			To send a table argument, you *must* wrap it in an array, otherwise it will get unpacked.
		* startingValue: (number, string, boolean, nil, kNilValue or array) starting value.
			It is also handled just like \c arguments.
	
	\return A \c Behavior with the result or results of the message.
]]--

function workerThreadMethods:requestB( params )

	check( 'WFWorkerThread', 'table' )
	
	local message = params.message
	local arguments, nArgs = collectArgs( params, 'arguments' )

	local E = self:requestE( message, unpack( arguments, 1, nArgs ) )

	local startingValue, nVals = collectArgs( params, 'startingValue' )

	return E:closingHoldB( unpack( startingValue, 1, nVals ) )

end
--------------------------------------------------------------------------------
--[[!
	\brief Enables back pressure benchmarking on this thread. Stats will
			dumped to diagnostic log.
]]--

function workerThreadMethods:enablePerClientBenchmarking()

	check( 'WFWorkerThread' )

	self._shouldBenchmarkPerClient = true

end

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

WFWorkerThreadClass = Rx.Class 'WFWorkerThread'( workerThreadMethods )

rawset( WFWorkerThreadClass, 'kNilValue', { n = 1 } )

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

local function validateProxy( _, initScriptName, optArg )
	
	local selector = assert( initScriptName )

	if optArg then
		selector = string.format( '%s :: %s', initScriptName, optArg )
	end

	return selector

end

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

-- FIXME [scouten 2014-07-09 via jandrew]: This causes a crash in
-- the WFModelsTest test suite, but only on the build server.

--[[ WFWorkerThreadClass = WFObjectProxy( 'WFWorkerThread',
									 WFWorkerThreadClass,
									 validateProxy,
									 true )
]]

WFWorkerThreadClass = WFObjectProxy( 'WFWorkerThread',
									  WFWorkerThreadClass,
									  validateProxy )

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

return WFWorkerThreadClass
