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

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

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

	-- Rx
local Rx = require 'Rx'

	-- WF.core
local WFConfiguration = require 'WFConfiguration'

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

	-- WF.core.lua
local KSData = require 'KSData'

	-- WF.core.rxlua
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'

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

	-- WF.rtmfp.imp
local WFRtmfpErrorCodes = require 'WFRtmfpErrorCodes'

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'
local KSJsonEncoder = require 'KSJsonEncoder'

local Observable = Rx.Observable
local ReplaySubject = Rx.ReplaySubject

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

local logger =  WFLogger 'WFRtmfpPartnerSession'

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

local DISABLE_SESSION_TIMEOUT = WFConfiguration.RTMFP.DisableSessionTimeout
local BIG_TIMEOUT = 9999999
local runUntilDone = WFRxRunUntilDone

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

-- FIXME [jandrew 1/22/2015]: This should be a real secret.
local _theSecret = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'

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

-- Handshake Requests/Responses

local _hiCommand = 'HI'

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

local _hiMessage = KSJsonEncoder.encodeOzUploadJson { command = _hiCommand, secret = _theSecret }

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

local function partnerSessionCommandMetadata()

	return KSTableUtils.shallowcopy( 
	{ 

			_sessionPartnerCommand = true, 
			
			no_echo = true 
		
	} )
	

end

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

local _goodbyeCommand = 'GOODBYE'

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

local _goodbyeMessage = KSJsonEncoder.encodeOzUploadJson { command = _goodbyeCommand, secret = _theSecret } 

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

local _pingCommand = 'PING'

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

local _pingMessage = KSJsonEncoder.encodeOzUploadJson { command = _pingCommand } 

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

local _helloCommand = 'HELLO'

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

local _howdyCommand = 'HOWDY'

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

local _oneSecondInMillsec = 1000

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

local _oneMinuteInMilliseconds = 60 * _oneSecondInMillsec

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

local _oneHourInMilliseconds = 60 * _oneMinuteInMilliseconds

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

local _partnerSessionModifiers = {

	maxRTXTime = _oneSecondInMillsec * 5,

	maxIdleTime = _oneMinuteInMilliseconds,

	nearCloseTime = _oneSecondInMillsec * 5,

	sessionKeepaliveTime = 
		DISABLE_SESSION_TIMEOUT and 
			BIG_TIMEOUT or _oneHourInMilliseconds,
	
}

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

--[[!
	\class WFRtmfpPartnerSession
	\brief This namespace provides a layer on top of WF RTMFP for communicating
	with another WF RTMFP node. WFRtmfpPartnerSession objects have a one-to-one
	relationship with other WFRtmfpPartnerSession objects.
]]--

-- @mark WFRtmfpPartnerSession namespace

local WFRtmfpPartnerSession = {

	_rtmfpClient = false,
		-- ( WFRtmfpClient ) RTMFP object
		
	_maybeCommandFlowStream = false,
		--- ( ReplaySubject< WFRtmfpFlow > ) 
		
	_maybeHighPriorityFlowStream = false,
		-- ( ReplaySubject< WFRtmfpFlow > ) 
		
	_maybeLowPriorityFlowStream = false,
		-- ( ReplaySubject< WFRtmfpFlow > ) 

	_maybeNormalPriorityFlowStream = false,
		-- ( ReplaySubject< WFRtmfpFlow > ) 

	_isManagingFlowGUIDs = false,
		-- ( boolean ) True if we initialized session opening. False if a partner did.
	
	_sawGoodbyeStream = false,
		-- ( Rx.ReplaySubject( 1 ) ) True if this received a goodbye message on incoming stream
		
	_reestablishNewFlowsSubscription = false,
		-- ( Rx observer ) Sets _sessionNeedsToReestablishFlows and _newFlowCreationCurrentlyActive status.

	_sessionNeedsToReestablishFlows = false,
		-- ( boolean ) Indicates if the session needs to be reinitialized
	
	_newFlowCreationCurrentlyActive = false,
		-- ( boolean ) Indicates if the session is in the process of reinitializing
	
	_returnSessionSelfCloseTimerSubscr = false,
		-- ( Rx.SerialDisposable )
		-- 
		-- subscription to a timer that will cause self:close() if there are no flows.
	
	_closeFlowsCompletedForSignalIdsStream = false,
	_closeFlowsSignalIdStream = false,
	_closeFlowsCompletedForSignalIdsStreamSubscr = false,
	-- 		Rx.Subject()s
	-- 		
	-- 	used to implement synchronous calls to _closeFlows

		
	_incomingStream = false,
		-- ( ReplaySubject< what, ... > ) A stream for receiving data from a session partner

	_incomingStreamSubscription = false,
		-- ( Rx Disposable ) A subscription to the _incomingStream to handle special messages.

	_partnerAddress = false,
		-- ( string ) The network address of the partner
		
	_partnerPort = false,
		-- ( int ) The network port of the partner

	_establishFlowsResultStream = false,
		-- ( BehaviorSubject ) Resolves once the session has opened or reinitialized.

	_isShutdown = false,
		-- (boolean) Set when shut down

	_flowTerminatedSubscriptions = false,
		-- (table ) A table of flow status subscriptions.
		-- key: Rx disposable
		-- value: true

	_pingPartnerSubscription = false,
		-- ( Rx Disposable ) A subscription to the ping-partner heartbeat.
		
	_disposeOnClose = false,
		-- ( table ) A table of disposables to dispose when closed.
		-- key: Rx disposable
		-- value: true

	WFCloseSupport.addCloseSupportMethods,
	
}

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

