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

WFWorkerThreadTest.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.os
local WFPlatformInfo = require 'WFPlatformInfo'

	-- WF.core.reactive
local KSReactive = require 'KSReactive'

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

	-- WF.core.self_test
local WFAsyncTestCase = require 'WFAsyncTestCase'
local WFRxTestCase = require 'WFRxTestCase'

	-- WF.core.threading
local WFWorkerThread = require 'WFWorkerThread'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTestUtils = require 'KSTestUtils'


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

local WFWorkerThreadTest = {}

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

-- FIXME [scouten 2011-12-16]: Convert to WFAsyncTestCase. Note that this
-- requires implementation of the planned :onErrorE handling.

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

local requestTestHandlerFailed = nil

local requestTestHandler = WFWorkerThread( 'WFRequestTestHandler.lua' )

local currentTest = nil
local expectedTestResults = nil

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

function WFWorkerThreadTest.setUp()

	if requestTestHandlerFailed then
		error( requestTestHandlerFailed )
	end

end


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

-- @mark :request testing (RxLua support)

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

function WFWorkerThreadTest.rxTestRequest_basic( rxTest )

	rxTest:makeVerbose( true )
	local echo = requestTestHandler:request( 'echo', 1, 2, 3 )
	rxTest:expectSingleValue( echo,
							   3, 1, 2, 3 ) -- echo inserts a count at front
	rxTest:expectSingleValue( echo,
							   3, 1, 2, 3 )
		-- Ensures that we can subscribe to a request several times

end

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

function WFWorkerThreadTest.rxTestRequest_dataThenError( rxTest )

	local o = rxTest:observe( requestTestHandler:request( 'sendValueThenError' ) )
	
	o:expectData( 1 )
	o:expectError( '2' )
		
end

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

function WFWorkerThreadTest.rxTestRequest_requestBeforeThreadReady( rxTest )

	local worker = WFWorkerThread( 'WFRequestTestHandler.lua',
									'notInitedYet:' .. tostring( KSIdGenerator.localCounter() ) )

	-- As of this point, worker thread will not have had time to receive
	-- the thread-ready message.

	rxTest:expectSingleValue( worker:request( 'echo', 1, 2, 3 ),
							   3, 1, 2, 3 ) -- echo inserts a count at front

	worker:close()
		
end

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

function WFWorkerThreadTest.rxTestRequest_cancel( rxTest )

	local id = KSIdGenerator.uuid()

	local req = requestTestHandler:request( 'delayAndWatchForCancel', 1, id )

	local gotData
	local unsubscriber = req:subscribe( function() gotData = true end )
	
	rxTest:expectSingleValue( Observable.timer( .25, WFRxScheduler.sharedScheduler() ), 0 )
	
	unsubscriber:dispose()

	rxTest:expectSingleValue( Observable.timer( 1, WFRxScheduler.sharedScheduler() ), 0 )
	
	rxTest:expectSingleValue( requestTestHandler:request( 'didPreviousDelayGetCanceled' ), id )
		
end

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

