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

WFRxHttpClient.lua

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

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

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

	-- RxLua
local Rx = require 'Rx'

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

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

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

	-- WF.core.utils
local WFPlatformInfo = require 'WFPlatformInfo'
local WFArguments = require 'WFArguments'
local KSData = require 'KSData'
local KSDate = require 'KSDate'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFCurrentThread = require 'WFCurrentThread'
local KSTableUtils = require 'KSTableUtils'
local KSStringUtils = require 'KSStringUtils'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFConfiguration = require 'WFConfiguration'

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

local httpClientLog = WFLogger( 'WFRxHttpClient' )
--	:setLevel( 'DEBUG' )

local check = assert( WFArguments.check )

local runUntilDone = WFRxRunUntilDone

local now = KSDate.timeNowForAnimation

KSData.registerArgumentChecker()

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

local openHttpRequestCount = 0

local totalDownloadTime = 0
local totalDownloadCount = 0
local totalUploadTime = 0
local totalUploadCount = 0
local totalRetries = 0
local totalFailures = 0

local kTimings
local httpStats
local httpTimingStats
local httpClientImp
	-- loaded when a connection is made

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

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

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

local function initHttpStats()

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

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

	httpStats = WFDiagnosticLog.makeRingBuffer {
		title = "Number of HTTP requests",
		accumulate = true,
		"uploads", "downloads"
	}

	WFDiagnosticLog.addLogSectionFunc( function()
	
		local threadName = WFCurrentThread.DEBUG_getThreadName() or '( unknown )'
		return {
	
			name = "HTTP Client : " .. threadName,
			id = 'WFRxHttpClient-' .. threadName:replace( ' :: ', '+' ),
			weight = 400,
			outputForAutomation = true,
			parentSection = WFDiagnosticLog.Sections.HTTP,
			
			body = {
				requests = { httpStats:visualize "boxAndWhisker" },
				openHttpRequestCount = openHttpRequestCount,
				timings = { httpTimingStats:visualize "boxAndWhisker" },
				totalDownloadTime = totalDownloadTime, -- includes time waiting for http request to get issued
				totalDownloadCount = totalDownloadCount,
				totalUploadTime = totalUploadTime, -- includes time waiting for http request to get issued
				totalUploadCount = totalUploadCount,
				totalRetries = totalRetries,
				totalFailures = totalFailures,
			}
		}
	
	end )

	return httpStats

end

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

local function httpStatsTransferRequestStarted( retryObservable, isPut )

	openHttpRequestCount = openHttpRequestCount + 1

	kTimings[ retryObservable ] = now()

	if isPut then
		httpStats:add( 1 )
	else
		httpStats:add( nil, 1 )
	end

end

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

local function httpStatsTransferRequestClosed( retryObservable, isPut )

	if kTimings[ retryObservable ] then

		openHttpRequestCount = openHttpRequestCount - 1

		kTimings[ retryObservable ] = nil

		if isPut then
			httpStats:add( -1 )
		else
			httpStats:add( nil, -1 )
		end

	end

end

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

local function httpStatsTransferRequestDone( retryObservable, isPut )
	
	local time = now() - kTimings[ retryObservable ]
	httpTimingStats:add( isPut and 0 or time, isPut and time or 0 )

	if isPut then
		totalUploadTime = totalUploadTime + time
		totalUploadCount = totalUploadCount + 1
	else
		totalDownloadTime = totalDownloadTime + time
		totalDownloadCount = totalDownloadCount + 1
	end

	httpStatsTransferRequestClosed( retryObservable, isPut )

end

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

local function isHigherPriority( p1, p2 )

	check( '?number', '?number' )

	p1 = p1 or 0
	p2 = p2 or 0

	return p1 > p2
	
end

--============================================================================--
--[[!
	\class WFRxHttpClient
	\brief This namespace provides basic HTTP services.
]]--

-- @mark WFRxHttpClient namespace