local function isGoodbyeMessage( rtmfpRequest )

	if not rtmfpRequest then

		return false

	end

	local metadata = rtmfpRequest:getMetadata()
	
	if not metadata or not metadata._sessionPartnerCommand then

		return false

	end

	local ok, data = pcall( KSJsonDecoder.decodeWebSafeJson, 
								rtmfpRequest:getData() )
				
	if not ok or not data or data.command ~= _goodbyeCommand then

		return false

	end

	return true

end

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

local function maybeCloseSelfIfGoodbyeSeen( self )

	return function( rtmfpRequest )

		if self._isManagingFlowGUIDs then

			-- if self has created the original flow guids, then it shouldn't
			-- respond to goodbye
			return

		end

		if DISABLE_SESSION_TIMEOUT then

			return

		end

		if not isGoodbyeMessage( rtmfpRequest ) then

			return

		end

		self._sawGoodbyeStream:onNext( true )

		logger:infof( "Saw goodbye message!" )

		-- The RTMFP Partner has ended the session
		self:close()

	end

end

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

local function maybeDisposeIncomingStreamSubscription( self )

	return function()
		
		if not self._incomingStreamSubscription then

			return

		end

		self._incomingStreamSubscription:dispose()

		self._incomingStreamSubscription = nil

	end

end

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

function WFRtmfpPartnerSession:__init( rtmfpClient )
	
	check( 'WFRtmfpPartnerSession', 'WFRtmfpClient' )
	
	logger:debugf( "__init( rtmfpClient ID = %s )", rtmfpClient:getId() )

	self._rtmfpClient = rtmfpClient

	self._flowTerminatedSubscriptions = setmetatable( {}, { __mode = 'k' } )

	self._maybeCommandFlowStream = Rx.BehaviorSubject()
	self._maybeHighPriorityFlowStream = Rx.BehaviorSubject()
	self._maybeLowPriorityFlowStream = Rx.BehaviorSubject()
	self._maybeNormalPriorityFlowStream = Rx.BehaviorSubject()

	self._sawGoodbyeStream = Rx.ReplaySubject( 1 )

	-- Set up event stream for managing incoming message
	
	self._incomingStream = ReplaySubject( 1 )

	self._incomingStreamSubscription = 
		self._incomingStream
			:subscribe(
				maybeCloseSelfIfGoodbyeSeen( self ),
				maybeDisposeIncomingStreamSubscription( self ),
				maybeDisposeIncomingStreamSubscription( self ) )

	self._returnSessionSelfCloseTimerSubscr = Rx.SerialDisposable()

	self:addCloseDisposable( self._returnSessionSelfCloseTimerSubscr )

end

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

function WFRtmfpPartnerSession:sawGoodbyeStream()

	-- fires only if goodbye seen
	return self._sawGoodbyeStream
			:filter( function( sawGoodbye ) return sawGoodbye end )
			:take( 1 )

end

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

--[[!

	\brief Retrieve the Command Flow.
	
	\return An Observable that fires once with the WF RTMFP Flow.
	
--]]

function WFRtmfpPartnerSession:getCommandFlowOnce()

	return self:maybeGetFlowOnceByPriorityName( "command" )
	
end

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

--[[!

	\brief Retrieve the Small Data Flow.
	
	\return An Observable that fires once with the WF RTMFP Flow.
	
--]]

function WFRtmfpPartnerSession:getDefaultDataFlowOnce()

	return self:maybeGetFlowOnceByPriorityName( "normal" )
			
end

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

--[[!

	\brief Ping the other partner.
	
	\return void
	
--]]

function WFRtmfpPartnerSession:pingPartner()

	check( 'WFRtmfpPartnerSession' )
	
	if self._isShutdown then

		local errMsg = "WFRtmfpPartnerSession:pingPartner() fail: instance has already been shutdown."

		logger:warn( errMsg )

		return Observable.throw( errMsg )

	end
	
	return self:getDefaultDataFlowOnce() 
			:flatMapLatest( 

				function( defaultDataFlow )

					return defaultDataFlow
							:send( _pingMessage, 
									partnerSessionCommandMetadata() )

				end )
		
end

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

--[[!

	\brief Setup a "heart beat" monitor between the RTMFP partners
	
	\param maybeOnErrorCallbackFn (optional function) This function is called on
		every failed beat.
		
	\param maybeOnSuccessCallbackFn (optional function) This function is called on
		every successful beat.
		
	\return void
	
--]]

function WFRtmfpPartnerSession:setupHeartBeatMonitor( maybeOnErrorCallbackFn, maybeOnSuccessCallbackFn )

	check( 'WFRtmfpPartnerSession', '?function', '?function' )

	if self._isShutdown then

		logger:warn( "WFRtmfpPartnerSession:setupHeartBeatMonitor fail: instance has already been shutdown." )

		return

	end

	if self._pingPartnerSubscription then
		self._pingPartnerSubscription:dispose()
		self._pingPartnerSubscription = nil
	end

	local noopFn = function() --[[noop]] end
	
	maybeOnSuccessCallbackFn = maybeOnSuccessCallbackFn or noopFn

	-- SPAM [jandrew 2017-06-26]: The following is a blind attempt to fix the
	-- 'daemon keeps crashing' error. It appears that when a computer
	-- goes to sleep, the connection between the daemon and Nimbus process
	-- is for some reason lost, which forces the daemon to restart. Hopefully
	-- this explicit pinging every second will prevent that.

	self._pingPartnerSubscription = Observable.timer( 2, 2, WFRxScheduler.sharedScheduler() ) 
	
		:flatMap( function()
			return self:pingPartner():concat( Observable.just( true ) )
		end )
		
		:tap( maybeOnSuccessCallbackFn, noopFn, maybeOnSuccessCallbackFn )
		
		:catch( function( ... )
		
			if maybeOnErrorCallbackFn then
				maybeOnErrorCallbackFn()
			end
			
			return Observable.empty()
			
		end )
		
		:subscribe()
		
