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

WFRtmfpFlow.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.

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

local WFRtmfpErrorCodes = require 'WFRtmfpErrorCodes'
	
	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local KSData = require 'KSData'
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFCloseSupport = require 'WFCloseSupport'

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


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

local logger =  WFLogger 'WFRtmfpFlow'

local check = assert( WFArguments.check )
KSData.registerArgumentChecker()

local runUntilDone = WFRxRunUntilDone

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

--[[!
	\class WFRtmfpFlow
	\brief This namespace provides basic RTMFP Flow services.
]]--

-- @mark WFRtmfpFlow namespace

local WFRtmfpFlow = {

	_wfRtmfpClient = false,
		-- WFRtmfpClient object
		
	_flowGuid = false,
		-- The unique identifier for this Flow
		
	_flowOpenStatus = false,
		-- An Rx ReplaySubject that contains the current status of the Flow
		
	_whenOpen = false,
		-- An Rx ReplaySubject that resolves once the flow is opened
	
	_statusSubscription = false,
		-- An Rx observer that updates _flowOpenStatus
		
	_responseStreams = {},
		-- key: (string) response Guid
		-- value: (Replay Subject) Observable stream associated with this Flow
	
	_responseStreamSubscriptions = {},
		-- key: (string) response Guid
		-- value: (subscription) subscriptions to the open response streams
	
	_isShutdown = false,
		-- (boolean) Set when shutdown

	WFCloseSupport.addCloseSupportMethods,
	
}

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

function WFRtmfpFlow:__init( wfRtmfpClient, flowGuid, whenOpen )
	
	check( 'WFRtmfpFlow', 'WFRtmfpClient', '+string', '?table' )
	
	self._wfRtmfpClient = wfRtmfpClient
	self._flowGuid = flowGuid
	self._whenOpen = whenOpen
	
	self._flowOpenStatus = ReplaySubject( 1 )
	
	self._statusSubscription = self._whenOpen:subscribe(
		nil,
		function( errStr )
			self._flowOpenStatus:onNext( false, string.format( "saw error when opening flow: %s", errStr ) )
		end,
		function()
			self._flowOpenStatus:onNext( true )
		end )
		
end

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

--[[!

	\brief Get the unique identifier for the Flow

	\return string Flow Guid
	
--]]

function WFRtmfpFlow:getGuid()

	check( 'WFRtmfpFlow' )
	return self._flowGuid

end

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

--[[!

	\brief Wait for the Flow to succesfully open to the RTMFP peer.

	\return <code>Observable</code> An observable that resolves once the Flow has opened.
	
--]]

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

--[[!

	\brief Get the status of the Flow.

	\return <code>Observable</code> An observable that contains the open/close status
	of the Flow.
	
--]]

function WFRtmfpFlow:flowOpenStatus()

	check( 'WFRtmfpFlow' )
	
	return self._flowOpenStatus

end

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

--[[!

	\brief An internal method used to noticy this Flow object that the underlying
	 native Flow had an exception and closed.

	\return <void>
	
--]]

