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

WFRxAsyncRetryableRelayTest.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 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'

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

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

local _dataValues = {}

_dataValues[ 1 ] = "one"

_dataValues[ 2 ] = "dos"

_dataValues[ 3 ] = "drei"

_dataValues[ 4 ] = 4

_dataValues[ 5 ] = "55/11"

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

local function updateSendCommandsLoggerStream(
				sendCommandsLoggerStream, dataValueSequence, 
				sendCommand, value )

	local sendCommandDataValueSeqCommands = 
			sendCommandsLoggerStream:getValue()

	local sendCommandsForDataValueSequence =
			sendCommandDataValueSeqCommands[ dataValueSequence ] or {}

	sendCommandsForDataValueSequence[ #sendCommandsForDataValueSequence + 1 ] =
		{ sendCommand, value }

	sendCommandDataValueSeqCommands[ dataValueSequence ] = 
		sendCommandsForDataValueSequence

	sendCommandsLoggerStream
		:onNext( sendCommandDataValueSeqCommands )
	
end

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

local function setRef( ref, v )

	ref[ 1 ] = v

end

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

local function derefRef( ref )

	return ref[ 1 ]

end

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

local function makeRelayDataResultStreamForTest( 
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				sendCommandsLoggerStream,
				disposedBeforeDataValueSequenceSentCountsStream,
				thisDataValueSequence, dataValue )

	local dataValueSequenceToSendCommandResultStreamStreamArray =
				derefRef( dataValueSequenceToSendCommandResultStreamStreamArrayRef  )

	local sendCommandResultStreamStreamArray =
				dataValueSequenceToSendCommandResultStreamStreamArray[ thisDataValueSequence ]

	if not sendCommandResultStreamStreamArray then

		return Rx.Observable.never()

	end

	local sendCommandResultStreamStreamArrayIndex

	local dataWasSent

	return Rx.Observable.create(
			function( outerObserver )

				local subscr = 
					Rx.Observable.create(
						function( observer )

							sendCommandResultStreamStreamArrayIndex = 
								( sendCommandResultStreamStreamArrayIndex or 0 ) + 1

							local sendCommandResultStreamStream =
									sendCommandResultStreamStreamArray[ sendCommandResultStreamStreamArrayIndex ]

							return sendCommandResultStreamStream
									:switch()
									:subscribe( observer )

						end )
					:tapOnError(
						function()

							updateSendCommandsLoggerStream(
								sendCommandsLoggerStream, thisDataValueSequence, 
								"error", "error" )

						end )
					:take( 1 ) -- sendCommandResultStream only emits when it means "success"
					:takeLast( 0 ) -- we don't care what was emitted
					:tapOnCompleted(
						function()

							dataWasSent = true

							updateSendCommandsLoggerStream(
								sendCommandsLoggerStream, thisDataValueSequence, 
								"completes", dataValue )

						end )
					:subscribe( outerObserver )

				return Rx.Disposable.create(

					function()

						if not dataWasSent then

							local disposedBeforeDataValueSequenceSentCounts =
								disposedBeforeDataValueSequenceSentCountsStream
									:getValue()

							disposedBeforeDataValueSequenceSentCounts[ thisDataValueSequence ] =
								( disposedBeforeDataValueSequenceSentCounts[ thisDataValueSequence ] or 0 ) + 1

							disposedBeforeDataValueSequenceSentCountsStream
								:onNext( disposedBeforeDataValueSequenceSentCounts )

						end

						subscr:dispose()

					end )

			end )

end

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

local function makeTestStreamsForDeliverStreamMakerFunc( 
				deliverStreamMakerFunc )

	local dataValueStream = Rx.Subject()

	local seq = 0

	local sequencedDataValueStream =
			dataValueStream
				:map(
					function( ... )

						seq = seq + 1

						return seq, ...

					end )
	
	local dataValueSequenceToSendCommandResultStreamStreamArrayRef = {}

	local sendCommandsLoggerStream = Rx.BehaviorSubject( {} )

	local disposedBeforeDataValueSequenceSentCountsStream = 
			Rx.BehaviorSubject( {} )

	local asyncRetryableRelayDeliverDataValuesStream =
		deliverStreamMakerFunc(
			sequencedDataValueStream,
			makeRelayDataResultStreamForTest, 
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				sendCommandsLoggerStream,
				disposedBeforeDataValueSequenceSentCountsStream )

	return asyncRetryableRelayDeliverDataValuesStream, dataValueStream,
			dataValueSequenceToSendCommandResultStreamStreamArrayRef,
			sendCommandsLoggerStream, disposedBeforeDataValueSequenceSentCountsStream

end

--------------------------------------------------------------------------------
-- given commandSequenceArray, this sets up Rx.Observables so that data and
-- send events happen in the same sequence as commandSequenceArray.
--
-- In effect this emulates async send results without requiring async
--------------------------------------------------------------------------------

local function makeRunTestsOnSubscribeStream( 
				rxTest, actualResultsSubscr, dataValueStream,
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				commandSequenceArray )

	rxTest
		:finally(
			function()

				actualResultsSubscr:dispose()

			end )

	local dataValueSequenceToSendCommandResultStreamStreamArray = {}

	local dataValueSequence = 0

	local dataValueSequenceToDataValue = {}

	local runTestSequenceArray = {}

	-- Do two things: 
	-- 	1) set up a stream of streams that allow us to emulate async send results
	-- 	2) set up a sequence of stream and values to deliver on said streams
	
	for _, commandData in ipairs( commandSequenceArray ) do

		local streamForCommand = commandData[ 1 ]

		if streamForCommand == "data" then

			local dataCommand = commandData[ 2 ]

			local dataValue = commandData[ 3 ]

			if dataCommand == "data" then

				dataValueSequence = dataValueSequence + 1

				dataValueSequenceToDataValue[ dataValueSequence ] = dataValue

			end

			runTestSequenceArray[ #runTestSequenceArray + 1 ] = 
				{ dataValueStream, dataCommand, commandData[ 3 ] }

		elseif streamForCommand == "send" then
			
			-- add the "outer stream" to array of "outer streams" for sendCommandDataValueSeq
			local sendCommandDataValueSequence = commandData[ 2 ]

			local sendCommandResultStreamStreamArray =
					dataValueSequenceToSendCommandResultStreamStreamArray[ sendCommandDataValueSequence ] or
					{}

			local outerStream = Rx.Subject()

			sendCommandResultStreamStreamArray[ #sendCommandResultStreamStreamArray + 1 ] = outerStream

			dataValueSequenceToSendCommandResultStreamStreamArray[ sendCommandDataValueSequence ] =
				sendCommandResultStreamStreamArray

			-- now set up the "inner stream" payload to deliver on the "outer stream"
			local sendCommandType = commandData[ 3 ]

			local dataValueToSend =
					dataValueSequenceToDataValue[ sendCommandDataValueSequence ]

			local innerStream = 
					( sendCommandType == "completes" and
						Rx.Observable.just( dataValueToSend ) ) or
					( sendCommandType == "error" and
						Rx.Observable.throw( 
							string.format( "throw instead of %s", dataValueToSend ) ) ) or
					( sendCommandType == "never" and
						Rx.Observable.never() ) or
					Rx.Observable.empty()

			runTestSequenceArray[ #runTestSequenceArray + 1 ] = 
				{ outerStream, "data", innerStream }

		end

	end

	-- pass dataValueSequenceToSendCommandResultStreamStreamArray to our ref
	setRef( dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				dataValueSequenceToSendCommandResultStreamStreamArray )

	-- now return an observable that runs the test by executing each
	-- runTestSequenceArray element in order
	return Rx.Observable.from( runTestSequenceArray )
			:map(
				function( runTestSequenceEntry )

					local stream, command, value = unpack( runTestSequenceEntry )

					if command == "data" then

						stream:onNext( value )

					elseif command == "completes" then

						stream:onCompleted( value )

					elseif command == "error" then

						stream:onError( "fale wail" )

					end

				end )
			:takeLast( 0 )

end

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

local function runExpectedResultsArrayTest( 
				rxTest, deliverStreamMakerFunc, commandSequenceArray,
				expectedDataValueSequenceSendCommands,
				maybeExpectedDisposedBeforeDataValueSequenceSentCounts )

	local asyncRetryableRelayDeliverDataValuesStream, dataValueStream,
			dataValueSequenceToSendCommandResultStreamStreamArrayRef,
			sendCommandsLoggerStream,
			disposedBeforeDataValueSequenceSentCountsStream =
				makeTestStreamsForDeliverStreamMakerFunc(
					deliverStreamMakerFunc )

	local actualResults = Rx.ReplaySubject( 1 )

	local actualResultsSubscr =
			asyncRetryableRelayDeliverDataValuesStream
				:subscribe( actualResults )

	local runTestOnSubscribeStream =
			makeRunTestsOnSubscribeStream( 
				rxTest, actualResultsSubscr, dataValueStream,
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				commandSequenceArray  )

	rxTest:log( "runExpectedResultsArrayTest: check downstream results" )

	rxTest
		:expectOnlyCompleted( runTestOnSubscribeStream )

	rxTest
		:expectOnlyCompleted( 
			actualResults )

	rxTest
		:expectSingleValue( 
			sendCommandsLoggerStream:take( 1 ),
			expectedDataValueSequenceSendCommands )

	local expectedDisposedBeforeDataValueSequenceSentCounts =
			maybeExpectedDisposedBeforeDataValueSequenceSentCounts or {}

	rxTest
		:expectSingleValue( 
			disposedBeforeDataValueSequenceSentCountsStream:take( 1 ),
			expectedDisposedBeforeDataValueSequenceSentCounts )

end

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

local function makeDataValueStreamErrorTest( deliverStreamMakerFunc )

	return function( rxTest )

		local asyncRetryableRelayDeliverDataValuesStream, dataValueStream,
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				sendCommandsLoggerStream, disposedBeforeDataValueSequenceSentCountsStream =
					makeTestStreamsForDeliverStreamMakerFunc( 
						deliverStreamMakerFunc )

		local commandSequenceArray = 
			{ 
				{ "data", "data", _dataValues[ 1 ] }, -- start send 1
				{ "send", 1, "completes" }, -- send 1, wait
				{ "data", "data", _dataValues[ 2 ] }, -- start send 2
				{ "data", "data", _dataValues[ 3 ] }, 
				{ "data", "error" }, -- raise error on data stream
				{ "send", 2, "completes" }, -- no-op, the data stream has errors
			}

		local actualResults = Rx.ReplaySubject( 1 )

		local actualResultsSubscr =
				asyncRetryableRelayDeliverDataValuesStream
					:toArray()
					:subscribe( actualResults )

		local runTestOnSubscribeStream =
				makeRunTestsOnSubscribeStream( 
					rxTest, actualResultsSubscr, dataValueStream,
					dataValueSequenceToSendCommandResultStreamStreamArrayRef,
					commandSequenceArray  )

		rxTest:log( "makeDataValueStreamErrorTest: make sure emits error" )

		rxTest
			:expectOnlyCompleted( runTestOnSubscribeStream )

		rxTest
			:expectOnlyError(
				actualResults,
				"fale wail" )

		local expectedDataValueSequenceSendCommands = 
			{
				[ 1 ] = { { "completes", _dataValues[ 1 ] } },
			}

		rxTest
			:expectSingleValue( 
				sendCommandsLoggerStream:take( 1 ),
				expectedDataValueSequenceSendCommands )

		local expectedDisposedBeforeDataValueSequenceSentCounts =
			{ 
				[ 2 ] = 1 -- dispose the send error
			}

		rxTest
			:expectSingleValue( 
				disposedBeforeDataValueSequenceSentCountsStream:take( 1 ),
				expectedDisposedBeforeDataValueSequenceSentCounts )

	end

end

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

local function makeEmptyDataValueStreamTest( deliverStreamMakerFunc )

	return function( rxTest )

		local asyncRetryableRelayDeliverDataValuesStream, dataValueStream,
				dataValueSequenceToSendCommandResultStreamStreamArrayRef,
				sendCommandsLoggerStream, disposedBeforeDataValueSequenceSentCountsStream =
					makeTestStreamsForDeliverStreamMakerFunc( 
						deliverStreamMakerFunc )

		local commandSequenceArray = 
			{ 
				{ "data", "completes" }, -- complete without emitting data
			}

		local actualResults = Rx.ReplaySubject( 1 )

		local actualResultsSubscr =
				asyncRetryableRelayDeliverDataValuesStream
					:subscribe( actualResults )

		local runTestOnSubscribeStream =
				makeRunTestsOnSubscribeStream( 
					rxTest, actualResultsSubscr, dataValueStream,
					dataValueSequenceToSendCommandResultStreamStreamArrayRef,
					commandSequenceArray  )

		rxTest:log( "makeEmptyDataValueStreamTest: make sure test run completes" )

		rxTest
			:expectOnlyCompleted( runTestOnSubscribeStream )

		rxTest:log( "makeEmptyDataValueStreamTest: make actual result complete" )

		rxTest
			:expectOnlyCompleted( 
				actualResults )

		rxTest
			:expectSingleValue( 
				sendCommandsLoggerStream:take( 1 ),
				{} )

		rxTest
			:expectSingleValue( 
				disposedBeforeDataValueSequenceSentCountsStream:take( 1 ),
				{} )

	end

end

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

local function makeInterleavedTest( deliverStreamMakerFunc )
			
	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] },
			{ "send", 1, "completes" },
			{ "data", "data", _dataValues[ 2 ] },
			{ "send", 2, "completes" },
			{ "data", "data", _dataValues[ 3 ] },
			{ "send", 3, "completes" },
			{ "data", "data", _dataValues[ 4 ] },
			{ "send", 4, "completes" },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" },
			{ "send", 5, "completes" },
		}

	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "completes", _dataValues[ 1 ] } },
			[ 2 ] = { { "completes", _dataValues[ 2 ] } },
			[ 3 ] = { { "completes", _dataValues[ 3 ] } },
			[ 4 ] = { { "completes", _dataValues[ 4 ] } },
			[ 5 ] = { { "completes", _dataValues[ 5 ] } },
		}

	return function( rxTest )

		runExpectedResultsArrayTest( 
			rxTest, deliverStreamMakerFunc, commandSequenceArray,
			expectedDataValueSequenceSendCommands )

	end

