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

WFRtmfpClientTest.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 WFRtmfpErrorCodes = require 'WFRtmfpErrorCodes'

local KSData = require 'KSData'
local KSTestUtils = require 'KSTestUtils'
local WFRxTestCase = require 'WFRxTestCase'

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


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

local WFRtmfpClientTest = {}

local testWaitTime = 60
local dataTransitTime = 15 -- Time to wait for a response from the RTMFP Peer

-- RTMFP Echo Server. Lives on kibler.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 = "kibler.corp.adobe.com"
local serverPort = 1979
local rtmfpClient = false

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

-- Test Data

local KSPathUtils = require 'KSPathUtils'
local testDataPath = KSPathUtils.getStandardFilePath( 'testdata' )

local filePath = testDataPath .. '/rtmfp/1MB.txt'
local file = io.open( filePath, 'rb' )
local oneMbTestData = file:read("*all")
file:close()

filePath = testDataPath .. '/rtmfp/10Kb.txt'
file = io.open( filePath, 'rb' )
local tenKbTestData = file:read("*all")
file:close()

local DONT_CLOSE_FLOW_METADATA = {
	dont_close_flow = true
}

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

function WFRtmfpClientTest.setUp()

	if not rtmfpClient then
		rtmfpClient = WFRtmfpClient()
	end

end

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

function WFRtmfpClientTest.tearDownClass()

	if rtmfpClient then
		rtmfpClient:close()
	end

end

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

-- The underscore in the name is used to make sure that this test test runs
-- after all of the other tests. I could use tear down class method for this.

function WFRtmfpClientTest.rxTest_CloseInstance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	rxTest:expectOnlyCompleted( rtmfpClient:shutdown() )

	rtmfpClient._instance = nil

	rxTest:expectOnlyCompleted( rtmfpClient:shutdown() )
	
	rtmfpClient = false

end

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

