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

WFRtmfpClient.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2014 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

------------------------------------------------------------------------------]]
	
	-- WF.rtmfp.imp
local WFRtmfpInstance = require 'WFRtmfpInstanceImp' -- native
local WFRtmfpFlow = require 'WFRtmfpFlow'
local WFRtmfpRequest = require 'WFRtmfpRequest'

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

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

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

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


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

local rtmfpClientLog =  WFLogger 'WFRtmfpClient'
--rtmfpClientLog:enable( 'print' )

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

-- Setup custom argument checker used in this class
WFArguments.registerChecker( 'rtmfpRequestType', function( requestType )
	return type( requestType ) == 'string' and
		requestType == 'sendData' or
		requestType == 'requestData' or
		requestType == 'requestStream'
end )

local runUntilDone = WFRxRunUntilDone


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

local WFRtmfpClient

local _responseStreams = setmetatable( {}, { __mode = 'kv' } )
	-- key: (string) 
	-- 	request Guid
	-- value: (ReplaySubject) 
	-- 	data stream for that request

local _responseStreamDisposalWatchers = setmetatable( {}, { __mode = 'kv' } )
	-- key: (string) 
	-- 	request Guid
	-- value: (Disposable) 
	--  this disposable is linked to the response stream for the guid. If the
	--  response stream is disposed, both the _responseStreams and
	--  _responseStreamDisposalWatchers entries for the request guid are
	--  cleared.
	
local _partnerSessionStreams = setmetatable( {}, { __mode = 'kv' } )
	-- key: (string) 
	-- 	Flow Guid from a WF RTMFP Partner Session
	-- value: (ReplaySubject) 
	-- 	data stream for that flow

local _partnerSessionStreamDisposalWatchers = setmetatable( {}, { __mode = 'kv' } )
	-- key: (string) 
	-- 	request Guid
	-- value: (Disposable) 
	--  this disposable is linked to the response stream for the guid. If the
	--  response stream is disposed, both the _responseStreams and
	--  _responseStreamDisposalWatchers entries for the request guid are
	--  cleared.
	
local _openSendFlows = setmetatable( {}, { __mode = 'kv' } )
	-- key: (string) Flow Guid of an open Flow
	-- value: (Flow) The Flow associated with the Flow Guid
	
local _wfRtmfpClientInstance = false
local _hasRegisteredCompletionHandler = false

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

local function flowExceptionHandler( flowGuid )

	local openFlow = _openSendFlows[ flowGuid ]
	
	if openFlow then
		openFlow:_flowException()
		_openSendFlows[ flowGuid ] = nil
	end
	
end

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

local function getDispatchStreamAndType( requestGuid, flowGuid )

	local responseStream = _responseStreams[ requestGuid ]

	if responseStream then

		return responseStream, "response"

	end

	local partnerSessionStream = _partnerSessionStreams[ flowGuid ]
	
	if partnerSessionStream then

		return partnerSessionStream, "partner"

	end
	
	return _wfRtmfpClientInstance._incomingStream, "incoming"

end

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

local function incomingRequestHandler( status, ... )

	if not _wfRtmfpClientInstance then
	
		rtmfpClientLog:debugf( "This RTMFP instance has closed. Ignoring incoming data." )
		return
	
	end

	if status == 'error' then
		rtmfpClientLog:debugf( "Incoming request was an error: %s", ... )
		return
	end

	local farAddress, flowGuid, requestJson, requestData = ...
	
	-- All WF RTMFP messages contain metadata as JSON and
	-- data as KSData.
	
	assert( KSData.isData( requestData ), "Incoming request Data was not KSData." )
	
	-- Extract the request object and request GUID
	
	local decodedJson = KSJsonDecoder.decodeWebSafeJson_stripNull( requestJson )
	local request = decodedJson.request
	local requestGuid = request._wfrtmfp_guid
	
	if not requestGuid then
		rtmfpClientLog:debug( "Incoming request did not contain a request Id. Ignoring." )
		return
	end

	-- Check if the incoming message is associated with a request stream.
	-- If it is, we send the message across it.

	local dispatchStream, dispatchStreamType = 
			getDispatchStreamAndType( requestGuid, flowGuid )

	if dispatchStream:isDisposed() then

		rtmfpClientLog:warnf( "trying to send a request on disposed %s stream",
								dispatchStreamType )

		return

	end

	local isFinishRequest =
			KSData.length( requestData ) == 4 and 
			tostring( requestData ) == 'done' 

	-- a isFinishRequest is signal that no more data will be sent to a stream. For
	-- response streams, we use that as a cleanup.
	
	if isFinishRequest then
		
		-- we aren't going to use this request data any more, so we close it.
		requestData:close()

		if dispatchStreamType == "response" then

			-- this file manages the lifetime of response streams, so
			-- we close out the stream here.
			dispatchStream:onCompleted()

			dispatchStream:dispose()

			return

		end
		
		--[hari may 10 2021] - change from warnf to infof, since it's not an error 
		--and we close the submodel from the client side as well
		rtmfpClientLog:infof( "Unexpected 'done' request sent to %s stream. It is ignored and might be caused by the request being cleared on the client before the server sent done",
								dispatchStreamType )
		
		return

	end
		
	-- Note: At this point, requestGuid is the GUID generated by the RTMFP peer.

	request.guid = requestGuid
	request.data = requestData
	request.flowGuid = flowGuid
	request.farAddress = farAddress
	
	-- Create the new RTMFP request object
	
	local rtmfpRequest = WFRtmfpRequest( request )

	dispatchStream:onNext( rtmfpRequest )

