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

WFRxHttpClientImp.lua

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

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

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

This is the thread that manages access to the native implementation for WFRxHttpClient.
It should be accessed only by the public API class WFRxHttpClient.

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

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

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

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

	-- WF.core.lua.threading
local WFRequestDispatcher = require 'WFRequestDispatcher'

	-- WF.core
local KSDate = require 'KSDate'
local KSStringUtils = require 'KSStringUtils'
local WFConfiguration = require 'WFConfiguration'
local WFQueueing = require 'WFQueueing'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFCapabilities = require 'WFCapabilities'
local WFPlatformInfo = require 'WFPlatformInfo'

	-- WF.web
local KSJsonDecoder = require 'KSJsonDecoder'
local WFRxRateLimiter = require 'WFRxRateLimiter'

	-- WF.core.threading
local WFCurrentThread = require 'WFCurrentThread'

	-- Configurator
local ENABLE_USE_LIBCURL_HTTP_CLIENT = ( WFConfiguration.Shared.disableLibcurlHttpClient ~= true )
local ENABLE_HTTP_LOGGING = WFConfiguration.Shared.enableHttpLogging
if not WFPlatformInfo.isDesktop then
	ENABLE_HTTP_LOGGING = WFCapabilities.getCapability( "WFRxHttpClientImp.enableHttpResponseLog" )
end

local KSTableUtils_deepCopy
KSTableUtils_deepCopy = function( ... )
	KSTableUtils_deepCopy = require "KSTableUtils".deepCopy
	return KSTableUtils_deepCopy( ... )
end

	-- WF.web.http_client
local WFRxHttpClientUtils = require 'WFRxHttpClientUtils'
local useCurl = WFConfiguration.Http.useCurl or (WFPlatformInfo.isDesktop and ENABLE_USE_LIBCURL_HTTP_CLIENT)
local WFHttpClientImp_native
if useCurl then
	WFHttpClientImp_native = require 'WFHttpClientImp_curl'
else
	WFHttpClientImp_native = require 'WFHttpClientImp_native'
end

local WFHttpClientHeaderParser = require 'WFHttpClientHeaderParser'

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

local kMultipartRequestBoundary = '_846843213219879654635132_'

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

local httpConnectionStats = WFDiagnosticLog.makeRingBuffer {
	title = "Number of HTTP connections",
	accumulate = true,
	"GET", "PUT"
}

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

local kMaxNumSimultaneousRequests = WFConfiguration.Nimbus.maxNumSimultaneousHttpRequests or 10
	-- The number of slots in the HTTP rate limiter to allow at any given time. Different types
	-- of requests take up varying numbers of slots in the rate limiter.

local kRequestPriorityHighest = 5

local now = KSDate.timeNowForAnimation

local kTimings = setmetatable( {}, { __mode = 'kv' } )

local httpTimingStats = WFDiagnosticLog.makeRingBuffer {
	title = "Native HTTP client timings",
	"GET", "PUT"
}

local rateLimiter = WFRxRateLimiter.makeRateLimiter(
	WFQueueing.makePriorityQueueNew,
	kMaxNumSimultaneousRequests,
	kRequestPriorityHighest -- allow a single request when using the highest priority
)

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

local _errorObserver
local _httpHistorySubject, _httpHistoryObservable
local _httpActivityActive = false

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

local logger = WFLogger 'WFRxHttpClientImp'
--	:setLevel 'debug'

local kMessageType_Progress = 'progress'
local kMessageType_Result = 'result'
local kMessageType_RequestInProgress = 'requestInProgress'

local timeNow = assert( KSDate.timeNow, "KSDate.timeNow not found" )

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

local sUserAgent

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

local function getUserAgent()

	if not sUserAgent then
		sUserAgent = WFRxHttpClientUtils.getUserAgent()
	end

	return sUserAgent

end

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

-- @mark httpRequestSherpa (internal to this file)

local sherpaForRequest = {}
	-- key (native object): the native HTTP request object (only responds to :close)
	-- value (httpRequestSherpa instance)