function WFRtmfpClientTest.rxTest__ClosedInstanceCommands( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	local flowOpen = rxTest:observe( flow:flowOpenStatus():take( 1 ) )
	
	flowOpen:expectData( true )
	
	rxTest:expectOnlyCompleted( rtmfpClient:shutdown() )
	
	-- Try to open a new flow to the server
	
	local newFlow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	rxTest
		:expectSingleValue( 
			newFlow
				:flowOpenStatus()
				:take( 1 )
				:map(
					function( isOpen )

						return isOpen

					end ), 
			false )
	
	-- Try to send data on a _flow
	
	rxTest:expectOnlyError( flow:send( "This should fail!" ), ':ignore:', WFRtmfpErrorCodes.FLOW_IS_CLOSED )

	-- Try to close the flow. Shouldn't be a problem since it's a no-op of the istance is closed.
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

	-- Try to close a closed instance
	
	rxTest:expectOnlyCompleted( rtmfpClient:shutdown() )

	rtmfpClient = false

end

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

function WFRtmfpClientTest.rxTest___GCOpenInstance( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	rxTest:expectNotEqual( rtmfpClient, false )
	
	-- Exit without properly closing the instance.
	-- This will force garbage collection to close the instance.

end

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

function WFRtmfpClientTest.rxTestClientId( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local clientId = rtmfpClient:getId()
	rxTest:expectNotEqual( clientId, nil )
	
end

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

function WFRtmfpClientTest.rxTestClientPort( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local port = rtmfpClient:getPort()
	rxTest:expectNotEqual( port, nil )
	
end

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

function WFRtmfpClientTest.rxTestCreateTwoClientsOnSameLuaUniverse( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	KSTestUtils.testFails( function()
		return WFRtmfpClient()
	end )

end

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

function WFRtmfpClientTest.rxTestCloseInvalidFlow( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local requestGuid = "foo1"
	local flowGuid = "FAKE_FLOW"
	
	KSTestUtils.testFails( function()
		return rtmfpClient._instance:closeFlow( requestGuid, flowGuid )
	end )

end

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

function WFRtmfpClientTest.rxTestRegisterMultipleCompletionHandlers( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local completionHandler = function() return 'bar' end
	
	KSTestUtils.testFails( function()
		return rtmfpClient._instance:registerCompletionHandler( completionHandler )
	end )

end

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

function WFRtmfpClientTest.rxTestSendOnInvalidFlow( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local requestGuid = "foo1"
	local flowGuid = "FAKE_FLOW"
	local jsonData = "foo3"
	local data = KSData.make( "data" )
	local waitForResponse = false
	
	KSTestUtils.testFails( function()
		return rtmfpClient._instance:request( requestGuid, flowGuid, jsonData, data, waitForResponse )
	end )
	
end

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

local function testIncomingRequestErrorHandling( rxTest, testData )

	rxTest:setTimeLimit( testWaitTime )

	-- testData is a key used by the echo server to hand back an appropriate non-echo response
	
	local specialRequestData = KSData.make( testData )
	local normalRequestData = KSData.make( rxTest:testName() )
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	local listenStream = rtmfpClient:listen()
	
	local listener = rxTest:observe( listenStream
		
		:map( function( rtmfpRequest )
		
			local recvData = rtmfpRequest:getData()
			
			-- Make sure we don't get a response for our special error request.
			rxTest:expectNotEqual( recvData, specialRequestData )

			return tostring( rtmfpRequest:getData() )
			
		end ) )
		
	-- Send off the special request
			
	rxTest:expectOnlyCompleted( flow:send( specialRequestData, DONT_CLOSE_FLOW_METADATA ) )
			
	rxTest:expectOnlyCompleted( flow:send( normalRequestData ) )

	listener:expectDataWithinTime( dataTransitTime, tostring( normalRequestData ) )

	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

function WFRtmfpClientTest.rxTestIncomingRequestError( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	testIncomingRequestErrorHandling( rxTest, "INCOMING_REQUEST_ERROR" )
	
end

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

function WFRtmfpClientTest.rxTestIncomingRequestNoRequestId( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	testIncomingRequestErrorHandling( rxTest, "INCOMING_REQUEST_NO_REQUEST_ID" )
		
end

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

function WFRtmfpClientTest.rxTestIncomingRequestBadJson( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	testIncomingRequestErrorHandling( rxTest, "INCOMING_REQUEST_BAD_JSON" )
		
end

--------------------------------------------------------------------------------
--]]
function WFRtmfpClientTest.rxTestSimpleOpenAndCloseFlow( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	local flowOpen = rxTest:observe( flow:flowOpenStatus():take( 1 ) )
	flowOpen:expectData( true )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestOpenFlowToFakeAddress( rxTest )
	
	rxTest:setTimeLimit( testWaitTime )

	local nonRtmfpPeer = "jeffsfakeserver.corp.adobe.com" -- NOT A REAL SERVER

	local flow = rtmfpClient:openFlow( nonRtmfpPeer, serverPort )
	
	rxTest:expectSingleValue( 
		flow
			:flowOpenStatus()
			:take( 1 )
			:map(
				function( isOpen )

					return isOpen

				end ), 
		false )

end

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

function WFRtmfpClientTest.rxTestOpenMultipleFlows( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow1 = rtmfpClient:openFlow( serverAddr, serverPort, "lowest" )
	local flow2 = rtmfpClient:openFlow( serverAddr, serverPort, "low" )
	local flow3 = rtmfpClient:openFlow( serverAddr, serverPort, "normal" )
	local flow4 = rtmfpClient:openFlow( serverAddr, serverPort, "high" )
	local flow5 = rtmfpClient:openFlow( serverAddr, serverPort, "highest" )
	
	local flowOpen1 = rxTest:observe( flow1:flowOpenStatus():take( 1 ) )
	flowOpen1:expectData( true )
	
	local flowOpen2 = rxTest:observe( flow2:flowOpenStatus():take( 1 ) )
	flowOpen2:expectData( true )
	rxTest:expectNotEqual( flow1:getGuid(), flow2:getGuid() )
	
	local flowOpen3 = rxTest:observe( flow3:flowOpenStatus():take( 1 ) )
	flowOpen3:expectData( true )
	rxTest:expectNotEqual( flow1:getGuid(), flow3:getGuid() )
	rxTest:expectNotEqual( flow2:getGuid(), flow3:getGuid() )
			
	local flowOpen4 = rxTest:observe( flow4:flowOpenStatus():take( 1 ) )
	flowOpen4:expectData( true )
	rxTest:expectNotEqual( flow1:getGuid(), flow4:getGuid() )
	rxTest:expectNotEqual( flow2:getGuid(), flow4:getGuid() )
	rxTest:expectNotEqual( flow3:getGuid(), flow4:getGuid() )
			
	local flowOpen5 = rxTest:observe( flow5:flowOpenStatus():take( 1 ) )
	flowOpen5:expectData( true )
	rxTest:expectNotEqual( flow1:getGuid(), flow5:getGuid() )
	rxTest:expectNotEqual( flow2:getGuid(), flow5:getGuid() )
	rxTest:expectNotEqual( flow3:getGuid(), flow5:getGuid() )
	rxTest:expectNotEqual( flow4:getGuid(), flow5:getGuid() )
			
	-- Close the flows
	
	rxTest:expectOnlyCompleted( flow1:shutdown() )
	rxTest:expectOnlyCompleted( flow2:shutdown() )
	rxTest:expectOnlyCompleted( flow3:shutdown() )
	rxTest:expectOnlyCompleted( flow4:shutdown() )
	rxTest:expectOnlyCompleted( flow5:shutdown() )
		
end

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

function WFRtmfpClientTest.rxTestSimpleSend( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	rxTest:expectOnlyCompleted( flow:send( rxTest:testName() ) )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestOpenFlowBufferSize( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flowPri = "normal"
	local bufferSize = 127 * 1024 -- bytes
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort, flowPri, bufferSize )
	
	rxTest:expectOnlyCompleted( flow:send( rxTest:testName() ) )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestSendOnFullWriteBuffer( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flowPri = "normal"
	
	local bufferSize = 5 * 1024 -- 5kb
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort, flowPri, bufferSize )
	
	local testName = rxTest:testName()
	
	local listenStream = rtmfpClient:listen()
	
	local listener = rxTest:observe( listenStream
		
		:map( function( rtmfpRequest )
			return tostring( rtmfpRequest:getData() )
		end ) )
	
	-- Special metadata prevents the echo server from echoing.
	
	local metadata = {}
	metadata.no_echo = true

	-- Sending 20kb on a 5kb buffer will fill up the write buffer
	
	rxTest:expectOnlyCompleted( flow:send( tenKbTestData .. tenKbTestData, metadata ) )
			
	-- Doing it again should cause this message to wait until the write
	-- buffer is empty to send.
	
	rxTest:expectOnlyCompleted( flow:send( testName ) )

	listener:expectDataWithinTime( dataTransitTime, testName )

	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestSendWithTooBigOfMetadata( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local metadata = {}
	metadata.blob = oneMbTestData

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	rxTest:expectOnlyError( flow:send( rxTest:testName(), metadata ), ':ignore:', WFRtmfpErrorCodes.JSON_PAYLOAD_IS_TOO_LARGE )

	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestTryToSendBadData( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local requestGuid = "foo1"
	local flowGuid = "foo2"
	local jsonData = "foo3"
	local data = "I AM NOT KS DATA"
	local waitForResponse = false
	
	KSTestUtils.testFails( function()
		return rtmfpClient._instance:request(
			requestGuid, flowGuid, jsonData, data, waitForResponse )
	end )
	
end

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

function WFRtmfpClientTest.rxTestSimpleRequestData( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	local testName = rxTest:testName()
	
	local requestedData = flow:requestData( testName )
		:map( function( data )
			return tostring( data )
		end )
	
	rxTest:expectSingleValue( requestedData, testName )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

end


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

function WFRtmfpClientTest.rxTestSimpleListen( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- The server we are communicating with is an echo server. Therefore, request
	-- and send will result in this RTMFP instance receiving the same data it sends.
	-- We can use this to our advantage to test listen.

	local listenStream = rtmfpClient:listen()
	
	local listener = rxTest:observe( listenStream
		
		:map( function( rtmfpRequest )
			return tostring( rtmfpRequest:getData() )
		end ) )
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	local testName = rxTest:testName()

	rxTest:expectOnlyCompleted( flow:send( testName) )

	listener:expectDataWithinTime( dataTransitTime, testName ) 
		
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestRequestStream( rxTest )

	rxTest:setTimeLimit( testWaitTime )	

	local flow = rtmfpClient:openFlow( serverAddr, serverPort, "highest" )
	local _rtmfpRequest = false
	
	local responseCount = 0
	local clientCount = 0
	
	local testName = rxTest:testName()
	
	local subscription = rtmfpClient:listen():subscribe( 
		function( rtmfpRequest )
			local requestData = tostring( rtmfpRequest:getData() )
			if requestData == testName then
				clientCount = clientCount + 1
			end
			return clientCount
		end, 
		nil, 
		nil )
						
	local responseStream = flow:requestStream( testName )
						
	local responseListener = rxTest:observe( responseStream
		
		:map( function( rtmfpRequest )
			_rtmfpRequest = rtmfpRequest
			return tostring( rtmfpRequest:getData() )
		end ) )
						
	responseListener:expectDataWithinTime( dataTransitTime, testName )
	
	local responseCounterListener = rxTest:observe( responseStream
		
		:map( function( rtmfpRequest )
			
				local message = tostring( rtmfpRequest:getData() )
			
				if message == testName then
					responseCount = responseCount + 1
				end
				
				return responseCount
				
			end ) )
	
	-- Now respond multiple times to this request. Because the underlying request
	-- is a requestStream, the echo'd response should cause the response stream to fire.
	
	local metadata = {}
	metadata._responseGuid = _rtmfpRequest._responseGuid
	metadata.dont_close_flow = true
	
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, testName, metadata ) )
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, testName, metadata ) )
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, testName, metadata ) )
	
	responseCounterListener:expectDataWithinTime( dataTransitTime, 3 )
		
	rxTest:expectEqual( clientCount, 0 )
		
	subscription:dispose()
		
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestRequestStreamFailed( rxTest )
	
	rxTest:setTimeLimit( testWaitTime )
	
	local metadata = {}
	metadata.blob = oneMbTestData

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	-- The event stream should close when it realizes
	-- that the requestStream message never was delivered to the Peer.

	rxTest:expectOnlyError( flow:requestStream( 'this should fail', metadata ), ':ignore:', ':ignore:'  )

	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestGetMetadata( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local metadata = {}
	metadata.type = "this"
	metadata.to = "is"
	metadata.from = "metadata"
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	local clientStream = rtmfpClient:listen()

	local clientListener = rxTest:observe( clientStream
		
		:map( function( rtmfpRequest )
			return rtmfpRequest:getMetadata()
		end ) )

	rxTest:expectOnlyCompleted( flow:send( rxTest:testName(), metadata ) )

	clientListener:expectDataWithinTime( dataTransitTime, metadata )
		
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestRequestRespond( rxTest )
	
	rxTest:setTimeLimit( testWaitTime )
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	local _rtmfpRequest = false
	
	local clientStream = rtmfpClient:listen()
	
	local testName = rxTest:testName()
	
	local clientListener = rxTest:observe( clientStream

		:map( function( rtmfpRequest )
		
			local data = tostring( rtmfpRequest:getData() )
			
			if data == testName then
				_rtmfpRequest = rtmfpRequest
			end
			
			return data
			
		end ) )
	
	rxTest:expectOnlyCompleted( flow:send( testName, DONT_CLOSE_FLOW_METADATA ) )
	
	clientListener:expectDataWithinTime( dataTransitTime, testName )
	
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, "Here's your response!" ) )

	clientListener:expectDataWithinTime( dataTransitTime, "Here's your response!" )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

function WFRtmfpClientTest.rxTestRequestExpectingResponse( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	local _rtmfpRequest = false
	
	local clientStream = rtmfpClient:listen()
	
	local testName = rxTest:testName()
	
	local clientListener = rxTest:observe( clientStream
		
		:map( function( rtmfpRequest )
		
			local data = tostring( rtmfpRequest:getData() )
	
			if data == rxTest:testName() then
	
				_rtmfpRequest = rtmfpRequest
				_rtmfpRequest._expectingResponse = 'one'

			end

			return data
			
		end ) )
	
	rxTest:expectOnlyCompleted( flow:send( testName, DONT_CLOSE_FLOW_METADATA ) )

	clientListener:expectDataWithinTime( dataTransitTime, testName )
	
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, "Here's your response!" ) )

	clientListener:expectDataWithinTime( dataTransitTime, "Here's your response!" )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end


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

function WFRtmfpClientTest.rxTestRequestWithKSData( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	
	local testData = KSData.make( rxTest:testName() )
	
	rxTest:expectSingleValue( flow:requestData( testData )
		:map( function( value )
			return tostring( value )
		end ), tostring( testData ) )

	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

function WFRtmfpClientTest.rxTestLargeMessageQueue( rxTest )
	
	rxTest:setTimeLimit( testWaitTime )
	
	local flow = rtmfpClient:openFlow( serverAddr, serverPort, "highest" )
	
	local testName = rxTest:testName()
	
	local clientCount = 0
	
	local clientStream = rtmfpClient:listen()
		
	local clientCounter = rxTest:observe( clientStream
		
		:map( function( rtmfpRequest )
			
			local metadata = rtmfpRequest:getMetadata()
			
			if metadata.nameOfTest and metadata.nameOfTest == testName then
				clientCount = clientCount + 1
			end
			
			return clientCount
							
		end ) )
			
	local metadata = {
		nameOfTest = testName,
		dont_close_flow = true
	}
	
	for i = 1, 100 do
		
		rxTest:expectOnlyCompleted( flow:send( tostring( i ), metadata ) )
		
	end

	clientCounter:expectDataWithinTime( dataTransitTime * 2, 100 )

	rxTest:expectOnlyCompleted( flow:send( testName ) )
		
	rxTest:expectOnlyCompleted( flow:shutdown() )
		
end

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

function WFRtmfpClientTest.rxTestEndRequestStream( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local testName = rxTest:testName()

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )
	local _rtmfpRequest = false

	-- set up a flow that sends a request to the server and sends server
	-- responses on a response stream. This allows the server to send data to
	-- the client as long as it keeps its flow alive. 
	-- 
	-- However, the echo server will not keep non partner session flows open
	-- unless explicitly told dont_close_flow in the metadata, so we pass that
	-- on this request
	local responseStream = flow:requestStream( testName, DONT_CLOSE_FLOW_METADATA )

	local responseStreamObserver = rxTest:observe( responseStream )

	local responseListener = rxTest:observe( responseStream
		
		:map( function( rtmfpRequest )
			
			local data = tostring( rtmfpRequest:getData() )
			
			if data == testName then
				_rtmfpRequest = rtmfpRequest
			end
			
			return data
							
		end ) )

	responseListener:expectDataWithinTime( dataTransitTime, testName )

	-- Send off a 'SEND_DONE' request. We should receive 'done' back, forcing
	-- our event stream to close. Leave dont_close_flow blank here so that the
	-- server closes the flow after sending 'done'
			
	local metadata = {
		_responseGuid = _rtmfpRequest._responseGuid
	}
	
	rxTest:expectOnlyCompleted( _rtmfpRequest:respond( flow, 'SEND_DONE', metadata ) )

	responseStreamObserver:expectCompletedWithinTime( dataTransitTime )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )

end

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

function WFRtmfpClientTest.rxTestFlowDies_testNightly( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	-- testData is a key used by the echo server to hand back an appropriate non-echo response
	
	local specialRequestData = KSData.make( 'KILL_FLOW' )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	local flowStatusStream = 
			rxTest:observe( 
				flow:flowOpenStatus():map( function( isOpen ) return isOpen end ) )

	flowStatusStream:expectDataWithinTime( dataTransitTime, true )

	rxTest:expectOnlyCompleted( flow:send( specialRequestData, DONT_CLOSE_FLOW_METADATA ):catch( Observable.empty() ) )

	flowStatusStream:expectDataWithinTime( dataTransitTime, false )
	
	rxTest:expectOnlyError( flow:send( rxTest:testName() ), ':ignore:', WFRtmfpErrorCodes.FLOW_IS_CLOSED )	

	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

function WFRtmfpClientTest.rxTestSendCleanupWhenFlowDies( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local specialRequestData = KSData.make( 'KILL_FLOW' )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	local flowStatusStream = 
			rxTest:observe( 
				flow
					:flowOpenStatus()
					:map( function( isOpen ) return isOpen end ) )

	flowStatusStream:expectDataWithinTime( dataTransitTime, true )

	rxTest:expectOnlyCompleted( flow:send( specialRequestData, DONT_CLOSE_FLOW_METADATA ):catch( Observable.empty() ) )

	flowStatusStream:expectDataWithinTime( dataTransitTime, false )
	
	rxTest:expectOnlyError( flow:send( oneMbTestData ), ':ignore:', WFRtmfpErrorCodes.FLOW_IS_CLOSED )

	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

function WFRtmfpClientTest.rxTestRequestDataCleanupWhenFlowDies( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local specialRequestData = KSData.make( 'KILL_FLOW' )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	-- We should resolve the request with an error once we realize
	-- that the far end point has gone offline

	rxTest:expectOnlyError( flow:requestData( specialRequestData ), ':ignore:', WFRtmfpErrorCodes.ENDPOINT_OFFLINE )

	-- Closing the now-closed flow shouldn't harm anything

	rxTest:expectOnlyCompleted( flow:shutdown() )
	
end

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

local function nop() end

function WFRtmfpClientTest.rxTestRequestStreamCleanupWhenFlowDies( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local specialRequestData = KSData.make( 'KILL_FLOW' )

	local flow = rtmfpClient:openFlow( serverAddr, serverPort )

	local requestStream = flow:requestStream( specialRequestData )

	local sub = requestStream:subscribe( nil, nop, nop )

	rxTest:expectOnlyError( requestStream, ':ignore:', WFRtmfpErrorCodes.ENDPOINT_OFFLINE )
	
	rxTest:expectOnlyCompleted( flow:shutdown() )
	
	sub:dispose()
	
end

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

function WFRtmfpClientTest.rxTestWeakTables( rxTest )

	rxTest:setTimeLimit( testWaitTime )

	local flowGuid
	local openSendFlows = {}
	setmetatable( openSendFlows, { __mode = 'kv' } )

	do
		local flow = rtmfpClient:openFlow( serverAddr, serverPort )
		flowGuid = flow:getGuid()
		openSendFlows[ flowGuid ] = flow
	end

	rxTest:sleep( 1 )

	collectgarbage()
	collectgarbage()
	collectgarbage()
	
	rxTest:expectEqual( #openSendFlows, 0 )
	rxTest:expectEqual( openSendFlows[ flowGuid ], nil )
	
end

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

return WFRxTestCase.wrapSuite( WFRtmfpClientTest )