local WFRxHttpClient = {

	requestPriorityHighest = 5,
	requestPriorityHigh = 4,
	requestPriorityMedium = 3,
	requestPriorityLow = 2,
	requestPriorityLowest = 1,
	
	-- 'Cost' refers to how many HTTP slots the request takes up. An HTTP "slot" is 
	-- more conceptual than physical; every HTTP request is weighted by some cost.
	
	-- The default number of HTTP slots is 10. Using this, you can find the total number of
	-- concurrent requests that are able to run at once by dividing the total slot count 
	-- by the cost of the request.
	
	requestCostLow = 1,
	requestCostDefault = WFConfiguration.Http.requestCostDefault or 2,
	requestCostDownloadRendition = WFConfiguration.Http.requestCostDownloadRendition or 1,
	requestCostDownloadNegative = WFConfiguration.Http.requestCostDownloadNegative or 2,
	requestCostDownloadXmp = WFConfiguration.Http.requestCostDownloadXmp or 2,
	requestCostUploadNegative = WFConfiguration.Http.requestCostUploadNegative or 2,
	requestCostMaximizeBandwidth = WFConfiguration.Http.requestCostMaximizeBandwidth or 10,

	requestLimitBandwidthKB = WFConfiguration.Http.requestLimitBandwidthKB or 0,

	isHigherPriority = isHigherPriority

}

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

local requestMethodTimeouts = {
	DELETE = 60,
	GET = 60,
	HEAD = 60,
	POST = 240,
	PUT = 240,
}

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

local function getRetryDelayFromHeader( responseHeader )

	local have429 = responseHeader and responseHeader.status_code == '429'
	local retryAfter = have429 and tonumber( responseHeader[ 'Retry-After' ] )

	if have429 and retryAfter == nil and type(responseHeader[ 'Retry-After' ] ) == 'string' then
		-- NOTE: http-date formatted Retry-After fields are ignored. (Should they be handled?)
		httpClientLog:errorf( "Non numeric Retry-After value = %s", responseHeader[ 'Retry-After' ] )
	end

	return retryAfter

end

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