end

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

--[[!

	\brief Retrieve flow by priority
	
	\param (string) name of the desired flow (e.g. 'command', 'high', 'normal', 'low', etc.)
	
	\return An Observable that fires once with WF RTMFP Flow object with the given 
		priority or nil if none exists for the given priority
	
--]]

function WFRtmfpPartnerSession:maybeGetFlowOnceByPriorityName( priority )
	
	check( 'WFRtmfpPartnerSession' )
	
	if self._isShutdown then

		local errMsg = "WFRtmfpPartnerSession:maybeGetFlowOnceByPriorityName() failure: instance  has already been shutdown."

		logger:warn( errMsg )

		return Observable.throw( errMsg )

	end
	
	local maybeFlowStream
	
	if priority == 'high' then
		maybeFlowStream = self._maybeHighPriorityFlowStream
	elseif priority == 'low' then
		maybeFlowStream = self._maybeLowPriorityFlowStream
	elseif priority == 'normal' then
		maybeFlowStream = self._maybeNormalPriorityFlowStream
	elseif priority == 'command' then
		maybeFlowStream = self._maybeCommandFlowStream
	end

	if not maybeFlowStream then

		return Rx.Observable.throw( string.format( "Invalid priority: %s", priority ) )

	end
	
	return maybeFlowStream
			:filter( 
				function( maybeFlow )

					return maybeFlow

				end )
			:take( 1 )
	
end

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

local function startCloseIfNoReturnSessionOpenedTimer( self )

	-- If the session needs reinitialization, wait a set period of time for
	-- whomever initiated the session to reinitiate. If we hear nothing,
	-- assume the worst and close.

	-- Why 15 seconds? The old wait time was 0 seconds which didn't work
	-- and caused the hiccup NMB-6031. 15 seconds is sort of arbitrary, but
	-- it's more than enough time for the UI process to attempt to
	-- reinitialize a session ( the UI process will detect and restart the
	-- daemon process in, at most, 10s ) but not too much time such that
	-- the daemon appears zombied.
	-- 
	-- self._returnSessionSelfCloseTimerSubscr:setDisposable() ensures
	-- we cancel the previous timer if any

	local maxWaitTime = 15

	self._returnSessionSelfCloseTimerSubscr
		:setDisposable(
			Observable.timer( maxWaitTime, WFRxScheduler.sharedScheduler() )
				:take( 1 )
				:filter(
					function()
			
						return self._sessionNeedsToReestablishFlows

					end )
				:subscribe(

					function()

						logger:warn( "The RTMFP partner session is CLOSING because it has not been reinitialized." )

						self:close()
					
					end ) )

end

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

local function closeMostRecentCommandFlowStream( rtmfpClient, commandFlowStream ) 

	return commandFlowStream
			:take( 1 )
			:flatMapLatest( 
				function( maybeCommandFlow )
					
					logger:infof( "closeMostRecentCommandFlowStream: got a flow" )

					if not maybeCommandFlow then

						return Observable.just( true )

					end
				
					local commandFlow = maybeCommandFlow

					rtmfpClient
						:unlinkPartnerFlowToSessionStream( 
							commandFlow:getGuid() )
					
					return Rx.Observable.concat(

								-- say goodbye safely
								commandFlow
									:send( _goodbyeMessage, 
											partnerSessionCommandMetadata() )
									:catch( Observable.empty() ),

								-- shutdown the flow
								commandFlow:shutdown() )

				end )
			:takeLast( 0 )
			:tapOnCompleted(
				function()

					logger:infof( "command flow shutdown is complete" )

				end )

end

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

local function closeMostRecentFlowStream( rtmfpClient, flowStream )

	return flowStream
			:take( 1 )
			:flatMapLatest( 
				function( flow )
					
					logger:infof( "closeMostRecentFlowStream: got the flow" )
					
					if not flow then

						return Observable.just( true )

					end
		
					rtmfpClient
						:unlinkPartnerFlowToSessionStream( flow:getGuid() )
		
					return flow:shutdown()

				end )
			:takeLast( 0 )
			:tapOnCompleted(
				function()

					logger:infof( "closeMostRecentFlowStream: shutdown is complete" )

				end )

end

--------------------------------------------------------------------------------
				
local function finishThenCreateNewFlowStream( flowStream )

	flowStream:onCompleted()

	return Rx.BehaviorSubject()

end

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

local function resetFlowStreams( self )

	-- Shut down all the existing flow streams and create new flow streams.
	
	self._maybeCommandFlowStream = 
		finishThenCreateNewFlowStream( self._maybeCommandFlowStream )

	self._maybeHighPriorityFlowStream = 
		finishThenCreateNewFlowStream( self._maybeHighPriorityFlowStream )

	self._maybeLowPriorityFlowStream = 
		finishThenCreateNewFlowStream( self._maybeLowPriorityFlowStream )

	self._maybeNormalPriorityFlowStream = 
		finishThenCreateNewFlowStream( self._maybeNormalPriorityFlowStream )