local httpRequestSherpaMethods = {

	_httpRequest = false,
		-- (native object) the native HTTP request object (only responds to :close)

	_url = false,
		-- (string) URL we're fetching or putting to

	_options = false,
		-- (table) options passed to client thread get or put function

	_optData = false,
		-- (KSData or number) data to pass with request or total size if multipart post

	_performanceMetrics = false,
		-- (table) a table of performance metrics of the request, to send via WFPerformanceReport

	_headers = false,
		-- (table) headers received (they come in before data )

	_observer = false,
		-- observer that subscribed to this.

	_observable = false,
		-- Observable< what, ... > an observable for sending info back to the client.
		-- If what is "progress", then the ... = nBytesSoFar, nBytesExpected
		-- If what is "result", then ... = data

}

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

function httpRequestSherpaMethods:__init( url, options, optData, optPerformanceMetrics )

    -- Call getCapability here because on mobile platforms setCapability is not
	-- called prior to this file loading.
	ENABLE_HTTP_LOGGING = ENABLE_HTTP_LOGGING or WFCapabilities.getCapability( "WFRxHttpClientImp.enableHttpLogging" )

	if ENABLE_HTTP_LOGGING then
		logger:configure { logLevel = 'trace', action = 'logfile' }
	end

	self._url = url

	self._options = options
	if useCurl then
		self._options.loggerLevel = logger._priority
	else
		self._options.bandwidthRateLimit = nil
	end

	if _httpActivityActive then
		-- Request progress updates so activity sees intermediate values of
		-- bytesSoFar as the transfer occurs.
		self._options.sendProgress = 1
	end

	self._optData = optData or false
		-- If we don't put the "or false" clause in, then we error out with
		-- illegal read on this field later.

	httpConnectionStats:add( optData and 0 or 1, optData and 1 or 0 )
	kTimings[ self ] = now()

	self._observable = Observable.create( function( observer )

		assert( not self._observer, "Cannot subscribe multiple times" )
		
		self._observer = observer

		self._performanceMetrics = optPerformanceMetrics or {}
	
		self._performanceMetrics.startTime = now()

		local reqOrError, reason, msg = WFHttpClientImp_native.getOrPut( self._url,
															 self._options,
															 self._optData or nil )

		if reqOrError == 'error' then

			-- An error here means something was wrong with the parameters so log it.
			logger:errorf( "getOrPut returned error reason=%s msg=%s", reason, msg )

			self._observer:onError( reason, msg )

		else

			self._httpRequest = reqOrError

			sherpaForRequest[ self._httpRequest ] = self

		end

		if not ENABLE_HTTP_LOGGING and not _httpHistorySubject then
			if self._optData then

				local optDataType = type( self._optData )

				if optDataType ~= 'string' and optDataType ~= 'number' then
					-- release memory used by data ASAP
					self._optData:close()
				end

				self._optData = '< closed >'

			end
		end

		return function()
			self:close()
		end

	end )

end

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

function httpRequestSherpaMethods:close()

	-- Close off the HTTP request.

	httpConnectionStats:add( self._optData and 0 or -1, self._optData and -1 or 0 )
	local time = now() - (kTimings[ self ] or 0)
	kTimings[ self ] = nil
	httpTimingStats:add( self._optData and 0 or time, self._optData and time or 0 )

	local request = self._httpRequest

	logger:debugf( "Completed in %f s ( url = %s, options = %s )", time, self._url, self._options )

	if request then

		self._httpRequest = false
		request:close()

		sherpaForRequest[ request ] = nil

	end

	self._observer = '< closed >'
	self._observable = '< closed >'

end

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

local HttpRequestSherpa = Rx.Class 'HttpRequestSherpa'( httpRequestSherpaMethods )

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

local function contentTypeIsJson( headers )

	headers = headers or {}

	for _, h in ipairs( headers ) do
		if h.header == 'Content-Type' and h.value == 'application/json' then
			return true
		end
	end

	return false

end

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