end

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

local function makeInterleavedSendErrorsTest( deliverStreamMakerFunc )
			
	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] },
			{ "send", 1, "error" },
			{ "send", 1, "completes" },
			{ "data", "data", _dataValues[ 2 ] },
			{ "send", 2, "error" },
			{ "send", 2, "error" },
			{ "send", 2, "completes" },
			{ "data", "data", _dataValues[ 3 ] },
			{ "send", 3, "error" },
			{ "send", 3, "error" },
			{ "send", 3, "error" },
			{ "send", 3, "completes" },
			{ "data", "data", _dataValues[ 4 ] },
			{ "send", 4, "error" },
			{ "send", 4, "error" },
			{ "send", 4, "error" },
			{ "send", 4, "error" },
			{ "send", 4, "completes" },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" },
			{ "send", 5, "completes" },
		}

	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "error", "error" }, { "completes", _dataValues[ 1 ] } },
			[ 2 ] = { { "error", "error" }, { "error", "error" }, { "completes", _dataValues[ 2 ] } },
			[ 3 ] = { { "error", "error" }, { "error", "error" }, { "error", "error" }, { "completes", _dataValues[ 3 ] } },
			[ 4 ] = { { "error", "error" }, { "error", "error" }, { "error", "error" }, { "error", "error" }, { "completes", _dataValues[ 4 ] } },
			[ 5 ] = { { "completes", _dataValues[ 5 ] } },
		}

	local expectedDisposedBeforeDataValueSequenceSentCounts = 
		{
			[ 1 ] = 1, -- dispose all send errors
			[ 2 ] = 2, -- dispose all send errors
			[ 3 ] = 3, -- dispose all send errors
			[ 4 ] = 4, -- dispose all send errors
		}

	return function( rxTest )

		runExpectedResultsArrayTest( 
			rxTest, deliverStreamMakerFunc, commandSequenceArray,
			expectedDataValueSequenceSendCommands,
			expectedDisposedBeforeDataValueSequenceSentCounts )

	end