function WFRtmfpFlow:_flowException()

	check( 'WFRtmfpFlow' )
	self._flowOpenStatus:onNext( false, "flow exception raised by native code" )

	-- All open event streams should error out.

	local keys = {}
	
	for k in pairs( self._responseStreams ) do

		keys[ #keys + 1 ] = k

	end

	for _, k in ipairs( keys ) do

		local stream = self._responseStreams[ k ]
		
		if stream then
			stream:onError( "RTMFP Endpoint is offline", WFRtmfpErrorCodes.ENDPOINT_OFFLINE )
		end	

	end

end

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

--[[!

	\brief Closes this RTMFP Flow and returns the response via a promise.

	\return <code>Observable</code> An Observable that completes when the Flow
	is closed
	
--]]

function WFRtmfpFlow:shutdown()

	check( 'WFRtmfpFlow' )
	
	if self._isShutdown then
		return Observable.empty()
	end
	
	self._isShutdown = true
	
	return self._flowOpenStatus
		:take( 1 )
	
		:catch( 

			function( errStr )

				return Observable.just( false, errStr ) 

			end )
	
		:flatMapLatest( function( isOpen, reason )

			local notOpenReason =
					isOpen and
						"flow has been shutdown" or
						string.format( "flow was shutdown after closing for reason: %s", reason )
		
			local flowOpenStatus = self._flowOpenStatus

			self._flowOpenStatus = Rx.Observable.just( false, notOpenReason )

			flowOpenStatus:onCompleted()
			
			local statusSubscription = self._statusSubscription
			self._statusSubscription = nil
			statusSubscription:dispose()
			
			-- Dispose of all the response stream subscriptions

			for key, sub in pairs( self._responseStreamSubscriptions ) do

				sub:dispose()
				self._responseStreamSubscriptions[ key ] = nil
				
			end

			-- All open response streams should close

			for key, stream in pairs( self._responseStreams ) do

				stream:onCompleted()
				stream:dispose()
				self._responseStreams[ key ] = nil
				self._wfRtmfpClient:removeResponseStream( key )
				
			end
			
			if isOpen then
				return self._wfRtmfpClient:closeSendFlow( self._flowGuid )
			else
				return Observable.empty()
			end
			
		end )
		
		:catch( Observable.empty() )
		
end

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

--[[!

	\brief Closes the object.
	
	\return void
	
--]]

function WFRtmfpFlow:close()

	check( 'WFRtmfpFlow' )
	
	self:shutdown():call( runUntilDone )
		
end

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

local function flowStatusIsOpen( self )

	return self._flowOpenStatus
			:take( 1 )
			:flatMapLatest(
				function( isOpen, reason )

					if not isOpen then

						return Rx.Observable.throw( 
								string.format( "WFRtmfpFlow was closed because %s", reason ),
								WFRtmfpErrorCodes.FLOW_IS_CLOSED )
						
					end

					return Rx.Observable.just( true )

				end )

end

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

--[[!

	\brief Modifies the underlying Session of an RTMFP Flow

	\param flowGuid (string) The ID of a Flow object that's part of the Session
	you want to modify
	\param sessionModifiers (table) A table of session modifies. Please see
	WFRtmfpClient:modifyFlowSession for more information.
	
	\return <code>Observable</code> An Observable that completes once
	the session is modified.
	
--]]


function WFRtmfpFlow:modifySession( sessionModifiers )

	check( 'WFRtmfpFlow', 'table' )
	
	return flowStatusIsOpen( self )
		
		:flatMapLatest( function()
		
			return self._wfRtmfpClient:modifyFlowSession( self._flowGuid, sessionModifiers )
			
		end )
		
end

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

--[[!

	\brief Starts a RTMFP request and resolves to a response from the other peer.

	\param data (\c KSData or string) The data to send with the request.
	\param metadata (optional, table) metadata to include with the request.

	\return <code>Observable</code> An Observable that completes with a response 
	from the RTMFP peer.
	
--]]

function WFRtmfpFlow:requestData( data, metadata )
			
	check( 'WFRtmfpFlow', 'KSData|string', '?table' )
	
	logger:debugf( "WFRtmfpFlow:requestData( data = %s )", data )

	return flowStatusIsOpen( self )
	
		:flatMapLatest( function()
			
			logger:info( "sending requestData via self._wfRtmfpClient:message()" )

			local rtmfpMessageType = 'requestData'
			return self._wfRtmfpClient
					:message( rtmfpMessageType, self._flowGuid, data, metadata )
			
		end )

end

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

--[[!

	\brief Opens an observable that fires with responses from the RTMFP Peer, based
	on this particular request.

	\param data (\c KSData or string) The data to send with the request.
	\param metadata (optional, table) metadata to include with the request.

	\return <code>Observable</code> An Observable that fires with responses from the RTMFP Peer.
	
--]]

function WFRtmfpFlow:requestStream( data, metadata )
			
	check( 'WFRtmfpFlow', 'KSData|string', '?table' )
	
	-- Use a custom response Guid
	local responseGuid = KSIdGenerator.uuidInLowercaseHex()

	logger:debugf( "WFRtmfpFlow:requestStream( data = %s, responseGuid = %s )", data, responseGuid )

	metadata = metadata and KSTableUtils.shallowcopy( metadata ) or {}
	metadata._responseGuid = responseGuid

	local needCleanup = true

	local function streamCleanup()

		if needCleanup then

			needCleanup = false

			logger:debugf( "Closing request stream for response guid %s", responseGuid )

			self._wfRtmfpClient:removeResponseStream( responseGuid )

			self._responseStreams[ responseGuid ] = nil

			local responseStreamSubscription = self._responseStreamSubscriptions[ responseGuid ]

			if responseStreamSubscription then
				responseStreamSubscription:dispose()
			end

			self._responseStreamSubscriptions[ responseGuid ] = nil

		end

	end
	
	local responseStream = 
			self._wfRtmfpClient
				:createResponseStreamForGuid( responseGuid )

	local responseStreamSubscription = responseStream:subscribe( nil, streamCleanup, streamCleanup )
	
	self._responseStreams[ responseGuid ] = responseStream
	
	self._responseStreamSubscriptions[ responseGuid ] = responseStreamSubscription
	
	-- Remove the event stream from the look up table once it is closed.
	
	return flowStatusIsOpen( self )

		:flatMapLatest( function()

			-- send the request

			self._wfRtmfpClient:message( 'requestStream', self._flowGuid, data, metadata )
					
				:catch( function( ... )

					-- If the request failed to send, close the stream.
					local stream = self._responseStreams[ responseGuid ]
					
					if stream then
						stream:onError( ... )
					end
					
					return Observable.throw( ... )
					
				end )
					
				:call( runUntilDone )

			-- return a stream of responses
		
			return responseStream
			
		end )

		:finally( streamCleanup )

end

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

--[[!

	\brief Sends a RTMFP message and resolves when the message is sent.

	\param data (\c KSData or string) The data to send.
	\param metadata (optional, table) metadata to include with the request.
	
	\return <code>Observable></code> An Observable that completes when the
	data is sent.
	
--]]

function WFRtmfpFlow:send( data, metadata )
										
	check( 'WFRtmfpFlow', 'KSData|string', '?table' )
	
	logger:debugf( "WFRtmfpFlow:send( data = %s )", data )
	
	return flowStatusIsOpen( self )
		
		:flatMapLatest( function()
			
			local rtmfpMessageType = 'sendData'
			return self._wfRtmfpClient:message( rtmfpMessageType, self._flowGuid, data, metadata )
			
		end )

end

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

WFRtmfpFlow = Rx.Class( 'WFRtmfpFlow', WFRtmfpFlow )

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

return WFRtmfpFlow
