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

WFRxThrottledBatchedBulkDocValue.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2018 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.utils
local KSDate = require "KSDate"
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFFunctionUtils = require 'WFFunctionUtils'

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

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

local check = assert( WFArguments.check )

local WFFunctionUtils_reduce = WFFunctionUtils.reduce

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

local function addToChunkedArray( chunks, val, chunkSize )

	local currChunk = chunks[ #chunks ]

	local currChunkSize = #currChunk
	
	if currChunkSize > 0 and currChunkSize  % chunkSize == 0 then

		currChunk = {}

		chunks[ #chunks + 1 ] = currChunk

	end
		
	currChunk[ #currChunk + 1 ] = val

	return chunks

end

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

local function chunkedArray( arr, chunkSize )

	return WFFunctionUtils_reduce( { {} }, addToChunkedArray, arr, chunkSize )

end

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

local _lastBulkDocValueDeliveryTimes = KSTableUtils.makeWithMode( "kv" )

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

--[[!

	\brief A function used to batch/throttle calls to the bulkDocValue doc 
		store API.
	
	\param desc (string) description of operation to use for debugging and diagnostic logs
	\param docStore (docstore) A valid WFRxDocStore.
	\param assetIds (table) A table of asset ids.
		Optional arguments to this api are passed as named parameters in the
		assetIds table they are:
		* \c label (optional, string): Specifies what label-ed revision to
			get for the documents.
		* \c includeInitialChangeCounter (optional, boolean): if true, return the first
			changeCounter (sequence number) related to the documents.
	\param chunkSize (number) The batch size. AKA the number of doc
		values to fetch for each call to bulkDocValue.
	\param callsPerSecondLimit (number) The max number of API
		calls that we'll make to the doc store per second.
		
	\return <code>Observable</code> An observable that fires with batches of
		of doc values, as they're available.
		
--]]

return function( desc, docStore, assetIds, chunkSize, callsPerSecondLimit )

	check( 'string', 'WFRxDocStore', 'table', 'number', 'number' )

	local minWaitUpperBound = 1 / callsPerSecondLimit

	local chunkedAssetIds = chunkedArray( assetIds, chunkSize )

	local scheduler = WFRxScheduler.sharedScheduler()

	-- this uses :flatMerge( 1 ) to pull a new assetIdsChunk from
	-- chunkedAssetIds only after the previous 
	-- docStore:bulkDocValue( chunkedAssetIds ) completes. Coupled with the
	-- :delay() call, this effectively rate limits the :bulkDocValue() calls.
	
	return Rx.Observable.from( chunkedAssetIds )
	
		:map( function( assetIdsChunk )

			return Rx.Observable.defer( function()

				local lastDeliveredTime =
						_lastBulkDocValueDeliveryTimes[ docStore ]
					
				local delayInSecs =
						not lastDeliveredTime and
							0 or
							math.max(
								minWaitUpperBound -
									( KSDate.timeIntervalSinceAppLaunch() -
										lastDeliveredTime ),
								0 )

				-- pass along the optional arguments to bulkDocValue

				assetIdsChunk.label = assetIds.label
				assetIdsChunk.includeInitialChangeCounter = assetIds.includeInitialChangeCounter

				return docStore 
					:bulkDocValue( assetIdsChunk, nil, desc )
					:delay( delayInSecs, scheduler )
					:tap( function()

						_lastBulkDocValueDeliveryTimes[ docStore ] =
							KSDate.timeIntervalSinceAppLaunch()

					end )

			end )

		end )
		
		:flatMerge( 1 )

end