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

WFRtmfpClientPerformanceTest.lua

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

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

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

	-- WF.rtmfp
local WFRtmfpClient = require 'WFRtmfpClient'

local KSPathUtils = require 'KSPathUtils'

local WFRxTestCase = require 'WFRxTestCase'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

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

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

local WFRtmfpClientPerformanceTest = {}

local runUntilDone = WFRxRunUntilDone

-- RTMFP Echo Server. Lives on sj1010122073213.corp.adobe.com
-- You can create your own instance of the echo server to use for testing purposes - code
-- is found on blackroom/daemon in the echo.server branch. You'll find the rtmfp_server
-- xcode project in the rtmfp sub-folder.

local serverAddr = "sj1010122073213.corp.adobe.com"
local serverPort = 1979
local _rtmfpClient = false
local _flow = false
local testWaitTime = 60

-- A special option to prevent the echo server from echoing.
-- This makes the results more accurate.

local noEchoMetadata = {}
noEchoMetadata.no_echo = true

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

function WFRtmfpClientPerformanceTest.setUpClass()

	if not _rtmfpClient then
		_rtmfpClient = WFRtmfpClient()
	end
	
	-- Open a flow to the server
	_flow = _rtmfpClient:openFlow( serverAddr, serverPort )
	
end

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

-- The underscore in the name is used to make sure that this test test runs
-- after all of the other tests. I'm using this instead of the tearDownClass
-- method because it allows me to wait on the closeP method before, where as
-- tearDownClass does not.

function WFRtmfpClientPerformanceTest.rxTest_CloseInstance_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	_flow:shutdown()
	
		:flatMapLatest( function()
			return _rtmfpClient:close()
		end )
		
		:map( function()
			_rtmfpClient = false
			return true
		end )
		
end

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

function WFRtmfpClientPerformanceTest.rxTestRequest_BigData_1mb_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.45 seconds

	local testDataPath = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testDataPath .. '/rtmfp/1MB.txt'
	local file = io.open( filePath, 'r' )
	
	local testData = file:read("*all")
	file:close()
			
	rxTest:expectSingleValue( _flow:requestData( testData )
		:map( function()
			return true
		end ), true )

end

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

function WFRtmfpClientPerformanceTest.rxTestRequest_BigData_10kb_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.01 seconds

	local testDataPath = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testDataPath .. '/rtmfp/10Kb.txt'
	local file = io.open( filePath, 'r' )
	
	local testData = file:read("*all")
	file:close()
			
	rxTest:expectSingleValue( _flow:requestData( testData )
		:map( function()
			return true
		end ), true )

end

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

function WFRtmfpClientPerformanceTest.rxTestSend_BigData_1MB_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.25 seconds

	local testData = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testData .. '/rtmfp/1MB.txt'
	local file = io.open( filePath, 'r' )
	local content = file:read("*all")
	file:close()

	rxTest:expectOnlyCompleted( _flow:send( content, noEchoMetadata ) )

end

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

function WFRtmfpClientPerformanceTest.rxTestSend_BigData_10kb_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.01 seconds

	local testData = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testData .. '/rtmfp/10Kb.txt'
	local file = io.open( filePath, 'r' )
	local content = file:read("*all")
	file:close()
	
	rxTest:expectOnlyCompleted( _flow:send( content, noEchoMetadata ) )

end

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

function WFRtmfpClientPerformanceTest.rxTestSend_20_uniqueflow_testPerformance( rxTest )
	
	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.45 seconds
	
	local testName = rxTest:testName()
	
	local flows = {}
	
	local function testBigSend()
	
		local observable = Observable.empty()
	
		for i = 1, 20 do
	
			local tempFlow = _rtmfpClient:openFlow( serverAddr, serverPort )
			
			flows[ i ] = tempFlow
			
			observable = observable:merge( tempFlow:send( testName .. tostring( i ), noEchoMetadata ) )
		
		end
		
		return observable
		
	end
		
	local function closeAll()
	
		local observable = Observable.empty()
	
		for i = 1, 20 do
	
			local tempFlow = flows[ i ]
			flows[ i ] = nil
			
			observable = observable:concat( tempFlow:shutdown() )
			
		end
		
		return observable
		
	end
		
	rxTest:expectOnlyCompleted( testBigSend() )

	rxTest:expectOnlyCompleted( closeAll() )

end

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

function WFRtmfpClientPerformanceTest.rxTestSend_20_singleflow_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~0.3 seconds

	local testName = rxTest:testName()

	local tempFlow = _rtmfpClient:openFlow( serverAddr, serverPort )

	local function testBigSend()
	
		local observable = Observable.empty()
	
		for i = 1, 20 do

			observable = observable:merge( tempFlow:send( testName .. tostring( i ), noEchoMetadata ) )

		end
		
		return observable
		
	end
			
	rxTest:expectOnlyCompleted( testBigSend() )

	rxTest:expectOnlyCompleted( tempFlow:shutdown() )

end

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