local function getOrPut( url, options, data, progressReporter )

	httpStats = httpStats or initHttpStats()

	options = options or {}

	local retryCount = -1

	options.requestMethod = options.requestMethod or 'GET'
	options.requestPriority = options.requestPriority or WFRxHttpClient.requestPriorityMedium
	options.requestCost = options.requestCost or WFRxHttpClient.requestCostDefault

	if not options.bypassRateLimiter then
		options.bandwidthRateLimit = options.limitBandwidthKB or WFRxHttpClient.requestLimitBandwidthKB
	end

	options.timeout = options.timeout or requestMethodTimeouts[ options.requestMethod ] or 60
		
	local perfTable = {
		openHttpRequestCount = openHttpRequestCount,
		url = url,
		requestMethod = options.requestMethod,
		requestPriority = options.requestPriority,
		requestCost = options.requestCost,
	}
	
	local maxAttempts = options.maxAttempts

	local abortIfNot429Retry = false
	if not maxAttempts then
		-- default the max attempts to 4, but this will only happen on a 429 retry response
		-- if we don't get a 429 retry the maxAttempts is effectively 1 due to abortIfNot429Retry
		maxAttempts = 4
		abortIfNot429Retry = true
	end

	local function retryIfError429Filter( responseData, responseHeader )

		if not responseHeader or ( abortIfNot429Retry and responseHeader.status_code ~= '429' ) then
			return false -- abort the retry
		end

		-- try again with the specified delay
		return true, getRetryDelayFromHeader( responseHeader )

	end

	local retryObservable
	
	retryObservable = WFRxRetryObservable {

		maxAttempts = maxAttempts,

		retryDelay = options.retryDelay or nil,

		retryFilter = retryIfError429Filter,

		factory = function()

			retryCount = retryCount + 1

			if progressReporter then
				options.sendProgress = options.forceIndeterminentProgress and -1 or 1
				options.notifyWhenHttpRequestIsInProgress = true
			end

			httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

			local hasData = not not data

			httpStatsTransferRequestStarted( retryObservable, hasData )

			return httpClientImp:request( 'getOrPut', url, options, data )

				:map( function( messageType, progressRequestTimings, ... )
				
					if progressReporter then

						if messageType == kMessageType_RequestInProgress then

							progressReporter:setProcessedCount( 0, true, progressRequestTimings )
							
						elseif messageType == kMessageType_Progress then
						
							local bytesTransferred, bytesExpected = ...
						
							if bytesExpected and bytesExpected ~= 0 then
								progressReporter:setTotal( bytesExpected, true, progressRequestTimings )
							end
						
							if bytesTransferred and bytesTransferred ~= 0 then
								progressReporter:setProcessedCount( bytesTransferred, true, progressRequestTimings )
							end

						elseif messageType == kMessageType_Result then

							local status = ...

							if status == 'data' then
								progressReporter:done( progressRequestTimings )
							end

						end

					end
					
					return messageType, ...

				end )

				:filter( function( messageType ) --, status, responseData, headers

					return messageType == kMessageType_Result or messageType == kMessageType_RequestInProgress

				end )

				:flatMapLatest( function( messageType, status, responseData, headers, errorInfo )

					if messageType == kMessageType_Result then
					
						if headers and retryCount > 0 then
							headers.retryCount = retryCount
						end

						if messageType == kMessageType_Result and status == 'error' then
							return Rx.Observable.throw( responseData, headers, errorInfo )
						end

						return Rx.Observable.just( responseData, headers )

					else
					
						return Rx.Observable.just( messageType )
					
					end

				end )

				:tap( function( messageType )

						if messageType == kMessageType_Result then
							httpStatsTransferRequestDone( retryObservable, hasData )
						end
						
					end,
					function()

						local attempts = retryCount + 1

						if progressReporter then

							progressReporter:setErrorCount( attempts )

							if attempts < maxAttempts then
							
								progressReporter:reset()
								
							else

								progressReporter:done()
								
							end

						end

						httpStatsTransferRequestClosed( retryObservable, hasData )

						if attempts ~= maxAttempts then
							totalRetries = totalRetries + 1
						end

					end )

				:finally( function()

					httpStatsTransferRequestClosed( retryObservable, hasData )

				end )

		end,

	}
	:catch( function( ... )

		totalFailures = totalFailures + 1

		-- Use ... so we can rethrow exactly what was passed in, it's not always two values.
		local responseHeader = select( 2, ... )

		local function throwOrReturnResults( ... )

			-- Map the http errors back to data here if requested so the WFRxRetryObservable
			-- above will function the same independent of the setting as it requires errors
			-- be thrown in order to retry.

			if options.returnHttpErrorsAsData and ( responseHeader and responseHeader.status_code and responseHeader.status_code >= '400' ) then
				-- Only convert http errors to values.
				return Rx.Observable.just( ... )
			end

			return Rx.Observable.throw( ... )

		end

		-- Apply the delay if we are going to actually propogate the '429' since 
		-- WFRxRetryObservable only delays before attempting a retry.

		local retryAfter = getRetryDelayFromHeader( responseHeader )

		if retryAfter then

			local responseData = select( 1, ... )

			return Rx.Observable.timer( retryAfter, WFRxScheduler.sharedScheduler() )
			:flatMapLatest( function()
				return throwOrReturnResults( responseData, responseHeader )
			end )

		end

		return throwOrReturnResults( ... )

	end )

	return WFPerformanceReport.wrapObservable( 'WFRxHttpClient: getOrPut', 
		retryObservable, perfTable )