end

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

local function closeFlowResultStreamForSignalId( self, signalId, reason )

	check( 'WFRtmfpPartnerSession' )

	self._sessionNeedsToReestablishFlows = true

	return Rx.Observable.create(

		function( observer )

			logger:infof( "closeFlowResultStreamForSignalId %s begins for %s", signalId, reason )

			local closeAllFlowsCompleteStream =
				Rx.Observable.concat(
						closeMostRecentCommandFlowStream( 
								self._rtmfpClient, 
								self._maybeCommandFlowStream ),
						closeMostRecentFlowStream( 
								self._rtmfpClient, 
								self._maybeHighPriorityFlowStream ),
						closeMostRecentFlowStream( 
								self._rtmfpClient,
								self._maybeLowPriorityFlowStream ),
						closeMostRecentFlowStream( 
								self._rtmfpClient,
								self._maybeNormalPriorityFlowStream ) )
					:takeLast( 0 )

			return Rx.Observable.concat( 
						closeAllFlowsCompleteStream,
						Rx.Observable.just( signalId, true ) )
					:catch( 
						function( errStr )

							logger:warnf( "closeFlowResultStreamForSignalId %s FAILED for %s with error", 
												signalId, reason, errStr )

							return Rx.Observable.just( signalId, false, errStr )

						end )
					:map( 
						function( theSignalId, ok, maybeErrStr )
					
							resetFlowStreams( self )

							return theSignalId, ok, maybeErrStr
						
						end )
					:subscribe( observer )

	end )
	
end

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

local _currCloseFlowSignalId

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

local function closeFlows( self, reason )

	reason = reason or "no reason given"

	logger:infof( "Close flows called by: %s", reason )

	-- do not allow overlapping _closeFlows call. We do this by implementing a
	-- stream that watches for close signal id. When a signal id is received,
	-- all flows are closed, and the signal id is relayed down the stream.
	-- 

	if not self._closeFlowsCompletedForSignalIdsStream then

		self._closeFlowsCompletedForSignalIdsStream = Rx.Subject()

		self._closeFlowsSignalIdStream = Rx.Subject()

		self._closeFlowsCompletedForSignalIdsStreamSubscr =
			self._closeFlowsSignalIdStream
				:map(
					function( signalId, theReason )

						logger:infof( "STARTING closeFlows signalId %s which was initiated because: %s", 
										signalId, theReason)

						-- closeFlowResultStreamForSignalId waits for all the
						-- flows to close and then emits the signalId
						return closeFlowResultStreamForSignalId( 
									self, signalId, theReason )
								-- we delay transmitting the results so that we
								-- don't cause stack overflows. the closeFlows
								-- chain uses a lot of Rx.Observable.concat,
								-- which recurse. :flatMerge() also recurses,
								-- so the :delay is needed to break out of
								-- recursion
								:delay( 0.02, WFRxScheduler.sharedScheduler() )

					end )
				-- :flatMerge( 1 ) means we only allow one 
				-- closeFlowResultStreamForSignalId() to run. A new one won't
				-- start until the current one completes.
				:flatMerge( 1 )
				:subscribe( self._closeFlowsCompletedForSignalIdsStream )

	end

	local closeFlowsCompletedForSignalIdsStream = 
			self._closeFlowsCompletedForSignalIdsStream

	local closeFlowsSignalIdStream = self._closeFlowsSignalIdStream

	_currCloseFlowSignalId = ( _currCloseFlowSignalId or 0 ) + 1

	local thisSignalId = _currCloseFlowSignalId 

	return Rx.Observable.create(

		function( observer )

			local subscr = closeFlowsCompletedForSignalIdsStream
				:filter(
					function( signalId )

						return signalId == thisSignalId

					end )
				:take( 1 )
				:flatMapLatest(
					function( _, ok, maybeErrStr )

						if not ok then

							return Rx.Observable.throw( maybeErrStr )

						end

						return Rx.Observable.empty()

					end )
				:tapOnCompleted(
					function()

						logger:infof( "ONCOMPLETED closeFlows signalId %s which was initiated because: %s", thisSignalId, reason )

					end )
				:subscribe( observer )

			logger:infof( "SENDING closeFlows signalId %s which was initiated because: %s", thisSignalId, reason )

			closeFlowsSignalIdStream:onNext( thisSignalId, reason )
			
			return subscr

		end )
	:replay( nil, 1 )
	:refCount()

end

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

local function closeFlowsAndThrow( self, reason, errMsg )

	return closeFlows( self, reason )
			:concat( 
				Rx.Observable.create( 
					function( observer )
						
						logger:infof( "closeFlowsAndThrow: %s %s", 
										reason, errMsg )

						observer:onError( errMsg ) 
						
					end ) )

end

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

function WFRtmfpPartnerSession:maybeReinitializeFlows( reason )

	if self._isShutdown then

		logger:warn( "Won't reinitialize flows- the WFRtmfpPartnerSession is being shutdown." )

		return

	end

	logger:warnf( "The RTMFP partner session needs to be reinitialized because: %s", 
					reason )

	self._sessionNeedsToReestablishFlows = true

	if not self._isManagingFlowGUIDs then

		startCloseIfNoReturnSessionOpenedTimer( self )

		return

	end
	