function WFRtmfpClientPerformanceTest.rxTestFlowPriorityMessageSeries_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- Average results on Adobe ethernet: ~2.45 seconds

	local testName = rxTest:testName()

	local _flowHigh = _rtmfpClient:openFlow( serverAddr, serverPort, "highest" )
	local _flowLow = _rtmfpClient:openFlow( serverAddr, serverPort, "low" )
	local highPriorityN = 0
	local lowPriorityN = 0
	local messagesN = 3
	
	local priorityStream = _rtmfpClient:listen()
	
		:map( function( rtmfpRequest )
		
			if rtmfpRequest:getMetadata().testName == testName then
		
				if string.find( rtmfpRequest:getMetadata().priority, "High" ) then
					debug.print( "listenE fired: Priority: HIGH!" )
			 		highPriorityN = highPriorityN + 1
				elseif string.find( rtmfpRequest:getMetadata().priority, "Low" ) then
					debug.print( "listenE fired: Priority: LOW!" )
					lowPriorityN = lowPriorityN + 1
				end
			
				if lowPriorityN == messagesN and highPriorityN < messagesN then
					assert( false, "Low priority messages received before High" )
				end
			
			end
			
			return highPriorityN
			
		end )
			
	local priorityChecker = rxTest:observe( priorityStream )

	-- Message Info
	local testData = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testData .. '/rtmfp/1MB.txt'
	local file = io.open( filePath, 'rb' )
	local fileContent = file:read("*all")
	file:close()
			
	local function testBigSend()
	
		local observable = Observable.empty()
	
		for i = 1, messagesN * 2 do
	
			local content = tostring( i ) .. fileContent
			local priority = false
			
			if( i % 2 == 0 ) then
				priority = "Low"
			else
				priority = "High"
			end

			-- Use metadata to keep track of the type of data we're sending
			
			local metadata = {
				priority = priority,
				testName = testName,
				dont_close_flow = true,
			}
			
			if priority == 'Low' then
				observable = observable:concat( _flowLow:send( content, metadata ) )
			else
				observable = observable:concat( _flowHigh:send( content, metadata ) )
			end
			
		end
	
		return observable
	
	end

	rxTest:expectOnlyCompleted( testBigSend() )

	priorityChecker:expectDataWithinTime( testWaitTime, messagesN )

	rxTest:expectOnlyCompleted( _flowLow:send( testName ) )

	rxTest:expectOnlyCompleted( _flowHigh:send( testName ) )

	rxTest:expectOnlyCompleted( _flowHigh:shutdown() )

	rxTest:expectOnlyCompleted( _flowLow:shutdown() )

end

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

function WFRtmfpClientPerformanceTest.rxTestFlowPrioritySingleMessage_testPerformance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local testName = rxTest:testName()

	-- Average results on Adobe ethernet: ~0.01 seconds

	local _flowHigh = _rtmfpClient:openFlow( serverAddr, serverPort, "highest" )
	local _flowLow = _rtmfpClient:openFlow( serverAddr, serverPort, "low" )
	
	-- Message Info
	local testDataPath = KSPathUtils.getStandardFilePath( 'testdata' )
	local filePath = testDataPath .. '/rtmfp/1MB.txt'
	local file = io.open( filePath, 'r' )
	local content = file:read("*all")
	file:close()
	
	local highPriorityN = 0
	local lowPriorityN = 0
	
	local highPriorityStream = _rtmfpClient:listen()
	
		:map( function( rtmfpRequest )
			
			if rtmfpRequest:getMetadata().testName == testName then
			
				if string.find( rtmfpRequest:getMetadata().priority, "High" ) then
					debug.print( "listenE fired: Priority: HIGH!" )
			 		highPriorityN = highPriorityN + 1
				elseif string.find( rtmfpRequest:getMetadata().priority, "Low" ) then
					debug.print( "listenE fired: Priority: LOW!" )
					lowPriorityN = lowPriorityN + 1
				end
			
				assert( lowPriorityN <= highPriorityN )
			
			end
			
			return highPriorityN
			
		end )
	
	local highPriWatcher = rxTest:observe( highPriorityStream )
	
	local lowPriorityStream = _rtmfpClient:listen()
	
		:map( function()
			
			return lowPriorityN
			
		end )
	
	local lowPriWatcher = rxTest:observe( lowPriorityStream )
	
	-- Use metadata to keep track of the type of data we're sending
	
	local metadataLow = {}
	metadataLow.priority = "Low"
	metadataLow.testName = testName
	
	local metadataHigh = {}
	metadataHigh.priority = "High"
	metadataHigh.testName = testName
	
	_flowHigh:send( content, metadataHigh )

		:merge( _flowLow:send( content, metadataLow ) )

		:call( runUntilDone )

	-- Make sure we receive the high priority message before the low priority message.
	
	highPriWatcher:expectDataWithinTime( testWaitTime, 1 )
	
	-- Make sure we eventually receive the Low priority message
	
	lowPriWatcher:expectDataWithinTime( testWaitTime, 1 )
	
	rxTest:expectOnlyCompleted( _flowHigh:shutdown() )

	rxTest:expectOnlyCompleted( _flowLow:shutdown() )

end

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

return {}
-- return WFRxTestCase.wrapSuite( WFRtmfpClientPerformanceTest )