end

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

local function rxTestDeliverAllDataValuesStreamSomeBacklogged( rxTest )

	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] },
			{ "send", 1, "completes" },
			{ "data", "data", _dataValues[ 2 ] },
			{ "data", "data", _dataValues[ 3 ] },
			{ "send", 2, "completes" },
			{ "send", 3, "completes" },
			{ "data", "data", _dataValues[ 4 ] },
			{ "send", 4, "completes" },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" },
			{ "send", 5, "completes" },
		}

	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "completes", _dataValues[ 1 ] } },
			[ 2 ] = { { "completes", _dataValues[ 2 ] } },
			[ 3 ] = { { "completes", _dataValues[ 3 ] } },
			[ 4 ] = { { "completes", _dataValues[ 4 ] } },
			[ 5 ] = { { "completes", _dataValues[ 5 ] } },
		}

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverAllDataValuesStream,
		commandSequenceArray, 
		expectedDataValueSequenceSendCommands )

end

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

local function rxTestDeliverAllDataValuesStreamAllBacklogged( rxTest )

	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] },
			{ "data", "data", _dataValues[ 2 ] },
			{ "data", "data", _dataValues[ 3 ] },
			{ "data", "data", _dataValues[ 4 ] },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" },
			{ "send", 1, "completes" },
			{ "send", 2, "completes" },
			{ "send", 3, "completes" },
			{ "send", 4, "completes" },
			{ "send", 5, "completes" },

		}

	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "completes", _dataValues[ 1 ] } },
			[ 2 ] = { { "completes", _dataValues[ 2 ] } },
			[ 3 ] = { { "completes", _dataValues[ 3 ] } },
			[ 4 ] = { { "completes", _dataValues[ 4 ] } },
			[ 5 ] = { { "completes", _dataValues[ 5 ] } },
		}

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverAllDataValuesStream,
		commandSequenceArray, 
		expectedDataValueSequenceSendCommands )