end
--------------------------------------------------------------------------------
--[[!
	\brief Starts an HTTP GET request and returns the response as an Observable.

	\param url (string) The URL to fetch.
	\param options (optional, table) Options to modify the request.
	\param sequencing (optional, string) A string specifying the sequencing model
			Valid values include 'LIFO' and 'FIFO'. Defaults to FIFO if it doesn't
			find a match.
	\param progressReporter (optional, WFProgressReporter) An instance of 
			WFProgressReporter that is used to communicate transfer progress.

	\return <code>Observable< data, headers ></code> An observable that resolves
		to the response from HTTP server.

	\b Usage:

\code
local options = {
	requestHeaders = {					-- optional
		{ header = 'Content-Type', value = 'application/json' },
		{ header = 'Accept', value = 'application/json' },
	}
	requestMethod = 'GET',				-- optional
				-- (must be a valid HTTP request method)
	returnHttpErrorsAsData = true,		-- optional
	timeout = (# seconds),				-- optional
	manuallyManageCookies = false,		-- optional
	followRedirects = true,				-- optional
	requestPriority = (number 1-5) -- optional, The priority of the HTTP request where
		1 == highest priority and 5 == lowest priority.
	bypassRateLimiter = true/false, -- optional. If true, the request will bypass the rate limiter.
	notifyWhenHttpRequestIsInProgress = true/false, -- optional. If true, the resulting observable
		will fire with 'requestInProgress' once the HTTP request is initiated.
}

local urlDataObservable = WFRxHttpClient.get( url, options, sequencing, progressReporter )
	:subscribe( function( data, headers )

			-- The observable sends a single onNext notification when:

				-- (1) the HTTP response is valid and has a non-error (<300)
					-- HTTP status code, or
				-- (2) if the server returns an HTTP error code when
					-- returnHttpErrorsAsData = true.

			-- The data in this case is a KSData with the complete response body.
			-- and 'header' is	a Lua table of key/value pairs.
			-- A special key 'status_code' in the table contains a string
			-- with the HTTP response code (such as "200").

			-- The default HTTP request method is GET, but you can override this
			-- with options.requestMethod.

	end,
	function ( errorMessage )

			-- The observable sends an error notification when:

				-- (1) the system does not receive a valid HTTP response
					-- (for example, if the connection was refused or timed out), or
				-- (2) a valid HTTP response is received but the HTTP status
					-- code indicates an error (>= 300).

			-- You can override this behavior with returnHttpErrorsAsData = true.

	end )
)
\endcode

	This returns an observable. The HTTP request is canceled if the subscription to
	it is disposed.

	NOTE: This function does no caching. Calling this function always
	creates network traffic.
]]--

function WFRxHttpClient.get( url, options, sequencing, progressReporter )

	check( 'string', '?table', '?string', '?WFProgressReporter' )

	httpClientLog:debugf( "get( url = %s, options = %s )", url, options )

	return getOrPut( url, options, nil, progressReporter )

end