function WFWorkerThreadTest.rxTestRequest_retryFails( rxTest )

	rxTest:makeVerbose( true )

	local worker = WFWorkerThread( 'WFRequestTestHandler.lua',
									'notInitedYet:' .. tostring( KSIdGenerator.localCounter() ) )

	-- First try with thread that isn't ready yet:

	local counter = 0

	local echoThreadNotReady = worker:request( 'echo', 1, 2, 3 )
	:flatMapLatest( function()
		counter = counter + 1
		if counter == 1 then
			return Observable.throw("error")
		else
			return Observable.empty()
		end
	end)
	:retry( 2 )

	rxTest:expectOnlyError( echoThreadNotReady, "Canceled worker thread requests cannot be restarted: -echo" )
	rxTest:expectEqual( 2, counter )
	
	counter = 0
	
	echoThreadNotReady = worker:_request( 'echo', 1, 2, 3 )
	:flatMapLatest( function()
		counter = counter + 1
		if counter == 1 then
			return Observable.throw("error")
		else
			return Observable.empty()
		end
	end)
	:retry( 2 )

	rxTest:expectOnlyError( echoThreadNotReady, "Canceled worker thread requests cannot be restarted: echo" )
	rxTest:expectEqual( 2, counter )

	-- Now try with thread ready:

	counter = 0

	local echoThreadReady = worker:request( 'echo', 1, 2, 3 )
	:flatMapLatest( function()
		counter = counter + 1
		if counter == 1 then
			return Observable.throw("error")
		else
			return Observable.empty()
		end
	end)
	:retry( 2 )

	rxTest:expectOnlyError( echoThreadReady, "Canceled worker thread requests cannot be restarted: -echo" )
	rxTest:expectEqual( 2, counter )

	counter = 0

	echoThreadReady = worker:_request( 'echo', 1, 2, 3 )
	:flatMapLatest( function()
		counter = counter + 1
		if counter == 1 then
			return Observable.throw("error")
		else
			return Observable.empty()
		end
	end)
	:retry( 2 )

	rxTest:expectOnlyError( echoThreadReady, "Canceled worker thread requests cannot be restarted: echo" )
	rxTest:expectEqual( 2, counter )

	worker:close()

end

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

function WFWorkerThreadTest.rxTestRequest_retryViaFlatMapLatest( rxTest )

	local counter = 0

	rxTest:makeVerbose( true )

	local echo = Rx.Observable.just( true )
	:flatMapLatest( function()
		return requestTestHandler:request( 'echo', 1, 2, 3 )
	end)
	:flatMapLatest( function( ... )

		counter = counter + 1
		if counter < 3 then
			return Rx.Observable.throw("error")
		else
			return Rx.Observable.just( ... )
		end

	end)
	:retry(3)

	rxTest:expectSingleValue( echo, 3, 1, 2, 3 ) -- echo inserts a count at front
	rxTest:expectEqual( 3, counter )

end

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

-- @mark :requestP testing

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

local function checkTestResultsAndNotify( test, ... )

	local n = select( '#', ... )
	local expectedResults = { ... }

	return function( ... )

		if pcall( function( ... )
				KSTestUtils.assertMatchArgs( unpack( expectedResults, 1, n ) )( ... )
			end, ... ) == true
		then
			test:notifyTestDidSucceed()
		else
			test:notifyTestDidFail()
		end

	end

end

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

function WFWorkerThreadTest.testRequestP( test )

	local p = requestTestHandler:requestP( 'echo', 1, 2, 3 )

	p:whenResolvedDo( checkTestResultsAndNotify( test, 'data', 3, 1, 2, 3 ) ):runUntilDone()
	
	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestPTransitWithNil( test )

	local p = requestTestHandler:requestP( 'echo', 1, nil, 3 )

	p:whenResolvedDo( checkTestResultsAndNotify( test, 'data', 3, 1, nil, 3 ) ):runUntilDone()
	
	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestPWithData( test )

	local p = requestTestHandler:requestP( 'echo', "The data vaue" )

	p:whenResolvedDo( checkTestResultsAndNotify( test, 'data', 1, "The data vaue" ) ):runUntilDone()
	
	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestPWithError( test )

	local p = requestTestHandler:requestP( 'sendError', "Expecting this error" )

	p:whenResolvedDo( checkTestResultsAndNotify( test, 'error',
												 "Expecting this error" ) )
		:runUntilDone()
	
	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestPCancelation( test )

	local canceledB = KSReactive.variableB( false )

	WFAsyncTestCase( test )

		:waitFor( function()
			return requestTestHandler:threadIsReadyP()
				:observeP( function( ... )
					debug.print( "Thread yes ready ", ... )
				end )
		end  )
		:continueWhenValueMatches( 'data', true )
		
		:waitFor( function()
		
			local p = requestTestHandler:requestP( 'echo', "Hello!" )

			p = p:observeP( function( ... )
					canceledB:set( select( '#', ... ) == 0 )
				end )
		
			local a1 = p:whenResolvedDo( function() end ):runUntilDone()
		
			a1:close()
			
			return canceledB
			
		end )
		:continueWhenValueMatches( true )
	