end

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

local function rxTestDeliverLatestDataValueStreamSomeBacklogged( rxTest )

	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] }, -- start send 1
			{ "data", "data", _dataValues[ 2 ] },
			{ "send", 1, "completes" }, -- sends 1 and begin send 2, the latest
			{ "send", 2, "completes" }, -- sends 2 and waits
			{ "data", "data", _dataValues[ 3 ] }, -- start send 3
			{ "data", "data", _dataValues[ 4 ] },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" }, -- stream must deliver 5 or forever block 
			{ "send", 3, "completes" }, -- sends 3 and begin send 5, the latest
			{ "send", 5, "completes" }, -- sends 5 and completes (data done)
		}


	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "completes", _dataValues[ 1 ] } },
			[ 2 ] = { { "completes", _dataValues[ 2 ] } },
			[ 3 ] = { { "completes", _dataValues[ 3 ] } },
			[ 5 ] = { { "completes", _dataValues[ 5 ] } },
		}

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverLatestDataValueStream,
		commandSequenceArray, 
		expectedDataValueSequenceSendCommands )

end

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

local function rxTestDeliverLatestDataValueStreamAllBacklogged( rxTest )

	local commandSequenceArray = 
		{ 
			{ "data", "data", _dataValues[ 1 ] }, -- start send 1
			{ "data", "data", _dataValues[ 2 ] },
			{ "data", "data", _dataValues[ 3 ] },
			{ "data", "data", _dataValues[ 4 ] },
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" }, -- stream must deliver 5 or forever block 
			{ "send", 1, "completes" }, -- sends 1 and begin send 5, the latest
			{ "send", 5, "completes" }, -- sends 5 and completes (data done)
		}

	local expectedDataValueSequenceSendCommands = 
		{
			[ 1 ] = { { "completes", _dataValues[ 1] } },
			[ 5 ] = { { "completes", _dataValues[ 5] } },
		}

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverLatestDataValueStream,
		commandSequenceArray, 
		expectedDataValueSequenceSendCommands )