--------------------------------------------------------------------------------
--[[!
	\brief Starts an HTTP PUT request and returns the response as an Observable.

	\param url (string) the URL to fetch
	\param data (\c KSData|number|string|table) data to send with the request. Note,
		a string may be interpreted as a file path if the dataFormat option
		is set to "filePath".
	\param options (optional, table) options to modify the request
	\param progressReporter (optional, WFProgressReporter) An instance of 
			WFProgressReporter that is used to communicate transfer progress.

	\return <code>Observable< data, headers ></code> An observable that resolves to the
		response from HTTP server

	For now, this only supports URL encoding. Could easily be adapted for MIME
	encoding as well, but this hasn't proven necessary as yet.

	\b Usage:
\code
local options = {
	requestHeaders = {
		{ header = 'Content-Type', value = 'application/json' },
		{ header = 'Accept', value = 'application/json' },
	}
	requestMethod = 'PUT',				-- optional
				-- (must be a valid HTTP request method)
	timeout = (# seconds),				-- optional
	returnHttpErrorsAsData = true,		-- optional
	manuallyManageCookies = false,		-- optional
	followRedirects = true,				-- optional
	dataFormat = "data|filePath",
		-- optional, specify filePath to interpret data as a file path
		-- NOTE: caller must ensure file exists until the result resolves
		-- default value, "data" interprets data as bytes to send
	offset = 123, -- optional, start the upload from this byte offset
	length = 234, -- optional, only send this many bytes
	requestPriority = (number 1-5) -- optional, The priority of the HTTP request where
		1 == highest priority and 5 == lowest priority.
	bypassRateLimiter = true/false -- optional. If true, the request will bypass the rate limiter.
	notifyWhenHttpRequestIsInProgress = true/false, -- optional. If true, the resulting observable
		will fire with 'requestInProgress' once the HTTP request is initiated.
	multipartRequest = true/false -- optional. If true, the request is treated as a multipart
		request, and the appropriate header is attached.

}

local urlData = WFRxHttpClient.put( url, data, options, progressReporter )

	:subscribe( function( data, headers )

			-- The observable sends a single onNext notification when:

				-- (1) the HTTP response is valid and has a non-error (<300)
					-- HTTP status code, or
				-- (2) if the server returns an HTTP error code when
					-- returnHttpErrorsAsData = true.

			-- The data in this case is a KSData with the complete response body.
			-- and 'header' is	a Lua table of key/value pairs.
			-- A special key 'status_code' in the table contains a string
			-- with the HTTP response code (such as "200").

			-- The default HTTP request method is GET, but you can override this
			-- with options.requestMethod.

	end,
	function ( errorMessage )

			-- The observable sends an error notification when:

				-- (1) the system does not receive a valid HTTP response
					-- (for example, if the connection was refused or timed out), or
				-- (2) a valid HTTP response is received but the HTTP status
					-- code indicates an error (>= 300).

			-- You can override this behavior with returnHttpErrorsAsData = true.

	end )
\endcode

	This returns an observable. The HTTP request is canceled if the subscription to
	it is disposed.

]]--

function WFRxHttpClient.put( url, data, options, progressReporter )

	WFArguments.check( 'string', '?KSData|string|number|table', '?table', '?WFProgressReporter' )

	options = options or {}

--	httpClientLog:debugf( "put( url = %s, data = %s, options = %s )", url, tostring( data ), options )
		-- occasionally useful, but gets REALLY chatty in Xcode
	httpClientLog:debugf( "put( url = %s, options = %s )", url, options )

	options.requestMethod = options.requestMethod or 'PUT'

	return getOrPut( url, options, data, progressReporter )

end

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

function WFRxHttpClient.setProxyCredentials( user, password )

	httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

	return httpClientImp:request( 'setProxyCredentials', user, password )
		:call( runUntilDone )

end

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

function WFRxHttpClient.watchHttpErrors()

	httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

	return httpClientImp:request( 'watchHttpErrors' )

end

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

function WFRxHttpClient.setUserAgentAppName( appName )

	httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

	return httpClientImp:request( 'setUserAgentAppName', appName )
		:call( runUntilDone )

end

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

if WFPlatformInfo.isDesktop then

	function WFRxHttpClient.activeRequestCount()

		httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

		return httpClientImp:request( 'activeRequestCount' )

	end

	function WFRxHttpClient.stopAllRequestsAndWait( waitDurationInSeconds )

		httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

		return httpClientImp
				:request( 
					'stopAllRequestsAndWait', 
						waitDurationInSeconds )

	end

end

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

function WFRxHttpClient.responseMatchesContentType( headers, expectedContentType )

	local contentType = KSTableUtils.derefOpt( headers, 'Content-Type' )

	if contentType then
		contentType = KSStringUtils.getBaseContentType( contentType )
		return contentType == expectedContentType
	end

	return false

end

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

function WFRxHttpClient.responseContentTypeIsJson( headers )

	return WFRxHttpClient.responseMatchesContentType( headers, "application/json" )

end

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

function WFRxHttpClient.extractHttpTimingForPerformanceReport( eventName, logs, primaryValue, headers )

	if eventName == 'onNext' then
		logs[ 'http_time' ] = headers[ 'wf_time' ]
	end
	
	return logs
	
end

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

function WFRxHttpClient.DEBUG_activityStream( refreshSeconds )

	httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

	return httpClientImp:request( 'DEBUG_activityStream', refreshSeconds )

end

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

function WFRxHttpClient.DEBUG_historyStream( refreshSeconds )

	httpClientImp = httpClientImp or WFWorkerThread( 'WFRxHttpClientImp.lua' )

	return httpClientImp:request( 'DEBUG_historyStream', refreshSeconds )

end

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

return WFRxHttpClient