end

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

function WFWorkerThreadTest.testRequestPCancelationWithDelay( test )

	local echoDelayP = requestTestHandler:requestP( 'echoDelay', 2, "Hello!" )
	
	local a1 = echoDelayP:whenResolvedDo( function() end ):runUntilDone()
	
	WFAsyncTestCase( test, 4 )
	
		:waitFor( .5 )		 	-- step 1
		
		:waitFor( function()	-- step 2
		
			a1:close()
			
			return echoDelayP
		
		end )
		
		:continueWhenValueMatches() -- no results on cancel
		
end


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

-- @mark :requestE testing

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

local function testCallback( ... )

	local testResults = { ... }
	
	local success = true
	
	if testResults[ 1 ] ~= expectedTestResults.n then
		debug.print( "echo didn't respond with correct # of results" )
		success = false
	end
	
	for i = 1, expectedTestResults.n do
	
		if( testResults[ i + 1 ] ~= expectedTestResults[ i ] ) then
			debug.print( "unexpected result at index ", i, "  actual: ", tostring( testResults[ i + 1 ] ),
								"   expected: ", tostring( expectedTestResults[ i ] ) )
			success = false
		end
	
	end
	
	if success then
		currentTest:notifyTestDidSucceed()
	else
		currentTest:notifyTestDidFail()
	end

end

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

local function testRequestFinishedWithResults( test, _, ... )

	currentTest = test

	expectedTestResults = { ... }
	expectedTestResults.n = select( '#', ... )

	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestE( test )

	testRequestFinishedWithResults( test,
		requestTestHandler:requestE( 'echo', 1, 2, 3 ):ownedOnDataE( test, testCallback ),
		1, 2, 3 )

end

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

function WFWorkerThreadTest.testRequestB( test )

	local requestB = requestTestHandler:requestB {
		message = 'echoDelay',
		arguments = { 1, "delayed response" },
		startingValue = { "nothing", "as of yet" },
	}

	WFAsyncTestCase( test, 4 )
	
		:waitFor( function()
			return requestB
		end )
		:continueWhenValueMatches( "nothing", "as of yet" )
	
		:waitFor( function()
			return requestB
		end )
		:continueWhenValueMatches( "delayed response" )
	
end

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

function WFWorkerThreadTest.testRequestETransitWithNil( test )

	testRequestFinishedWithResults( test,
		requestTestHandler:requestE( 'echo', 1, nil, 3 ):ownedOnDataE( test, testCallback ),
		1, nil, 3 )

end

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

function WFWorkerThreadTest.testRequestEWithData( test )

	testRequestFinishedWithResults( test,
		requestTestHandler:requestE( 'echo', "The data value" )
			:ownedOnDataE( test, testCallback )
			:ownedOnErrorE( test, function() test:notifyTestDidFail() end ),
		"The data value" )

end

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

function WFWorkerThreadTest.testRequestEWithError( test )

	requestTestHandler:requestE( 'sendError', "expecting this error" )
		:ownedOnDataE( test, function() test:notifyTestDidFail() end )
		:ownedOnErrorE( test, function( message )
			if message == 'expecting this error' then
				test:notifyTestDidSucceed()
			else
				test:log( "wrong message: " .. message )
				test:notifyTestDidFail()
			end
		end )

	test:waitForTestToFinish( 2 )

end

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

