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

WFRxAsyncRetryableRelay.lua

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

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

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

local KSFunctionUtils = require "KSFunctionUtils"
local Rx = require 'Rx'

--==============================================================================
-- @mark deliverAllDataValuesStream
--==============================================================================

local function deliverAllDataValuesStreamImp( 
					dataStream, boundRelayDataResultStream )

	return dataStream
			:map( 
				function( ... )

					-- relay the data, retrying until complete
					return boundRelayDataResultStream( ... )
								:retry() 

				end )
			:flatMerge( 1 ) -- flatMerge + retry means we serialize data relay

end

--------------------------------------------------------------------------------
--[[!
	deliverAllDataValuesStream

	\param dataStream (Rx.Observable) 
			emits data that is to be relayed.
	\param makeRelayDataResultStream (function)
			a function that returns an Rx.Observable() of
			relayDataResultStream.  relayDataResultStream represents the
			results of relaying a value from the dataStream.
			relayDataResultStream errors will result resubscribing to the same
			relayDataResultStream. Subsequent dataStream values won't be sent
			until the current relayDataResultStream completes.
	\param ... 
			any additional arguments will be bound to front of
			makeRelayDataResultStream before it relays each dataValue 


	\return Rx.Observable 
			On subscribe, the returned Rx.Observable will serially relay 
			dataStream values throuh makeRelayDataResultStream, Each resulting
			relayDataResultStream is forwarded until it completes. On
			relayDataResultStream completion, the next relayDataResultStream
			(if any) gets a subscription. The stream terminates when dataStream
			completes and the last relayDataResultStream completes.

	Canonical example: send all Rx.Observable output to another process via
	RTMFP. When errors are seen, retry sending the same data value via RTMFP.
	Wait for RTMFP success before sending next data value. i.e. consume
	dataStream at the rate which it can be sent via RTMFP
]]--
	
local function deliverAllDataValuesStream( dataStream, 
											makeRelayDataResultStream, ... )

	local boundRelayDataResultStream = 
			KSFunctionUtils.bindAllToFront( makeRelayDataResultStream, ... )
			 
	return deliverAllDataValuesStreamImp( 
				dataStream, boundRelayDataResultStream )
			
end

--==============================================================================
-- @mark deliverLatestDataValueStream
--==============================================================================

local function haveValidCurrDataValueSendState( 
				currDataValueSendState, maybeLastSentSequence,
				latestDataValueSequence )

	if type( currDataValueSendState ) ~= "table" then

		return false

	end

	if not currDataValueSendState.sequence then

		-- if no sequence in the send state, it's not valid
		return false

	end

	if currDataValueSendState.sequence == maybeLastSentSequence then

		-- This happens because maybeLastSentSequence updates before 
		-- currDataValueSendState is cleared. In this case, the
		-- currDataValueSendState is no longer valid because we
		-- have already sent the sequence in currDataValueSendState.
		return false

	end

	if currDataValueSendState.didRetry and
			currDataValueSendState.sequence < latestDataValueSequence then

		-- the current send failed at least once and there is new data,
		-- so currDataValueSendState is no longer valid.
		return false

	end

	-- otherwise we are currently waiting for the results of the first attempt 
	-- to send a data value from the data stream. Depending on the data rate
	-- and how long sending takes, this may mean we are waiting for send
	-- results from stale data, but that is by design. 
	return true

end

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

local function shouldSendLatestDataValueOrComplete( currDataValueSendState,
											maybeLastSentSequence,
											dataStreamIsFinished,
											latestDataValueSequence )

	if haveValidCurrDataValueSendState( 
			currDataValueSendState, maybeLastSentSequence, 
			latestDataValueSequence ) then

		-- DO NOTHING: we are currently waiting for an async send operation to
		-- report success or failure.
		return false

	end

	if dataStreamIsFinished then

		if maybeLastSentSequence == latestDataValueSequence then

			-- CONTINUE: we've received all of the data and sent the last
			-- latestDataValueSequence, we should complete
			return true

		end

	end

	if maybeLastSentSequence == latestDataValueSequence then

		-- DO NOTHING: we're all caught up, nothing new to send, we'll
		-- start up again the next time the dataStream emits new data
		return false

	end

	-- CONTINUE: we have unsent data that needs be sent
	return true

end

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