end

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

local function completionHandler( message, ... )
	
	if message == "incoming" then
	
		return incomingRequestHandler( ... )
	
	elseif message == "flowException" then 

		return flowExceptionHandler( ... )
	
	end
	
end


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

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

-- @mark WFRtmfpClient namespace

WFRtmfpClient = {

	_instance = false,
		-- (WFRtmfpInstance native object)
		
	_instanceId = false,
		-- (string) The unique id of the RTMFP client.
		
	_incomingStream = false,
		-- (ReplaySubject) a stream for receiving data from RTFMP peers.
		
	_isShutdown = false,
		-- (boolean) Set when shutdown

	WFCloseSupport.addCloseSupportMethods,
		
}

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

--[[!

	\brief Creates and starts a new RTMFP instance

	\param port (optional, integer) The port of the RTMFP instance to create.
	\param options (optional, table) Options to configure the instance.

	\return RTMFP instance object
	
--]]

function WFRtmfpClient:__init( port, options )
	
	check( 'WFRtmfpClient', '?number', '?table' )
	
	rtmfpClientLog:debugf( "__init( port = %s, options = %s )", port, options )

	if _wfRtmfpClientInstance then
		rtmfpClientLog:debug( "Only one instance of WFRtmfpClient allowed per Lua universe." )
		return nil
	end

	port = port or 0
	options = options or {}

	-- Set up stream for managing incoming requests/data
	self._incomingStream = ReplaySubject( 1 )
	
	self._instanceId = KSIdGenerator.uuidInLowercaseHex()
	
	-- Create and start the RTMFP instance
	self._instance = WFRtmfpInstance( self._instanceId, port, options.loopbackOnly or false )
	
	if not _hasRegisteredCompletionHandler then
		self._instance:registerCompletionHandler( completionHandler )
		_hasRegisteredCompletionHandler = true
	end
	
	self._instance:run()

	_wfRtmfpClientInstance = self

end

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

--[[!

	\brief Retrieve the Id associated the WF RTMFP client.
	
	\return WF RTMFP Client Id
	
--]]

function WFRtmfpClient:getId()
	check( 'WFRtmfpClient' )
	return tostring( self._instance )
end

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

--[[!

	\brief Retrieve the Port associated the WF RTMFP client.
	
	\return WF RTMFP Client Port
	
--]]

function WFRtmfpClient:getPort()
	check( 'WFRtmfpClient' )
	return self._instance:getInstancePort()
end

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

local function removeGuidStreamLinks( 
				guid, guidToStreams, guidToStreamDisposalWatchers )

	guidToStreams[ guid ] = nil

	guidToStreamDisposalWatchers[ guid ] = nil

end

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

local function addGuidStreamLinks( 
				guid, stream, guidToStreams, guidToStreamDisposalWatchers )

	guidToStreams[ guid ] = stream

	if not Rx.Disposable:isClassOf( stream ) then

		return

	end

	-- guard against retaining a guid to stream ref if stream:dispose() is
	-- called elsewhere
	local streamDisposalWatcher =
		Rx.Disposable.create( 
			function()

				removeGuidStreamLinks( 
					guid, guidToStreams, guidToStreamDisposalWatchers )

			end )

	-- trigger above disposal if other code calls stream:dispose()
	streamDisposalWatcher
		:disposeWith( stream )

	guidToStreamDisposalWatchers[ guid ] = streamDisposalWatcher