end
		
--------------------------------------------------------------------------------

local function watchForFlowTermination( self, flow, flowName )

	if self._isShutdown then

		logger:warnf( "Won't watch for %s termination: the WFRtmfpPartnerSession has already been shutdown.", flowName )

		return

	end

	local subscr 

	local subscrCompletedInThisScope
	
	local function maybeCleanupSubscription()

		subscrCompletedInThisScope = true
		if not subscr then

			return

		end
		
		self._flowTerminatedSubscriptions[ subscr ] = nil 

		subscr:dispose()

		subscr = nil
		
	end

	subscr = 
		flow
			:flowOpenStatus()
			:filter( function( isOpen ) return not isOpen end )
			:take( 1 )
			:flatMapLatest(
				function( _, reason )

					return Rx.Observable.throw( reason )

				end )
			:subscribe(

				nil,

				function( errStr )

					local reinitReason =
							string.format( 
								"Problem with %s, so all session flows will be reinitialized. Reason: %s", 
								flowName, errStr )

					self:maybeReinitializeFlows( reinitReason )

					maybeCleanupSubscription()

				end,

				function ()

					maybeCleanupSubscription()

				end )

	if subscrCompletedInThisScope then

		subscr:dispose()

		return

	end
			
	self._flowTerminatedSubscriptions[ subscr ] = true
	
end

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

local function abandonCurrentFlowTerminatedSubscriptions( self )

	-- this is called before we are about to create new flows.

	for subscr in pairs( self._flowTerminatedSubscriptions ) do

		subscr:dispose()

		self._flowTerminatedSubscriptions[ subscr ] = nil
		
	end

end

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

local function createNewFlow( self, addr, port, flowStream, flowName, 
								flowPriority )
	
	local flow = self._rtmfpClient:openFlow( addr, port, flowPriority )

	flowStream:onNext( flow )
	
	watchForFlowTermination( self, flow, flowName )

	self._rtmfpClient
		:linkPartnerFlowToSessionStream( flow:getGuid(), self._incomingStream )

	return flow

end

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

local function validateHelloResponseJSON( 
				helloResponseJSON, verifyResponseFunction )

	local ok, helloResponse = 
		pcall( KSJsonDecoder.decodeWebSafeJson, tostring( helloResponseJSON ) )
	
	-- Verify the response type and the secret
	
	if not ok then

		return false, "bad json data in hello response"

	end
	
	if type( helloResponse ) ~= "table" then 
	
		return false, "Unable to decode hello response to table"

	end

	if helloResponse.command ~= _howdyCommand then

		return false, "Unexpected hello response type"

	end
	
	if helloResponse.secret ~= _theSecret then
		
			return false, "The RTMFP peer's secret did not match."

	end
	
	local didVerify, maybeErrorMessage = 
		verifyResponseFunction( helloResponse.other or {} )
	
	if not didVerify then

		return false, maybeErrorMessage or "Unable to verify the RTMFP peer's response."

	end
	
	return true
	
end

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

local function sendHiMessageResultStream( flowName, flow )

	return flow 
			:send( _hiMessage, partnerSessionCommandMetadata() )
			:tapOnError(

				function( errStr )

					logger:warnf( "Error sending 'hi' via %s: %s", 
									flowName, errStr )

				end )
			:tapOnCompleted(

				function()

					logger:infof( "Finished sending 'hi' via %s", flowName )

				end )

end

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

--[[!

	\brief Initiates a new session with an RTMFP partner session by opening three
	 Flow objects and performing the appropriate "HELLO"/"HOWDY" handshake.

	\param addr (string) The URL/IP of the RTMFP peer to connect to.
	\param port (integer) The port of the RTMFP peer to connect to.
	\param secret (optional string) The Partner Session secret. This must match the
		secret of the RTMFP peer.
	\param verifyResponseFunction (optional function) An optional function used to verify
		additional information within the response of the RTMFP partner. In the event
		the verify function succeeds, it should return true. If it fails, it should return
		false along with an error message. Example
		
		verifyResponseFunction( data )
		
			if data.version == myVersion
				return true
			else
				return false, "Incompatible version number"
			end
		
		end
		
	\return An observable that completes upon success, or resolves to an error state.
	
--]]