local function deliverLatestDataValueStreamImp( 
					dataStream, boundRelayDataResultStream ) 

	return Rx.Observable.create(

		function( observer )

			local currDataValueSendStateStream = Rx.BehaviorSubject( {} )

			local maybeLastSentSequenceStream = Rx.BehaviorSubject()

			local dataStreamIsFinishedStream = Rx.BehaviorSubject()

			local sentLatestValueAndDataStreamIsFinishedEvent =
					Rx.ReplaySubject( 1 )

			local latestDataValSeq = 0

			local rawSequencedLatestDataValueStream = 
					dataStream
						:map(
							function( ... )
									
								latestDataValSeq = latestDataValSeq + 1

								return latestDataValSeq, ...

							end )

			local dataStreamIsFinishedEventStream =
					Rx.Observable.create(

						function( innerObserver )

							dataStreamIsFinishedStream:onNext( true )

							innerObserver:onCompleted()

						end )


			local sequencedLatestDataValueStream = 
					Rx.Observable.concat(
							rawSequencedLatestDataValueStream,
							dataStreamIsFinishedEventStream )
						:startWith( nil )

			-- set up the following policy: 
			--
			-- when data values come in, don't try to send unless no data is being sent
			-- or stale data is being resent after one or more failures. After data has
			-- been sent successfully, check to see if a newer latest data value needs
			-- to be sent.
			local sendLatestDataResultSubscr =
				Rx.Observable.combineLatest(
						currDataValueSendStateStream,
						maybeLastSentSequenceStream, 
						dataStreamIsFinishedStream,
						sequencedLatestDataValueStream )
					:filter( shouldSendLatestDataValueOrComplete )
					-- flatMapLatest will preempt any active previous attempt
					:flatMapLatest(
						function ( _, maybeLastSentSequence,
									dataStreamIsFinished, latestDataValueSequence, ... )

							if dataStreamIsFinished and
									maybeLastSentSequence == latestDataValueSequence then

								sentLatestValueAndDataStreamIsFinishedEvent
									:onNext( true )

								return Rx.Observable.empty()

							end

							-- We are sending new data, update the current send
							-- state.
							currDataValueSendStateStream	
								:onNext( {

									sequence = latestDataValueSequence,

									didRetry = false

								} )

							-- start a retry forever cycle on this data, noting if we
							-- fail to send
							local boundRelayDataResultStreamStateUpdaterStream =
									boundRelayDataResultStream( ... )
										:catch(
											function( ... )

												-- we're about to retry, update the current
												-- send state
												currDataValueSendStateStream	
													:onNext( {

														sequence = latestDataValueSequence,

														didRetry = true

													} )

												return Rx.Observable.throw( ... )

											end )
										:retry() 

							local updateLatestSequenceAndResetCurrStateStream =
									Rx.Observable.create(
										function( innerObserver )

											-- completion indicates boundRelayDataResultStream( ... )
											-- successfully sent the data, update these variables
											-- accordingly
											maybeLastSentSequenceStream
												:onNext( latestDataValueSequence )

											-- clear the send state
											currDataValueSendStateStream	
												:onNext( {} )

											innerObserver:onCompleted()

										end ) 

							return Rx.Observable.concat(
									boundRelayDataResultStreamStateUpdaterStream,
									updateLatestSequenceAndResetCurrStateStream )

						end )
					-- keep subscribing to the above until the dataStream finishes and
					-- we successfully deliver the final dataStream value
					:takeUntil( sentLatestValueAndDataStreamIsFinishedEvent )
					:subscribe( observer )

		-- tear down all disposables when subscription is disposed
		return Rx.CompositeDisposable(
				currDataValueSendStateStream, maybeLastSentSequenceStream,
				dataStreamIsFinishedStream,
				sentLatestValueAndDataStreamIsFinishedEvent,
				sendLatestDataResultSubscr )

	end )

end 

--------------------------------------------------------------------------------
--[[!
	
	deliverLatestDataValueStream

	\param dataStream (Rx.Observable) 
			emits data that is to be relayed.
	\param makeRelayDataResultStream (function)
			a function that returns an Rx.Observable() relayDataResultStream.
			relayDataResultStream represents the results of relaying a value
			from the dataStream. relayDataResultStream errors will result in a
			new subscription to a relayDataResultStream. Subsequent dataStream
			values won't be sent until the current relayDataResultStream
			raises error or completes.
	\param ... 
			any additional arguments will be bound to front of
			makeRelayDataResultStream before it relays each dataValue 


	\return Rx.Observable 
			On subscribe, the returned Rx.Observable will pipe dataStream
			values through makeRelayDataResultStream if there are no active send 
			requests or if the current one has failed at least once and new
			data has arrived.  The resulting relayDataResultStream is
			forwarded until it completes. On relayDataResultStream completion,
			if there is a newer latest dataStreamValue, a new
			relayDataResultStream is created for it.  The stream terminates
			when dataStream completes and the relayDataResultStream for the last
			dataStream value completes.

	Canonical example: send only the lastest Rx.Observable output to another
	process via RTMFP. When errors are seen, either retry or start a new
	RTMFP send with the latest data. Discard multiple datastream values
	that accumulate while waiting for the send. i.e. consume dataStream
	data, discarding intermediate data that is generated faster than we can
	send it via RTMFP.
]]--
--------------------------------------------------------------------------------

local function deliverLatestDataValueStream( dataStream, 
												makeRelayDataResultStream, ... )

	local boundRelayDataResultStream = 
			KSFunctionUtils.bindAllToFront( makeRelayDataResultStream, ... )

	return deliverLatestDataValueStreamImp( 
			dataStream, boundRelayDataResultStream )

end 

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

return {

	deliverAllDataValuesStream = deliverAllDataValuesStream,

	deliverLatestDataValueStream = deliverLatestDataValueStream,

}