end

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

--[[!

	\brief Link a request to a new response stream. This should only be called
	by a WFRtmfpFlow object.
	
	\return void
	
--]]

function WFRtmfpClient:createResponseStreamForGuid( requestGuid )
	check( 'WFRtmfpClient', '+string' )

	local responseStream = Rx.ReplaySubject( 1 )

	addGuidStreamLinks( 
		requestGuid, responseStream, 
		_responseStreams, _responseStreamDisposalWatchers )

	return responseStream

end

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

--[[!

	\brief Remove an existing response response stream. This should only be called
 	by a WFRtmfpFlow object.
	
	\return void

--]]

function WFRtmfpClient:removeResponseStream( requestGuid )
	check( 'WFRtmfpClient', '+string' )

	removeGuidStreamLinks( 
		requestGuid, _responseStreams, _responseStreamDisposalWatchers )

end

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

--[[!

	\brief Link a flow to a response stream. This should only be called
	by a WFRtmfpPartnerSession object.
	
	\return void
	
--]]


function WFRtmfpClient:linkPartnerFlowToSessionStream( flowGuid, stream )
	check( 'WFRtmfpClient', '+string', 'table' )

	addGuidStreamLinks( 
		flowGuid, stream, 
		_partnerSessionStreams, _partnerSessionStreamDisposalWatchers )

end

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

--[[!

	\brief Remove an existing response response stream. This should only be called
	by a WFRtmfpPartnerSession object.
 
	\return void

--]]

function WFRtmfpClient:unlinkPartnerFlowToSessionStream( flowGuid )
	check( 'WFRtmfpClient', '+string' )

	removeGuidStreamLinks( 
		flowGuid, _partnerSessionStreams, 
		_partnerSessionStreamDisposalWatchers )

end

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

local function makeObservableFromWRtmfpInstanceFunc( 
				observableCallback, wfRtmfpInstanceFunc, wfRtmfpInstance, ... )

	-- The native WFRtmfpInstanceImp object is an AgDirectObject, and all of
	-- its methods expect the first parameter to be a reference to itself (
	-- self ).  WFRxNativeObservable, however, always tries to call native
	-- functions with the observable callback as the first parameter. By
	-- passing makeObservableFromWRtmfpInstanceFunc as the 'native function' to
	-- WFRxNativeObservable, we can properly order the parameters.
	
	return wfRtmfpInstanceFunc( wfRtmfpInstance, observableCallback, ... )

end

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

--[[!

	\brief A simple, internal method to simplify native observable creation.
	
	\return <code>Observable</code> An observable. 
	
--]]

function WFRtmfpClient:_rxNativeHelper( wfRtmfpInstance, 
											wfRtmfpInstanceFuncName, ... )

	check( 'WFRtmfpClient', 'userdata', 'string' )

	local wfRtmfpInstanceFunc = wfRtmfpInstance[ wfRtmfpInstanceFuncName ]

	if type( wfRtmfpInstanceFunc ) ~= "function" then

		local errMsg = string.format( "WFRtmfpInstance is missing required method: %s", wfRtmfpInstanceFuncName )

		rtmfpClientLog:error( errMsg )

		error( errMsg )

	end

	return WFRxNativeObservable( 
			makeObservableFromWRtmfpInstanceFunc, 
				wfRtmfpInstanceFunc, wfRtmfpInstance, ... )
	
		:catch( 

			function( errorMaybeTable )

				if type( errorMaybeTable ) == 'table' and
						type( errorMaybeTable.errorMessage ) == "string" and
						type( errorMaybeTable.errorCode ) == "number" then

					rtmfpClientLog:errorf( "rtmfpInstance.%s returned error code %s: %s",
												wfRtmfpInstanceFuncName,
												errorMaybeTable.errorCode,
												errorMaybeTable.errorMessage )

					return Observable.throw( errorMaybeTable.errorMessage, errorMaybeTable.errorCode )

				end

				local errStr = errorMaybeTable

				if type( errorMaybeTable ) == 'table' then

					errStr = "got an error table, but it didn't have expected error data"

				elseif type( errorMaybeTable ) ~= 'string' then

					errStr = string.format( "got a non string, non table error: %s", errorMaybeTable )

				end

				rtmfpClientLog:errorf( "rtmfpInstance.%s threw an error: %s",
											wfRtmfpInstanceFuncName, errStr )

				return Observable.throw( errStr )
		
		end )