function WFRtmfpPartnerSession:openSessionWithPartner( addr, port, secret, verifyResponseFunction )

	check( 'WFRtmfpPartnerSession', 'string', 'number', '?string', '?function' )
	
	if self._isShutdown then

		local errMsg = "WFRtmfpPartnerSession:openSessionWithPartner() fails because instance has already been shutdown."

		logger:warn( errMsg )

		return Observable.throw( errMsg )

	end
	
	abandonCurrentFlowTerminatedSubscriptions( self )
	
	self._partnerAddress = addr

	self._partnerPort = port

	self._isManagingFlowGUIDs = true
	
	secret = secret or _theSecret

	verifyResponseFunction = verifyResponseFunction or function() return true end
	
	-- Create the command flow and register it with the RTMFP Client object.
	-- This allows the client to know when to forward along incoming messages
	-- to this partner's event stream.
	local commandFlow = 
			createNewFlow( 
				self, addr, port, self._maybeCommandFlowStream, 
				'commandFlow', 'high' )
	
	-- Create the high priority flow
	local highPriorityFlow =
			createNewFlow( 
				self, addr, port, self._maybeHighPriorityFlowStream,
				'highPriorityFlow', 'high' )
	
	-- Create the low priority flow
	local lowPriorityFlow =
			createNewFlow( 
				self, addr, port, self._maybeLowPriorityFlowStream,
				'lowPriorityFlow', 'low' )
	
	-- Create the normal priority flow
	local normalPriorityFlow =
			createNewFlow( 
				self, addr, port, self._maybeNormalPriorityFlowStream,
				'defaultFlow', 'normal' )

	-- Send a large "Hello" message to the peer on the command flow. This includes
	-- the Flow Guids, as well as a secret. The Flow Guids are included so that the
	-- session partner can open return flows using the same Guids.
	
	local helloMessage = 
		KSJsonEncoder.encodeOzUploadJson(
		{
	
			command = _helloCommand,

			secret = secret,

			commandFlowGuid = commandFlow:getGuid(),

			lowPriorityFlowGuid = lowPriorityFlow:getGuid(),

			normalPriorityFlowGuid = normalPriorityFlow:getGuid(),

			highPriorityFlowGuid = highPriorityFlow:getGuid(),
			
		} )
		
	-- In RTMFP, Flows do not open until data is sent across them. So we send
	-- flow across all of them now.
		
	return Rx.Observable.concat(

			-- start by sending "hi" messages, which are fire and forget, we
			-- group them with concat for clarity.
			Rx.Observable.concat(

				-- open big data flow by sending a hi message
				sendHiMessageResultStream( "lowPriorityFlow", lowPriorityFlow ),

				-- open small data flow by sending a hi message
				sendHiMessageResultStream( "normalPriorityFlow", normalPriorityFlow ),

				-- open high priority data flow by sending a hi message
				sendHiMessageResultStream( 
					"highPriorityFlow", highPriorityFlow ) ),

			-- now send a "hello" message, from which we expect a response
			commandFlow
				:requestData( 
					helloMessage, 
					partnerSessionCommandMetadata() ) )

			-- wait for and check our "hello" response

			:map( 
				function( helloResponseJSON )
				
					return validateHelloResponseJSON( 
							helloResponseJSON, verifyResponseFunction )

				end )

			-- if we got a valid "hello" response, modify our session to our liking
			:flatMapLatest( 

				function( success, maybeErrorMessage )
				
					if not success then
					
						return closeFlowsAndThrow( 
									self, 
									string.format( "openSessionWithPartner error: %s", 
														maybeErrorMessage ),
									maybeErrorMessage )
					
					end
			
					return commandFlow:modifySession( _partnerSessionModifiers )
			
				end )

			:tapOnCompleted( 
			
				function()
				
					self._sessionNeedsToReestablishFlows = false
				
				end )

			:tapOnError(

				function( errStr )

					logger:warnf( "openSessionWithPartner had error: %s", errStr )

				end )
			
			:replay( nil, 1 )
			
			:refCount()

end

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

--[[!

	\brief Reinitiates a session with the RTMFP partner if the session was closed.

	\return An observable that completes upon success, or resolves to an error state.

--]]

local _reopenSessionWithPartnerIfNeededSeq = 1

function WFRtmfpPartnerSession:reopenSessionWithPartnerIfNeeded( maybeWillReopenSessionCallback, maybeFinalizationCallback )

	check( 'WFRtmfpPartnerSession', '?function', '?function' )

	assert( self._isManagingFlowGUIDs, 
			"reopenSessionWithPartnerIfNeeded should only be called on sessions that have called openSessionWithPartner" )

	if self._isShutdown then

		local errMsg = "WFRtmfpPartnerSession:reopenSessionWithPartnerIfNeeded() fails- the instance has already been shutdown" 

		logger:warnf( errMsg )

		return Observable.throw( errMsg )

	end
	
	if not self._sessionNeedsToReestablishFlows then
		
		-- if the flows are ok, it's a no-op
		return Observable.empty()

	end
	
	if self._newFlowCreationCurrentlyActive then

		-- If the session is already reinitializing, just join that stream
		
		return self._establishFlowsResultStream

	end

	-- otherwise, re-establish our flows

	if self._reestablishNewFlowsSubscription then

		self._reestablishNewFlowsSubscription:dispose()

		self._reestablishNewFlowsSubscription = nil

	end

	logger:warnf( "About to close flows for session sequence %s and open new ones.",
					_reopenSessionWithPartnerIfNeededSeq )

	_reopenSessionWithPartnerIfNeededSeq = 
		( _reopenSessionWithPartnerIfNeededSeq or 0 ) + 1
	
	local thisReopenSessionWithPartnerIfNeededSeq =
			_reopenSessionWithPartnerIfNeededSeq

	self._newFlowCreationCurrentlyActive = true
	
	self._establishFlowsResultStream = 
	
		-- 1) Close the existing flows.
		closeFlows( self, "reopenSessionWithPartnerIfNeeded" )
			:concat(
		
				Rx.Observable.create( 

					function( observer )


					-- 2) notify callback when we are about to reinitialize

						logger:warnf( "Now reopening the RTMFP partner session seq %s",
										thisReopenSessionWithPartnerIfNeededSeq )

			
						if maybeWillReopenSessionCallback then

							maybeWillReopenSessionCallback()

						end

						-- 3) reestablish the flows
						return self:openSessionWithPartner( 
									self._partnerAddress, 
									self._partnerPort )
								:subscribe( observer )

					end ) 
					
				)

			-- 4) If we fail to close flows or open a partner session, we will
			--    run steps 1) - 3) again unless the RTMFP instance has closed.

			:retryWhen( 

				function( failures )

					return failures
						:map( 
							function( errorStr, errorCode )
						
								local shouldRetry =
										errorCode ~= WFRtmfpErrorCodes.INSTANCE_CLOSED

								logger:warnf( "Will%s reopen partner session affer error (errCode %s): %s", 
												shouldRetry and "" or "NOT", errorCode, errorStr )
							
								return shouldRetry
								
							end )

				end )
		
			:replay( nil, 1 )
		
			:refCount()
	
	self._reestablishNewFlowsSubscription = 
			self._establishFlowsResultStream
				:subscribe(
					nil,
					function( err )

						logger:errorf( "Error with re-establish flows (seq %s): %s", 
										thisReopenSessionWithPartnerIfNeededSeq, err )

					end,
					function()
					
						logger:warnf( "Finished reopening the RTMFP partner session seq %s",
										thisReopenSessionWithPartnerIfNeededSeq )

						if maybeFinalizationCallback then

							maybeFinalizationCallback()

						end
					
						self._sessionNeedsToReestablishFlows = false

						self._newFlowCreationCurrentlyActive = false
						
					end )
	
	return self._establishFlowsResultStream
			