function WFWorkerThreadTest.testRequestEUpdates( test )

	local counterB = requestTestHandler:requestE( 'counterE', 4, 7 ):holdB( nil )
	
	local values = { }
	local valuesB = counterB:mapB( function( value )
		values[ #values + 1 ] = value
		return unpack( values )
	end )
	
	WFAsyncTestCase( test, 4 )

		:waitFor( valuesB )
		:continueWhenValueMatches( 4, 5, 6, 7 )
		
		-- test for counterB closing?
		
end

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

function WFWorkerThreadTest.testClosedBeforeInitedRequestE( test )

	local KSPrefs = require 'KSPrefs'
	local KSReactivePromiseBridge = require 'KSReactivePromiseBridge'

	local worker = WFWorkerThread( 'WFRequestTestHandler.lua', "baz" )
	KSPrefs.testClosedBeforeInited = "not bar"
	
	local requestE = worker:requestE( 'setPref', "testClosedBeforeInited", "bar" )

	worker:close()
	
	local prefB = KSPrefs.prefB( 'testClosedBeforeInited' )
	WFAsyncTestCase( test, 4 )

		:waitFor( prefB )
		:continueWhenValueMatches( 'bar' )
		:waitFor( KSReactivePromiseBridge.behaviorToPromise( prefB, function( pref )
			return pref == 'bar'
		end ) )
		:whenResolvedDo( function()
			KSPrefs.testClosedBeforeInited = nil
			requestE:close()

			if WFPlatformInfo.isWindows then
				local KSPrefsImpLua = require 'KSPrefsImpLua'
				KSPrefsImpLua.deletePrefsFile()
			end

		end )

end

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

function WFWorkerThreadTest.testClosedBeforeInitedRequestP( test )

	local worker = WFWorkerThread( 'WFRequestTestHandler.lua', "buzz" )
	
	local echoP = worker:requestP( 'getProcessorId', "GOANDY" )
	worker:close()
	
	WFAsyncTestCase( test, 4 )

		:waitFor( echoP )
		:continueWhenValueMatches( 'data', 'buzz', 'GOANDY' )
	
end

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

local function initWorkerThread( rxTest, workerThread )

	-- Worker threads are not initialized until a request is made.

	local throwaway = workerThread:request( 'echo', rxTest:testName() )
	rxTest:expectSingleValue( throwaway, 1, rxTest:testName() )

end

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

local KSDate = require 'KSDate'
local KSDate_timeNowForAnimation = KSDate.timeNowForAnimation

function WFWorkerThreadTest.DISABLED_rxTestRequest_testPerformance( rxTest )

	rxTest:setTimeLimit( 100 )
	
	local size = 1000
	
	local echoTable = {}
	
	local echo = function( ... )
		return Observable.just( select( '#', ... ), ... )
	end 
	
	local start = KSDate_timeNowForAnimation()
	
	for i=1,size do
		-- Establish a baseline
		rxTest:expectSingleValue( echo( 1, 2, 3, i ), 4, 1, 2, 3, i )
	end
	
	local baseline = KSDate_timeNowForAnimation() - start
	
	initWorkerThread( rxTest, requestTestHandler )
	
	start = KSDate_timeNowForAnimation()
	
	for i=1,size do
		echoTable[ i ] = requestTestHandler:request( 'echo', 1, 2, 3, i )
	end
	
	for i=1,size do
		rxTest:expectSingleValue( echoTable[ i ], 4, 1, 2, 3, i )
	end

	local requestPerf = KSDate_timeNowForAnimation() - start

	-- On both Mac and Windows, the worker thread request takes about 4x longer
	-- than the non-worker thread request
	
	rxTest:expectEqual( requestPerf / baseline < 5, true )

end

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

function WFWorkerThreadTest.DISABLED_rxTestMultipleThreads_testPerformance( rxTest )

	rxTest:setTimeLimit( 100 )
	
	local n = 100
	
	local echoTable = {}
	local workers = {}
	
	---------------------------------------------------------------------------
	
	-- Establish a baseline of n requests on one thread.
	
	initWorkerThread( rxTest, requestTestHandler )
	
	local start = KSDate_timeNowForAnimation()
	
	for i=1,n do
		echoTable[ i ] = requestTestHandler:request( 'echo', rxTest:testName(), i )
	end
	
	for i=1,n do
		rxTest:expectSingleValue( echoTable[ i ], 2, rxTest:testName(), i )
		echoTable[ i ] = nil
	end
	
	local baseline = KSDate_timeNowForAnimation() - start
	
	---------------------------------------------------------------------------
	
	-- Gather performance data on same number of requests but one per thread.
	
	for i=1,n do
		workers[ i ] = WFWorkerThread( 'WFRequestTestHandler.lua', tostring( i ) )
		initWorkerThread( rxTest, workers[ i ] )
	end
	
	start = KSDate_timeNowForAnimation()
	
	for i=1,n do
		echoTable[ i ] = workers[ i ]:request( 'echo', rxTest:testName(), i )
	end
	
	for i=1,n do
		rxTest:expectSingleValue( echoTable[ i ], 2, rxTest:testName(), i )
	end

	local requestPerf = KSDate_timeNowForAnimation() - start

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

	for i=1,n do
		workers[ i ]:close()
		workers[ i ] = nil
		echoTable[ i ] = nil
	end
	
	collectgarbage()
	collectgarbage()
	collectgarbage()

	-- On both Mac and Windows, one request per n threads takes about the same amount of time
	-- as n requests on 1 thread.

	rxTest:expectEqual( requestPerf / baseline < 1.1 and requestPerf / baseline > 0.9 , true )
	
end

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

function WFWorkerThreadTest.DISABLED_rxTestMultipleThreadsMultipleRequests_testPerformance( rxTest )

	rxTest:setTimeLimit( 100 )
	
	local n = 500
	
	n = math.sqrt( n )
	
	local echoTable = {}
	local workers = {}
	
	---------------------------------------------------------------------------
	
	-- Establish a baseline of n requests on one thread.
	
	initWorkerThread( rxTest, requestTestHandler )
	
	local start = KSDate_timeNowForAnimation()
	
	for i=1,n*n do
		echoTable[ i ] = requestTestHandler:request( 'echo', rxTest:testName(), i )
	end
	
	for i=1,n*n do
		rxTest:expectSingleValue( echoTable[ i ], 2, rxTest:testName(), i )
		echoTable[ i ] = nil
	end
	
	local baseline = KSDate_timeNowForAnimation() - start
	
	---------------------------------------------------------------------------
	
	-- Gather performance data on same sqrt(n) requests on sqrt(n) threads.
	
	for i=1,n do
		workers[ i ] = WFWorkerThread( 'WFRequestTestHandler.lua', tostring( i ) )
		initWorkerThread( rxTest, workers[ i ] )
	end
	
	start = KSDate_timeNowForAnimation()
	
	for i=1,n do
		for j=1,n do
			echoTable[ i .. ':' .. j ] = workers[ j ]:request( 'echo', rxTest:testName(), i, j )
		end
	end
	
	for i=1,n do
		for j=1,n do
			rxTest:expectSingleValue( echoTable[ i .. ':' .. j ], 3, rxTest:testName(), i, j )
		end
	end

	local requestPerf = KSDate_timeNowForAnimation() - start

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

	for i=1,n do
		workers[ i ]:close()
		workers[ i ] = nil
	end
	
	collectgarbage()
	collectgarbage()
	collectgarbage()

	-- On both Mac and Windows, sqrt(n) requests per sqrt(n) threads takes about the same amount
	-- of time as n requests on 1 thread.
	
	-- In other words, 10 requests on 10 threads takes the same amount of time as
	-- 100 requests on 1 thread.
	
	rxTest:expectEqual( requestPerf / baseline < 1.1 and requestPerf / baseline > 0.9 , true )

end

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

return WFRxTestCase.wrapSuite( 'WFWorkerThreadTest', WFWorkerThreadTest )