end

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

--[[!

	\brief Closes a RTMFP instance and all underlying Flows
	
	\return <code>Observable</code> that completes when the client is closed.
	
--]]

function WFRtmfpClient:shutdown()
										
	check( 'WFRtmfpClient' )
	
	if self._isShutdown then
		return Observable.empty()
	end

	self._isShutdown = true
	
	_wfRtmfpClientInstance = nil
	
	local incomingStream = self._incomingStream
	self._incomingStream = nil
	
	if incomingStream then
		incomingStream:dispose()
	end
	
	if self._instance then
	
		return self:_rxNativeHelper( self._instance, "closeInstance" )
		
			:tapOnCompleted( function()
			
				if _wfRtmfpClientInstance then
					_wfRtmfpClientInstance._instance = nil
				end
				
			end )
			
	else
		return Observable.empty()
	end

end

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

--[[!

	\brief Closes a RTMFP instance and all underlying Flows
	
	\return void
	
--]]

function WFRtmfpClient:close()
										
	check( 'WFRtmfpClient' )
	
	self:shutdown():call( runUntilDone )

end


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

--[[!

	\brief Opens an RTMFP Flow and returns the Flow object.

	\param url (string) The URL/IP of the RTMFP server to connect to.
	\param port (integer) The port of the RTMFP server to connect to.
	\param priority (optional, string) The priority of the Flow to open.
	Valid values are as follows:

		"lowest" : Lowest priority. Only use for very unimportant messages.
		"low" : Use for bulk data, low-priority responses.
		"normal" : Default.
		"high" : Use for high-priority command flows.
		"highest" : Highest priority. Use with caution. Only use for real-time streaming.
	
	\param bufferSize (optional, integer) The size of the Send & Recv buffer for the flow in bytes.
		
	\return <code>WFRtmfpFlow</code> The RTMFP Flow.
	
--]]

function WFRtmfpClient:openFlow( url, port, priority, bufferSize )

	check( 'WFRtmfpClient', 'string', 'number', '?string', '?number' )

	priority = priority or "normal"
		-- "normal" is the default priority.
		
	bufferSize = bufferSize or 0
		-- 0 Indicates to use the default WF RTMFP buffer size
		
	rtmfpClientLog:debugf( "openFlow( url = %s, port = %s, priority = %s, bufferSize = %s )", url, port, priority, bufferSize )

	-- Generate a GUID for this new flow.
	
	local flowGuid = KSIdGenerator.uuidInLowercaseHex()

	-- Create an observable. This completes when the flow is succesfully opened, and can
	-- be accessed from the flow object.
	
	local observable = self:_rxNativeHelper( 
		self._instance, "openFlow", 
		flowGuid, 
		url, 
		port, 
		priority, 
		bufferSize )
	
	-- Create the new flow object
	
	local flow = WFRtmfpFlow( self, flowGuid, observable )
	
	-- Store a reference to the Flow by it's Flow Guid. 
	-- We need a reference to it in case the native Flow object closes due to an exception.
	
	_openSendFlows[ flowGuid ] = flow
	
	-- Return the flow
	
	return flow
	
end

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

--[[!

	\brief Opens a return RTMFP Flow to a RTMFP peer that has a Flow opened to us
	using the Flow Guid of the peers Send Flow.

	\param flowGuid (string) Flow Guid of the Peer's Send Flow
	\param priority (optional, string) The priority of the Flow to open. See
	WFRtmfpClient:openFlow for more info.
	\param bufferSize (optional, integer) The size of the Send & Recv buffer for
	the flow in bytes.
	
	\return <code>WFRtmfpFlow</code> The RTMFP Flow.
	
--]]