end

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

local function createFlowWithGuid( self, flowStream, guid, flowName, flowPriority )

	local flow = self._rtmfpClient:openReturnFlow( guid, flowPriority )

	flowStream:onNext( flow )
	
	watchForFlowTermination( self, flow, flowName )

	self._rtmfpClient
		:linkPartnerFlowToSessionStream( guid, self._incomingStream )

	return flow

end

--[[!

	\brief Completes the opening of a session with an RTMFP partner. This method is
	called in response to openSessionToPeerP.

	\param rtmfpRequest (WFRtmfpRequest) The WFRtmfpRequest object associated with the
	'HELLO' message created by the partner.
	\param cmdInfo (table) Contains information about the session.
	\param otherData (optional table) Contains additional information to be sent to the partner.
	
	\return An observable that completes upon success, or resolves to an error state.
	
--]]

function WFRtmfpPartnerSession:openReturnSessionWithPartner( rtmfpRequest, cmdInfo, otherData )

	check( 'WFRtmfpPartnerSession', 'WFRtmfpRequest', 'table' )

	assert( not self._isManagingFlowGUIDs, 
			"openReturnSessionWithPartner should not be called on sessions that have called openSessionWithPartner" )

	if self._isShutdown then

		local errMsg = "Will not openReturnSessionWithPartner because the WFRtmfpPartnerSession instance has already been shutdown."

		logger:warn( errMsg )

		return Observable.throw( errMsg )
	end

	abandonCurrentFlowTerminatedSubscriptions( self )

	-- Create the command flow and register it with the RTMFP Client object.
	-- This allows the client to know when to forward along incoming messages to
	-- this partner session's event stream.
	
	local commandFlow = self._rtmfpClient:openReturnFlow( cmdInfo.commandFlowGuid, 'high' )

	self._maybeCommandFlowStream:onNext( commandFlow )
	
	watchForFlowTermination( self, commandFlow, "commandFlow" )

	self._rtmfpClient:linkPartnerFlowToSessionStream( commandFlow:getGuid(), self._incomingStream )
	
	-- Check for bogus handshakes
	
	if type( cmdInfo ) ~= "table" or
			cmdInfo.command ~= _helloCommand or
			cmdInfo.secret ~= _theSecret then

		return rtmfpRequest
				:respond( commandFlow, _goodbyeMessage, partnerSessionCommandMetadata() )
				:concat( closeFlowsAndThrow( 
							self, 
							"openReturnSessionWithPartner bogus handshake",
							"Unable to establish a session." ) )
				:replay( nil, 1 )
				:refCount()

	end
		
	-- Set up the flows based on the incoming handshake data
	
	-- Create the high priority flow
	local highPriorityFlow =
			createFlowWithGuid( 
				self, self._maybeHighPriorityFlowStream,
				cmdInfo.highPriorityFlowGuid, 
				'highPriorityFlow', 'high' )

	-- Create the Big Data Flow
	local lowPriorityFlow =
		createFlowWithGuid( 
			self, self._maybeLowPriorityFlowStream,
			cmdInfo.lowPriorityFlowGuid, 
			'lowPriorityFlow', 'low' )
	
	-- Create the Small Data Flow
	local normalPriorityFlow = 
			createFlowWithGuid( 
				self, self._maybeNormalPriorityFlowStream,
				cmdInfo.normalPriorityFlowGuid, 
				'defaultFlow', 'normal' )

	-- Send a "Howdy" message to the peer. This lets the peer know we now have a complete
	-- WF RTMFP partner session opened.
	
	local howdyMessage = 
			KSJsonEncoder.encodeOzUploadJson {
		
				command = _howdyCommand,

				secret = _theSecret,

				other = otherData or {},
				
			}
		
	return Rx.Observable.concat(
				-- send hi to establish lowPriorityFlow
				lowPriorityFlow:send( _hiMessage, partnerSessionCommandMetadata() ),
				-- send hi to establish normalPriorityFlow
				normalPriorityFlow:send( _hiMessage, partnerSessionCommandMetadata() ),
				-- send hi to establish highPriorityFlow
				highPriorityFlow:send( _hiMessage, partnerSessionCommandMetadata() ),
				-- send "HOWDY" in response along the command flow
				rtmfpRequest:respond( commandFlow, howdyMessage, partnerSessionCommandMetadata() ),
				-- modify the command flow session
				commandFlow:modifySession( _partnerSessionModifiers ) )
			:tapOnCompleted( function()
	
				-- clear this flag now that things are set up
				self._sessionNeedsToReestablishFlows = false
			
			end )
			:catch( function( errStr )
		
				return closeFlowsAndThrow( 
						self, 
						string.format( "openReturnSessionWithPartner: %s",
										errStr ),
						errStr )
		
			end )
			:replay( nil, 1 )
			:refCount()