local function sendHistory( sherpa, msgType, responseData )

	local performanceMetrics = sherpa._performanceMetrics or {}
	local startTime = performanceMetrics.startTime or 0
	local responseHeaders = sherpa._headers or {}
	local options = KSTableUtils_deepCopy( sherpa._options ) or {}
	local isMetadataToOz = ( options.requestMethod == 'POST' or options.requestMethod == 'PUT' )
		and contentTypeIsJson( options.requestHeaders )

	local request = {
		msgType = msgType,
		url = sherpa._url,
		options = options,
		startTime = startTime,
		duration = now() - startTime,
		nBytesSoFar = performanceMetrics.nBytesSoFar,
		nBytesExpected = performanceMetrics.nBytesExpected,
		responseHeaders = responseHeaders,
		requestData = isMetadataToOz and tostring( sherpa._optData ),
		responseData = responseHeaders[ 'Content-Type' ] == 'application/json' and tostring( responseData ),
	}

	_httpHistorySubject:onNext( request )

end

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

-- This function gets triggered from native code when it has data to report
-- for any of the outstanding requests.

 local function completionHandler( request, status, data, optData )

	local sherpa = sherpaForRequest[ request ]

	if not sherpa then
		local errorMsg = "Completion handler got called for a request that we don't recognize"
		logger:warnf( errorMsg )
		return errorMsg
	end
	
	if status == 'header' then

		WFHttpClientHeaderParser.standardizeHeaders( data )
		sherpa._headers = data
			-- Save headers off; we'll include them as part of the final response.

	elseif status == 'data' then

		if ENABLE_HTTP_LOGGING then

			local logEntryOptions = KSTableUtils_deepCopy( sherpa._options )
			for _, v in ipairs( logEntryOptions.requestHeaders or {} ) do
				if v.header == "Authorization" then
					v.value = "-"
				end
			end

			if string.find( sherpa._url, "photos%-stage%.adobe%.io" ) or string.find( sherpa._url, "photos%.adobe%.io" ) then
				if sherpa._optData then
					logger:tracef( "Http url = %s, options = %s sent = %s response = %s, response_headers = %s )", sherpa._url, logEntryOptions, tostring(sherpa._optData), tostring(data), sherpa._headers )
				else
					logger:tracef( "Http url = %s, options = %s response = %s, response_headers = %s )", sherpa._url, logEntryOptions, tostring(data), sherpa._headers )
				end
			elseif string.find( sherpa._url, "storage%.adbephotos%-stage%.com") or string.find( sherpa._url, "storage%.adbephotos%.com") then
				-- don't log out binary file response
				logger:tracef( "Http url = %s, options = %s response_headers = %s )", sherpa._url, logEntryOptions, sherpa._headers )
			end

		end

		if _httpHistorySubject then
			sendHistory( sherpa, 'data', data )
		end

		if ENABLE_HTTP_LOGGING or _httpHistorySubject then

			if sherpa._optData then

				local optDataType = type( sherpa._optData )

				if optDataType ~= 'string' and optDataType ~= 'number' then
					sherpa._optData:close();
				end

				sherpa._optData = '< closed >'

			end

		end

		local headers = sherpa._headers

		if not headers then
			logger:error( "Huh ... got data but not headers!" )
		elseif headers.status_code >= '400' then
			logger:infof( "HTTP error on %s: \n\n%s\n\n%s", sherpa._url, headers, data )
		end

		if headers then
			
			local timing = kTimings[ sherpa ]
			
			if timing then
				headers[ 'wf_time' ] = now() - timing
			end

		end

		if headers and headers.status_code < '400' then	
			sherpa._observer:onNext( 'result', 'data', data, headers )
		else
			sherpa._observer:onNext( 'result', 'error', data, headers )
			if _errorObserver then
				_errorObserver:onNext( data, headers )
			end
		end

		-- make sure sherpa._observer is not "< closed >"
		if type( sherpa._observer ) == 'table' then
			sherpa._observer:onCompleted()
		end

		if data and type( data ) ~= 'string' then
		
			WFRequestDispatcher.deferredClose( data )
				-- Close the KSData object after processing the current request.
				-- If we call deferredClose before the :onNext statements above,
				-- we risk closing the object _before_ we fire those data events.
				-- This was seen in bug WF-1294 on the _errorObserver stream.
				-- Just like :onNext this also applies to :onCompleted()
		end

		-- Performance Reporting
		
		local performanceMetrics = sherpa._performanceMetrics or {}
		
		if performanceMetrics.startTime then
			performanceMetrics.requestDuration = now() - performanceMetrics.startTime
			WFPerformanceReport.recordSingleValue( 'WFRxHttpClientImp: getOrPut', performanceMetrics )
		end
		
	elseif status == 'progress' then

		if _httpActivityActive then
			local performanceMetrics = sherpa._performanceMetrics or {}
			performanceMetrics.nBytesSoFar = data
			performanceMetrics.nBytesExpected = optData
			if sherpa._performanceMetrics == nil then
				sherpa._performanceMetrics = performanceMetrics
			end
		end

		sherpa._observer:onNext( 'progress', data, optData )	-- really nBytesSoFar, nBytesExpected

	elseif status == 'error' then

		if ENABLE_HTTP_LOGGING then
			logger:tracef( "Http url = %s, options = %s error = %s, %s )", sherpa._url, sherpa._options, tostring(data), tostring(optData) )
		end

		sherpa._observer:onNext( 'result', 'error', data, nil, optData )
		sherpa._observer:onCompleted()
		
		if _errorObserver then
			_errorObserver:onNext( data, nil, optData )
		end

		if _httpHistorySubject then
			sendHistory( sherpa, 'error', data, optData )
		end

	elseif status == 'log' then

		local output = "NativeImpl: " .. optData
		if data == 'warn' then
			logger:warnf( output )
		elseif data == 'trace' then
			logger:tracef( output )
		elseif data == 'debug' then
			logger:debugf( output )
		elseif data == 'info' then
			logger:infof( output )
		elseif data == 'error' then
			logger:errorf( output )
		elseif data == 'fatal' then
			logger:fatalf( output )
		else
			logger:infof( output )
		end

	else

		local errorMsg = string.format( "Bogus status from native code: %s", status )
		logger:errorf( errorMsg )
		return errorMsg

	end

	return true