function WFRtmfpClient:openReturnFlow( flowGuid, priority, bufferSize )

	check( 'WFRtmfpClient', '+string', '?string', '?number' )
		
	rtmfpClientLog:debugf( "openReturnFlow( flowGuid = %s )", flowGuid )

	priority = priority or "normal"
		-- "normal" is the default priority.
		
	bufferSize = bufferSize or 0
		-- 0 Indicates to use the default WF RTMFP buffer size

	-- Create an observable. This completes when the flow is succesfully opened, and can
	-- be accessed from the flow object.
	
	local observable = self:_rxNativeHelper( 
		self._instance, "openReturnFlow", 
		flowGuid, 
		priority, 
		bufferSize )
		
	-- Create the new flow object
	
	local flow = WFRtmfpFlow( self, flowGuid, observable )
	
	-- Store a reference to the Flow by it's Flow Guid. 
	
	_openSendFlows[ flowGuid ] = flow
	
	-- Return the flow
	
	return flow
	
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. Valid entries:
	
		1.) maxRTXTime
		
		the maximum time, in milliseconds, that retransmission of unacknowledged flow
		data or keepalive traffic on this session will be attempted before the session is
		declared dead and torn down.
		
		2.) maxIdleTime
		
		the time, in milliseconds, that a session may go continuously with no open
		flows in it before the session is closed.
		
		3.) nearCloseTime
		
		set the time, in milliseconds, that this session will spend retransmitting
		Session Close Request messages if this end initiated a session close action, before
		giving up and declaring the session closed and dead.
		
		4.) sessionKeepaliveTime

		set the maximum interval in milliseconds that this session may go without
		sending data requiring acknowledgement to the far end, before actively
		probing the far end to see that it's still there. a dead far end will be
		detected in at most keepaliveTime + maxRTXTime.

	\return <code>Observable</code> that completes when the session is modified.
	
--]]

function WFRtmfpClient:modifyFlowSession( flowGuid, sessionModifiers )

	check( 'WFRtmfpClient', 'string', 'table' )

	sessionModifiers.maxRTXTime = sessionModifiers.maxRTXTime or 0
	sessionModifiers.maxIdleTime = sessionModifiers.maxIdleTime or 0
	sessionModifiers.nearCloseTime = sessionModifiers.nearCloseTime or 0
	sessionModifiers.sessionKeepaliveTime = sessionModifiers.sessionKeepaliveTime or 0

	rtmfpClientLog:debugf( "modifyFlowSession( flowGuid = %s )", flowGuid )

	return self:_rxNativeHelper( 
		self._instance, "modifyFlowSession",
		flowGuid,
		sessionModifiers.maxRTXTime,
		sessionModifiers.maxIdleTime,
		sessionModifiers.nearCloseTime,
		sessionModifiers.sessionKeepaliveTime )

end

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

--[[!

	\brief Closes an RTMFP Send Flow and returns the response via an observable.

	\param flowID (string) The ID of the flow to close

	\return <code>Observable</code> that completes when the flow is closed.
	
--]]

function WFRtmfpClient:closeSendFlow( flowGuid )

	check( 'WFRtmfpClient', 'string' )

	rtmfpClientLog:debugf( "closeSendFlow( flowGuid = %s )", flowGuid )

	-- Remove our internal reference to the Flow
	
	_openSendFlows[ flowGuid ] = nil

	return self:_rxNativeHelper( self._instance, "closeSendFlow", flowGuid )

end

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

--[[!

	\brief Closes an RTMFP Recv Flow and returns the response via an observable.

	\param flowID (string) The ID of the flow to close

	\return <code>Observable</code> that completes when the flow is closed.
	
--]]

function WFRtmfpClient:closeRecvFlow( flowGuid )
										
	check( 'WFRtmfpClient', 'string' )

	rtmfpClientLog:debugf( "closeRecvFlow( flowGuid = %s )", flowGuid )

	return self:_rxNativeHelper( self._instance, "closeRecvFlow", flowGuid )
	
end

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

--[[!

	\brief Sends a RTMFP message and resolves when the message was sent, or when
	a respones was received.

	\param rtmfpRequestType (string) Indicates the type of message we're about to send.
	This has implications on how and when the message resolves. Valid message types are:
	
		1.) sendData
		2.) requestData
		3.) requestStream
	
	\param flowGuid (string) The GUID of the RTMFP Flow to perform the request on.
	\param data (\c KSData or string) The data to send with the request.
	\param metadata (optional, table) Metadata that's packaged as JSON and sent along with the request.
	
	\return <code>Observable</code> An observable that completes when the message is delivered,
	or when a response is received ( determined by the rtmfpMessageType parameter ).
	