end

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

local function rxTestDeliverLatestDataValueStreamErrorsWhenBacklogged( rxTest )

	-- dataStream completes before final send
	local commandSequenceArray1 = 
		{ 
			{ "data", "data", _dataValues[ 1 ] }, -- start send 1
			{ "send", 1, "error" }, -- raise an error on 1
			{ "send", 1, "never" }, -- start new send request on 1 that won't complete
			{ "data", "data", _dataValues[ 2 ] }, -- cancels send request 1 and starts send request 2
			{ "data", "data", _dataValues[ 3 ] }, 
			{ "send", 2, "error" }, -- cancels 2 and begin 3, the latest
			{ "send", 3, "error" }, -- wait for 3, the latest
			{ "send", 3, "completes" }, -- sends 3 and waits, since latest
			{ "data", "data", _dataValues[ 4 ] }, -- start send 4
			{ "data", "data", _dataValues[ 5 ] },
			{ "data", "completes" }, -- stream must deliver 5 or forever block 
			{ "send", 4, "completes" }, -- sends 4 and begin send 5, the latest
			{ "send", 5, "error" }, -- wait for 5, the last
			{ "send", 5, "completes" }, -- sends 5 and completes (data done)
		}

	-- dataStream completes after final send
	local commandSequenceArray2 = 
		{ 
			{ "data", "data", _dataValues[ 1 ] }, -- start send 1
			{ "send", 1, "error" }, -- raise an error on 1
			{ "send", 1, "never" }, -- start new send request on 1 that won't complete
			{ "data", "data", _dataValues[ 2 ] }, -- cancels send request 1 and starts send request 2
			{ "data", "data", _dataValues[ 3 ] },
			{ "send", 2, "error" }, -- cancels 2 and begin 3, the latest
			{ "send", 3, "error" }, -- wait for 3, the latest
			{ "send", 3, "completes" }, -- sends 3 and waits, since latest
			{ "data", "data", _dataValues[ 4 ] }, -- start send 4
			{ "data", "data", _dataValues[ 5 ] },
			{ "send", 4, "completes" }, -- sends 4 and begin send 5, the latest
			{ "send", 5, "error" }, -- wait for 5, the last
			{ "send", 5, "completes" }, -- sends 5 and completes (data done)
			{ "data", "completes" }, -- stream must deliver 5 or forever block 
		}

	for i, commandSequenceArray in ipairs( { commandSequenceArray1, commandSequenceArray2 } ) do

		rxTest:logf( "running commandSequenceArray%s", i )

		local expectedDataValueSequenceSendCommands = 
			{
				[ 1 ] = { { "error", "error" } },
				[ 2 ] = { { "error", "error" } },
				[ 3 ] = { { "error", "error" }, { "completes", _dataValues[ 3 ] } },
				[ 4 ] = { { "completes", _dataValues[ 4 ] } },
				[ 5 ] = { { "error", "error" }, { "completes", _dataValues[ 5 ] } },
			}

		local expectedDisposedBeforeDataValueSequenceSentCounts = 
			{
				[ 1 ] = 2, -- dispose send error and the never send
				[ 2 ] = 1, -- dispose send error
				[ 3 ] = 1, -- dispose send error
				[ 5 ] = 1, -- dispose send error
			}

		runExpectedResultsArrayTest( 
			rxTest, WFRxAsyncRetryableRelay.deliverLatestDataValueStream,
			commandSequenceArray, 
			expectedDataValueSequenceSendCommands,
			expectedDisposedBeforeDataValueSequenceSentCounts )

	end