end

WFHttpClientImp_native.registerCompletionHandler( completionHandler )

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

-- Return true if the headers specified by these options include a User-Agent.
-- Also return the header array, creating one in options.requestHeaders if necessary.

local function hasUserAgentHeader( options )

	local headers = options.requestHeaders

	if not headers then
		headers = {}
		options.requestHeaders = headers
	end

	local hasUserAgent = false

	for _, h in ipairs( headers ) do
		if h.header == 'User-Agent' then
			hasUserAgent = true
		end
	end

	return hasUserAgent, headers

end

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

local function hasContentTypeHeader( options )

	local headers = options.requestHeaders or {}
	options.requestHeaders = headers

	for _, h in ipairs( headers ) do
		if h.header == 'Content-Type' then
			return true
		end
	end

	return false

end

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

local offsetError = "Error, offset greater than length"

local function downgradeDataForLegacyNativeImplementations( options, data )

	local KSData = require 'KSData'

	local usingFilePath = options.dataFormat == 'filePath'
	if usingFilePath then

		data = KSData.memoryMappedContentsOfFile( data )
			-- GRUNGE [ arahn 2012-08-22]: Using MMAP means we can't relocate this
			-- file on Windows so long as the KSData is alive.	We must remember
			-- to close this before trying to move the file!
	end

	if ( options.offset and options.offset > 1 ) or options.length then

		do
			local dataStr = tostring( data )
			local offset = options.offset or 1
			local length = options.length or ( #dataStr - offset + 1 )
            if length < 0 then
				if usingFilePath then
					data:close()
				end
			    error( offsetError, 2 )
			end

			local endpos = offset + length - 1
			dataStr = string.sub( dataStr, offset, endpos )
			data = KSData.make( dataStr )
		end

		-- Ensure that temp lua string is cleaned up
		collectgarbage "collect"
		collectgarbage "collect"
		collectgarbage "collect"

	end

	options.dataFormat = 'data'
	options.offset = nil
	options.length = nil

	return data

end

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

local function elideFirstProgressEvent( messageType, extraProgressInfo, ... )

	-- The first progress event for an HTTP request has proved to be very
	-- innacurate, so we skip it. In the event that the HTTP request only
	-- had the one progress event, we replay it within gatherProgressData.

	if messageType == kMessageType_Progress then

		local _, totalBytes = ...

		extraProgressInfo.totalBytes = totalBytes

		extraProgressInfo.eventCount = extraProgressInfo.eventCount
				and extraProgressInfo.eventCount + 1 or 1

		if extraProgressInfo.eventCount == 1 then

			return false

		end

	end

	return true

end

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

local function gatherProgressData( messageType, extraProgressInfo, ... )

	extraProgressInfo.timestamp = timeNow()

	if messageType == kMessageType_RequestInProgress then

		if not extraProgressInfo.startTime then
			extraProgressInfo.startTime = timeNow()
		end

	elseif messageType == kMessageType_Result then

		if not extraProgressInfo.endTime then
			extraProgressInfo.endTime = timeNow()
		end

		if extraProgressInfo.eventCount == 1 then

			-- We ignored the first progress event because the first event is always
			-- pretty innacurate, but there was only one event. In this case we still
			-- need to provide the bytes-per-second, which we do using the request start
			-- and end times.

			local reissuedProgressEvent  = {
				startTime = extraProgressInfo.startTime,
				timestamp = extraProgressInfo.timestamp
			}

			return Observable.just( kMessageType_Progress, reissuedProgressEvent, extraProgressInfo.totalBytes, extraProgressInfo.totalBytes )
				:concat( Observable.just( kMessageType_Result, extraProgressInfo, ... ) )

		end

	end

	return Observable.just( messageType, extraProgressInfo, ... )

end

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

local function formatMultipartData( multipartContent )

	-- The following was adapted form AgHttpClient.postMultipart

	local boundaryPrefix = '--'

	local eol = '\r\n'

	local contentStr = ""

	local insertRaw = function( rawData )
		contentStr = string.format( "%s%s", contentStr, rawData )
	end

	for _, part in ipairs( multipartContent ) do

		local data = part.value

		if part.filePath then
			data = ""
		end

		insertRaw( boundaryPrefix )
		insertRaw( kMultipartRequestBoundary )
		insertRaw( eol )

		insertRaw( 'Content-Disposition: form-data' )

		if part.name then
			insertRaw( string.format( '; name=%q', part.name ) )
		end

		if part.filePath then

			if part.fileName then
				insertRaw( string.format( '; filename=%q', part.fileName ) )
			end

			insertRaw( eol )

			if part.contentType then
				insertRaw( string.format( 'Content-Type: %s', part.contentType ) )
				insertRaw( eol )
			else
				insertRaw( 'Content-Type: image/jpeg' )
				insertRaw( eol )
			end

			insertRaw( eol )

			local fileHandle = io.open( part.filePath, 'rb' )

			local fileContents = fileHandle:read( "*all" )

			insertRaw( fileContents )

			fileContents = nil
			fileHandle:close()
			fileHandle = nil

		else

			insertRaw( eol )

			if part.contentType then
				insertRaw( string.format( 'Content-Type: %s', part.contentType ) )
				insertRaw( eol )
			end

			insertRaw( eol )

			local dataStr = tostring( data )
			insertRaw( dataStr )

		end

		insertRaw( eol )

	end

	insertRaw( boundaryPrefix )
	insertRaw( kMultipartRequestBoundary )
	insertRaw( boundaryPrefix )
	insertRaw( eol )

	return contentStr

end

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

local function splitLines( str, pattern )

	return KSStringUtils.split( str, pattern or '\r?\n' )

end

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

local function extractMultipartBoundary( multipartHeaders )

	local boundary

	if multipartHeaders and type( multipartHeaders ) == 'table' then

		local contentType = multipartHeaders[ 'Content-Type' ]

		if contentType and string.find( contentType, "^multipart" ) then

			-- Match a boundary ending in ; or by end of string.

			local matchedBoundary = string.match( contentType, "boundary%=(.+);" )
					or string.match( contentType, "boundary%=(.+)$" )

			if matchedBoundary then

				boundary = string.format( "--%s", matchedBoundary )

			end

		end

	end

	return boundary

end

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

local function parseMultipartResponse( boundary, multipartResponse )

	local parts = WFRxHttpClientUtils.splitMultipartResponseAtBoundary( multipartResponse, boundary )

	local responses = {}

	for _, part in pairs( parts ) do

		local headerEnd, dataStart = string.find( part, "\r?\n\r?\n" )

		if headerEnd then

			-- Parse Headers

			local header = string.sub( part, 1, headerEnd )

			local headerLines = splitLines( header )

			local headers = {}

			for _, headerLine in ipairs( headerLines ) do

				local headerLineParts = KSStringUtils.split( headerLine, "%:%s*" ) or {}
				local key = headerLineParts[ 1 ]
				local value = headerLineParts[ 2 ]

				if key and value then

					value = string.gsub( value, '[\r\n]$', '' ) -- Remove trailing line breaks and new lines

					headers[ key ] = value

				end

			end

			-- Extract Part Data

			local partData = string.sub( part, dataStart + 1 )

			-- If the content type header is application/json, decode it to Lua

			if headers[ 'Content-Type' ] == 'application/json' then
				partData = KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty( partData )
			end

			-- Form Response

			local response = {
				headers = headers,
				data = partData
			}

			responses[ #responses + 1 ] = response

		end

	end

	return responses

end

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

local function maybeParseMultipartResponse( messageType, status, ... )

	if messageType == kMessageType_Result and status == 'data' then

		local data, headers = ...

		local boundary = extractMultipartBoundary( headers )

		if boundary then

			local dataStr = tostring( data )

			data = parseMultipartResponse( boundary, dataStr )

		end

		return messageType, status, data, headers

	end

	return messageType, status, ...

end

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

local desktopRequestDispatcherOnMessage, _disallowNewThreadRequestsForDesktop

if WFPlatformInfo.isDesktop then

	-- define a request dispatcher that will throw for all requests that
	-- come in after stopAllFutureHttpRequests is called
	desktopRequestDispatcherOnMessage = function( message, handler )

		WFRequestDispatcher.onMessage( message, function( ... )

			if _disallowNewThreadRequestsForDesktop then

				return Observable.throw( string.format( "cannot invoke %s: HTTP requests are no longer allowed", message ) )

			end

			return handler( ... )

		end )

	end

end

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

local WFRequestDispatcher_onMessage =
	WFPlatformInfo.isDesktop and
		desktopRequestDispatcherOnMessage or
		WFRequestDispatcher.onMessage 

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

WFRequestDispatcher_onMessage( 'getOrPut', function( url, options, optData )

	if not WFHttpClientImp_native.supportsFileAndOffset or options.forceLegacyFileAndOffset then

		logger:debug( "warning using downgradeDataForLegacyNativeImplementations" )

		local success
		success, optData = pcall( downgradeDataForLegacyNativeImplementations, options, optData )

		if not success then

			if optData == offsetError then
				return Observable.throw( offsetError )
            end

			return Observable.throw( 'fileio', optData )
		end

	end

	if options.multipartRequest and optData then

		if not hasContentTypeHeader( options ) then

			local headers = options.requestHeaders

			local contentTypeHeader = {
				header = 'Content-Type',
				value = string.format( 'multipart/form-data; boundary=%s', kMultipartRequestBoundary ),
			}

			headers[ #headers + 1 ] = contentTypeHeader

		end

		optData = formatMultipartData( optData )

	end

	local hasUserAgent, headers = hasUserAgentHeader( options )
	if not hasUserAgent then
		headers[ #headers + 1 ] = { header = 'User-Agent', value = getUserAgent() }
	end

	logger:debugf( "getOrPut( url = %s, options = %s )", url, options )
	
	local requestPriority = options.requestPriority
	local requestCost = options.requestCost
	local bypassRateLimiter = options.bypassRateLimiter or false
	local requestDescription = options.requestDescription or 'Unknown'
	local notifyWhenHttpRequestIsInProgress = options.notifyWhenHttpRequestIsInProgress or false
	
	local maybeRateLimitedObservable
	
	local perfTable = {
		url = url,
		requestMethod = options.requestMethod,
		requestPriority = requestPriority,
		requestCost = requestCost,
	}

	if not bypassRateLimiter then
	
		-- Use a flatMap so we can re-issue a new request
		-- if the rate limited observable needs to be rescheduled.
		-- This is neccessary because the HTTP sherpa observables
		-- cannot be restarted by re-subscribing to them.
	
		local observableToLimit = Observable.just( true )
			:flatMap( function()
				local sherpa = HttpRequestSherpa( url, options or {}, optData, perfTable )
				return sherpa._observable
			end )
		
		if notifyWhenHttpRequestIsInProgress then
		
			observableToLimit = observableToLimit:startWith( kMessageType_RequestInProgress )

		end
			
		maybeRateLimitedObservable = rateLimiter:enqueueObservable( 
				observableToLimit, requestPriority, requestCost, requestDescription )
				
	else
	
		local sherpa = HttpRequestSherpa( url, options or {}, optData, perfTable )
		maybeRateLimitedObservable = sherpa._observable
	
		if notifyWhenHttpRequestIsInProgress then
		
			maybeRateLimitedObservable = maybeRateLimitedObservable:startWith( kMessageType_RequestInProgress )

		end
	
	end

	local extraProgressInfo = {}

	return maybeRateLimitedObservable

		:map( maybeParseMultipartResponse )

		:map( function( messageType, ... )
			return messageType, extraProgressInfo, ...
		end )

		:filter( elideFirstProgressEvent )

		:flatMap( function( ... )

			return options.sendProgress
					and gatherProgressData( ... )
					or Observable.just( ... )

		end )

end )

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

WFRequestDispatcher_onMessage( 'setProxyCredentials', function( user, password )
		
	if WFHttpClientImp_native.setProxyCredentials then
		WFHttpClientImp_native.setProxyCredentials( user, password )
	end

	logger:debug( "setProxyCredentials( user = %s )", user )
	return true

end )

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

WFRequestDispatcher_onMessage( 'watchHttpErrors', function()
		
	if not _errorObserver then
		_errorObserver = Rx.Subject()
	end

	return _errorObserver

end )

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

WFRequestDispatcher_onMessage( 'setUserAgentAppName', function( appName )

	WFCapabilities.setCapability( "WFHttpClient_UserAgentAppName", appName );
	sUserAgent = nil

	return true

end )

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

WFRequestDispatcher_onMessage( 'DEBUG_dumpCoverageP', function( path )

	require 'SetupCoverage'.dumpCoverage( path )
	return true

end )

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

WFRequestDispatcher_onMessage( 'DEBUG_normalizeAppNumber', function( appNumber )

	return Observable.just( WFRxHttpClientUtils.normalizeAppNumber( appNumber ) )

end )

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

WFRequestDispatcher_onMessage( 'DEBUG_completionHandler', function( create, status, data, optData )

	local debugRequest = 'DebugSherpaRequest'

	if create then

		local mockDebugSherpa = {}
		for k, v in pairs(httpRequestSherpaMethods) do
			mockDebugSherpa[k] = v
		end
		mockDebugSherpa._url = "MockURL"
		mockDebugSherpa._options = {}
		
		mockDebugSherpa._observable = Observable.create( function( observer )
			mockDebugSherpa._observer = observer
			return function()
				sherpaForRequest[ debugRequest ] = nil	
			end
		end )

		sherpaForRequest[ debugRequest ] = mockDebugSherpa

		return mockDebugSherpa._observable

	else

		return completionHandler( debugRequest, status, data, optData )

	end

end )

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

local function getActiveRequests()

	local activeRequests = {}
	for _, httpRequest in pairs( sherpaForRequest ) do

		local performanceMetrics = httpRequest._performanceMetrics or {}
		local startTime = performanceMetrics.startTime or now()

		local logEntryOptions = KSTableUtils_deepCopy( httpRequest._options )

		local logEntry = {
			url = httpRequest._url,
			options = logEntryOptions,
			startTime = startTime,
			duration = now() - startTime,
			nBytesSoFar = performanceMetrics.nBytesSoFar,
			nBytesExpected = performanceMetrics.nBytesExpected,
			observerState = httpRequest._observer:isDisposed() and "disposed" or "active",
		}

		for _, v in ipairs( logEntryOptions.requestHeaders or {} ) do
			if v.header == "Authorization" then
				v.value = "-"
			end
		end

		table.insert( activeRequests, logEntry )

	end

	return activeRequests

end

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

WFDiagnosticLog.addLogSectionFunc( function()

	local threadName = WFCurrentThread.DEBUG_getThreadName() or '( unknown )'

	local activeRequests = getActiveRequests()

	return {

		name = "WFRxHttpClientImp: " .. threadName,
		id = 'WFRxHttpClientImp',
		weight = 401,
		outputForAutomation = true,
		parentSection = WFDiagnosticLog.Sections.HTTP,
		
		body = {
			activeRequests = activeRequests,
			connections = { httpConnectionStats:visualize "boxAndWhisker" },
			timings = { httpTimingStats:visualize "boxAndWhisker" },
		}
	}

end )

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

local WFRxScheduler

WFRequestDispatcher_onMessage( 'DEBUG_activityStream', function( refreshSeconds )

	_httpActivityActive = true

	WFRxScheduler = WFRxScheduler or require "WFRxScheduler"

	return Rx.Observable.timer( 0, refreshSeconds, WFRxScheduler.sharedScheduler() )
		:map( function()
			return getActiveRequests()
		end )

end )

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

WFRequestDispatcher_onMessage( 'DEBUG_historyStream', function()

	if not _httpHistorySubject then
		_httpHistorySubject = Rx.Subject()
		_httpHistoryObservable = _httpHistorySubject:share()
	end

	return _httpHistoryObservable

end )

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

if WFPlatformInfo.isDesktop then

--------------------------------------------------------------------------------
-- desktop only code
--------------------------------------------------------------------------------

	WFRequestDispatcher_onMessage( 'activeRequestCount', 
		function()

			local count = 0

			for _ in pairs( sherpaForRequest ) do

				count = count + 1

			end

			return count

		end )

--------------------------------------------------------------------------------
	-- stopAllRequestsAndWait() is intended for use at shutdown time to ensure
	-- that all outstanding requests are stopped and cleaned up. Requests
	-- running at shutdown can cause crashes. The waitDurationInSeconds
	-- parameter is used to allow extra time for WFAsyncTasks to cancel.

	WFRequestDispatcher_onMessage( 'stopAllRequestsAndWait', 
	function( waitDurationInSeconds )

		_disallowNewThreadRequestsForDesktop = true

		waitDurationInSeconds = tonumber( waitDurationInSeconds ) or 0

		-- First prevent the native http code from initiating new requests by
		-- calling WFHttpClientImp_native.shutdown() 
		if type( WFHttpClientImp_native.shutdown ) == "function"  then

			local ok, errStr = pcall( WFHttpClientImp_native.shutdown )

			if not ok then

				logger:errorf( "WFHttpClientImp_native.shutdown failed: %s", errStr )

			end

		end

		-- Now stop all the requests that are currently in progress. This
		-- assumes that native http requests are managed by the sherpa object
		-- map in this file
		local stoppedRequestCount = 0

		for _, sherpa in pairs( sherpaForRequest ) do

			stoppedRequestCount = stoppedRequestCount + 1

			-- closing the sherpa stops the native request
			sherpa:close()

		end

		if stoppedRequestCount == 0 or waitDurationInSeconds == 0 then
		
			-- no need to wait, return nothing immediately
			return Rx.Observable.empty()

		end

		WFRxScheduler = WFRxScheduler or require "WFRxScheduler"
		-- wait a bit to give WFAsyncTasks a chance to cancel in flight
		-- requests, then return nothing 
		return Rx.Observable.timer( 
					waitDurationInSeconds, 
					WFRxScheduler.sharedScheduler() )
				:takeLast( 0 )

	end )

end