--]]

function WFRtmfpClient:message( rtmfpRequestType, flowGuid, data, metadata )
								
	check( 'WFRtmfpClient', 'rtmfpRequestType', 'string', 'KSData|string', '?table' )
	
	rtmfpClientLog:debugf( "message( flowGuid = %s, metadata = %s )", flowGuid, metadata )
	
	-- Message configuration
	
	local requestGuid = KSIdGenerator.uuidInLowercaseHex()
		-- The guid used to resolve this message.
		
	local responseGuid = false
		-- The guid of the RTMFP Peers request. This is only set if we are calling message
		-- via a WFRtmfpRequest object.
		
	local expectingResponse = false
		-- Is the message expecting a response from the peer?
		
	local waitForResponse = false
		-- Should the message resolve when the data is sent, or when a response is received?
	
	-- If the metadata contains a response Guid, we know that this method was called
	-- from a WFRtmfpRequest object, and is a response to an RTMFP Peer.
	
	metadata = metadata and KSTableUtils.shallowcopy( metadata ) or {}
	
	if metadata._responseGuid then
		responseGuid = metadata._responseGuid
		metadata._responseGuid = nil
			-- Remove _responseGuid from the metadata, since this is just an internal setting
			-- and it shouldn't be visible to the user.
	end
	
	-- Analyze the message further to understand the expected response
	
	metadata._expectingResponse = false
	
	if rtmfpRequestType == 'requestData' or rtmfpRequestType == 'requestStream' then
		
		-- We expect responses from all request* messages
		
		expectingResponse = true
		
		if rtmfpRequestType == 'requestData' then
		
			-- We only want to wait for a response for requestData messages.
			-- We only expect one response for requestData messages.
	
			waitForResponse = true
			metadata._expectingResponse = 'one'
			
		elseif rtmfpRequestType == 'requestStream' then
		
			-- We expect many responses from requestStream messages.
			
			metadata._expectingResponse = 'many'
		
		end
		
	end
	
	--[[
		WF RTMFP messages are two-part. The message itself is composed of
	 	5 bytes that contains the length of the JSON ( 2 bytes of hex = 4
		characters, plus a NULL terminator ). Then, a JSON payload
		of the formerly mentioned length that contains information about the
		request. The last part of the message is a binary payload of the actual
		data as KSData, starting at the \0 of the JSON.
	
		JSON Payload:
		
		The JSON payload contained in the message must include a GUID.
		For messages expecting responses, or responses, the guid must match the requestGuid so
		we can properly resolve the request once a response is received. For requests
		not expecting responses, the guid must be random to prevent unexpected resolution
		of requests.
		
		Other fields may be added as needed.
		
		{
			"request": {
				"guid": "Lowercase Hex GUID",
				"metadata": {},
			}
		}
	--]]
	
	-- Construct the JSON payload
	
	local jsonData = false
	
	if metadata._overrideJsonPayload then
	
		-- This field is used in unit testing to create special requests for
		-- complete code coverage.
		jsonData = metadata._overrideJsonPayload
		
	else
	
		local requestBody = {}
		
		if responseGuid then
			requestBody[ "_wfrtmfp_guid" ] = responseGuid
		elseif expectingResponse then
			requestBody[ "_wfrtmfp_guid" ] = requestGuid
		else
			requestBody[ "_wfrtmfp_guid" ] = KSIdGenerator.uuidInLowercaseHex()
		end
		
		requestBody[ "metadata" ] = metadata
		
		local requestJson = {}
		
		requestJson[ "request" ] = requestBody
		
		jsonData = KSJsonEncoder.encodeWebSafeJson( requestJson )
	
	end
	
	-- Make sure our data is KSData.
	
	if not KSData.isData( data ) then
		data = KSData.make( data )
	end
	
	return self:_rxNativeHelper( self._instance, "request", requestGuid, flowGuid, jsonData, data, waitForResponse )

end

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

--[[!

	\brief Retrieve an observable stream of incoming RTMFP requests.

	\return <code>Observable<RTMFP requests></code> that fires with incoming 
	RTMFP requests
	
--]]

function WFRtmfpClient:listen()
	
	check( 'WFRtmfpClient' )

	return self._incomingStream
	
end

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

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

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

return WFRtmfpClient