end

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

local _extremeDataValueCount = 1000

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

local function rxTestDeliverLatestDataValueStreamExtremeInterleave( rxTest )
			
	local commandSequenceArray = {}

	local expectedDataValueSequenceSendCommands = {}

	local expectedDisposedBeforeDataValueSequenceSentCounts = {}

	for i = 1, _extremeDataValueCount do

		commandSequenceArray[ #commandSequenceArray + 1 ] =
			{ "data", "data", i }

		commandSequenceArray[ #commandSequenceArray + 1 ] =
			{ "send", i, "error" }

		commandSequenceArray[ #commandSequenceArray + 1 ] =
			{ "send", i, "completes" }

		expectedDataValueSequenceSendCommands[ #expectedDataValueSequenceSendCommands + 1 ] = 
			{ 
				{ "error", "error" },
			 	{ "completes", i }
			}

		expectedDisposedBeforeDataValueSequenceSentCounts[ #expectedDisposedBeforeDataValueSequenceSentCounts + 1 ] = 1

	end

	commandSequenceArray[ #commandSequenceArray + 1 ] =
		{ "data", "completes" }

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverLatestDataValueStream,
		commandSequenceArray, expectedDataValueSequenceSendCommands,
		expectedDisposedBeforeDataValueSequenceSentCounts )

end

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

local function rxTestDeliverLatestDataValueStreamExtremeBacklog( rxTest )
			
	local commandSequenceArray = {}

	local expectedDataValueSequenceSendCommands = {}

	local expectedDisposedBeforeDataValueSequenceSentCounts = {}

	expectedDisposedBeforeDataValueSequenceSentCounts[ 1 ] = 1

	expectedDataValueSequenceSendCommands[ 1 ] = 
		{ { "error", "error" } }

	expectedDataValueSequenceSendCommands[ _extremeDataValueCount ] =
		{ { "completes", _extremeDataValueCount } }

	for i = 1, _extremeDataValueCount do

		commandSequenceArray[ #commandSequenceArray + 1 ] =
			{ "data", "data", i }

		if i == _extremeDataValueCount then

			commandSequenceArray[ #commandSequenceArray + 1 ] =
				{ "send", 1, "error" }

			commandSequenceArray[ #commandSequenceArray + 1 ] =
				{ "send", _extremeDataValueCount, "completes" }

		end

	end

	commandSequenceArray[ #commandSequenceArray + 1 ] =
		{ "data", "completes" }

	runExpectedResultsArrayTest( 
		rxTest, WFRxAsyncRetryableRelay.deliverLatestDataValueStream,
		commandSequenceArray, expectedDataValueSequenceSendCommands,
		expectedDisposedBeforeDataValueSequenceSentCounts )

end

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

return require "WFRxTestCase".wrapSuite( 'WFRxAsyncRetryableRelayTest', {

	rxTestDeliverAllDataValuesStreamDataValueStreamError = 
		makeDataValueStreamErrorTest(
			WFRxAsyncRetryableRelay.deliverAllDataValuesStream ),

	rxTestDeliverAllDataValuesStreamEmptyDataValueStream = 
		makeEmptyDataValueStreamTest(
			WFRxAsyncRetryableRelay.deliverAllDataValuesStream ),

	rxTestDeliverAllDataValuesStreamInterleaved = 
		makeInterleavedTest( WFRxAsyncRetryableRelay.deliverAllDataValuesStream ),

	rxTestDeliverAllDataValuesStreamInterleavedSendErrors = 
		makeInterleavedSendErrorsTest( 
			WFRxAsyncRetryableRelay.deliverAllDataValuesStream ),

	rxTestDeliverAllDataValuesStreamSomeBacklogged = 
		rxTestDeliverAllDataValuesStreamSomeBacklogged,

	rxTestDeliverAllDataValuesStreamAllBacklogged = 
		rxTestDeliverAllDataValuesStreamAllBacklogged,

	rxTestDeliverLatestDataValueStreamDataValueStreamError = 
		makeDataValueStreamErrorTest(
			WFRxAsyncRetryableRelay.deliverLatestDataValueStream ),

	rxTestDeliverLatestDataValueStreamEmptyDataValueStream = 
		makeEmptyDataValueStreamTest(
			WFRxAsyncRetryableRelay.deliverLatestDataValueStream ),

	rxTestDeliverLatestDataValueStreamInterleaved = 
		makeInterleavedTest( WFRxAsyncRetryableRelay.deliverLatestDataValueStream ),

	rxTestDeliverLatestDataValueStreamInterleavedSendErrors = 
		makeInterleavedSendErrorsTest(
			WFRxAsyncRetryableRelay.deliverLatestDataValueStream ),

	rxTestDeliverLatestDataValueStreamSomeBacklogged = 
		rxTestDeliverLatestDataValueStreamSomeBacklogged,

	rxTestDeliverLatestDataValueStreamAllBacklogged = 
		rxTestDeliverLatestDataValueStreamAllBacklogged,

	rxTestDeliverLatestDataValueStreamErrorsWhenBacklogged =
		rxTestDeliverLatestDataValueStreamErrorsWhenBacklogged,

	rxTestDeliverLatestDataValueStreamExtremeInterleave =
		rxTestDeliverLatestDataValueStreamExtremeInterleave,

	rxTestDeliverLatestDataValueStreamExtremeBacklog =
		rxTestDeliverLatestDataValueStreamExtremeBacklog,

} )