end

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

--[[!

	\brief Closes this WF RTMFP partner session and all of it's Flow objects.

	\return <code>Observable</code> An observable that completes after the session
	has been properly closed.
	
--]]

function WFRtmfpPartnerSession:shutdown()

	check( 'WFRtmfpPartnerSession' )

	if self._isShutdown then
		-- Slight race condition here, if two callers call shutdown
		-- in quick succession, the second may complete before the 
		-- first.  However, as I think about how we manage lifetime
		-- and do a managed close of this object, the likelihood of two
		-- shutdown calls in a row seems small, and the problem with
		-- trying to support that via :replay() and :refCount() is that
		-- our call to closeFlows() could end up not being disposed
		-- (which could cause resource leaks) or if the second call
		-- happens much later after the first shutdown() has disposed
		-- it's subscription then it would still cause problems if the
		-- second subscription caused refCount() to re-subscribe.
		return Observable.empty()
	end

	self._isShutdown = true

	logger:debugf( "Closing WFRtmfpPartnerSession" )
	
	local pingPartnerSubscription = self._pingPartnerSubscription
	self._pingPartnerSubscription = nil
	
	if pingPartnerSubscription then
		pingPartnerSubscription:dispose()
	end
	
	return closeFlows( self, "shutdown" )
	
		:catch( Observable.empty() )
	
		:concat( 
			Rx.Observable.create(
			
				function( observer )
				
					logger:debugf( "Cleaning up WFRtmfpPartnerSession" )
					
					local incomingStream = self._incomingStream
					self._incomingStream = nil
				
					if incomingStream then

						logger:infof( "completing self._incomingStream for shutdown" )

						incomingStream:onCompleted()
						incomingStream:dispose()
					end
			
					self._rtmfpClient = nil

					observer:onCompleted()

				end ) )

end

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

--[[!

	\brief Add a disposable to be called during close.

	\param disposable (Rx.Disposable) A disposable to dispose if closed.

	\return nothing.

--]]

function WFRtmfpPartnerSession:addCloseDisposable( disposable )

	if self._isShutdown then

		logger:warn( "Will not addCloseDisposable the WFRtmfpPartnerSession instance has already been shutdown." )

		return

	end

	if not self._disposeOnClose then

		self._disposeOnClose = {}

		self:addCloseAction( 
			function()

				for disposeOnCloseDisposable in pairs( self._disposeOnClose ) do

					disposeOnCloseDisposable:dispose()

				end

				self._disposeOnClose = {}

			end )

	end

	self._disposeOnClose[ disposable ] = true

end

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

--[[!

	\brief Remove a disposable to be called during close.

	\param disposable (Rx.Disposable) The disposable to remove.

	\return nothing.

--]]

function WFRtmfpPartnerSession:removeCloseDisposable( disposable )

	if self._isShutdown then

		logger:warn( "Will not removeCloseDisposable- the WFRtmfpPartnerSession instance has already been shutdown." )

		return

	end

	if self._disposeOnClose then
		self._disposeOnClose[ disposable ] = nil
	end

end

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

--[[!

	\brief Close this object.

--]]

function WFRtmfpPartnerSession:close()

	return self:shutdown()
		:call( runUntilDone )
		
end

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

--[[!

	\brief Retrieve a stream for incoming messages from a connected
	partner session.

	\return <code>ReplaySubject<RTMFP requests></code> Incoming messages
	
--]]

function WFRtmfpPartnerSession:listen( maybeListenerName )

	local listenerName = maybeListenerName or "unnamed listener"
	
	check( 'WFRtmfpPartnerSession' )

	return Rx.Observable.create(

		function( observer )

			if self._isShutdown then

				logger:warn( "WFRtmfpPartnerSession:listen() will complete immediately: instance has already been shutdown." )

				observer:onCompleted()

				return

			end

			if not self._incomingStream then

				logger:warn( "WFRtmfpPartnerSession:listen() will complete immediately: no incoming stream to which to listen." )

				observer:onCompleted()

				return

			end

			logger:infof( "BEGIN :listen( %s ) to incoming stream", listenerName )

			return self._incomingStream
					:filter( 
						function( rtmfpRequest )
			
							local requestMetadata = rtmfpRequest:getMetadata()

							return not requestMetadata._sessionPartnerCommand
							
						end )
					:tapOnCompleted(
						function()

							logger:infof( "FINISH :listen( %s ) to incoming stream", listenerName )

						end )
					:subscribe( observer )

		end )

end

--==============================================================================
-- @mark unit test exports
-- 
-- Export some internal methods for the unit tests as class methods. Methods
-- referenced below can be accessed in unit tests via
-- 
-- WFRtmfpPartnerSession.unitTestExports.methodName( ... )

WFRtmfpPartnerSession.class = {

	unitTestExports = {

		closeFlows = closeFlows

	}
}

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

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

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

return WFRtmfpPartnerSession
