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

ADOBE CONFIDENTIAL
------------------
Copyright 2016 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 WFCloseSupport = require 'WFCloseSupport'

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

-- @mark WFRxReplaySubjectWithCloseSupport namespace

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

local checkDisposed = Rx.Disposable.checkDisposed

local WFCloseSupport_hasBeenOrWillBeClosed = WFCloseSupport.hasBeenOrWillBeClosed

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

local replaySubjectWithCloseSupportMethods = {

	Rx.ReplaySubject
}

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

local function tableShallowCopy(orig)

	local copy = {}

	for orig_key, orig_value in pairs(orig) do
		copy[orig_key] = orig_value
	end

	return copy

end

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

function replaySubjectWithCloseSupportMethods:_enqueueValue(value, now)

	self._qHead = self:_incQueueIndex(self._qHead)

	local headValue = self._q[self._qHead]
	if headValue then
	
		local obj = headValue.value[ 1 ]
		local newObj = value and value[ 1 ]
		
		if type( obj ) == 'table' then
		
			local closeFn = obj.close
			
			if type( closeFn ) == 'function' and obj ~= newObj
				and not WFCloseSupport_hasBeenOrWillBeClosed( obj ) then
				
				obj:close()
				
			end
			
		end
		
	end
	
	self._q[self._qHead] = { interval = now, value = value }

	if not self._qTail then

		self._qTail = 1

	elseif self._qTail == self._qHead then

		self._qTail = self:_incQueueIndex(self._qTail)
		-- queue was full, move the tail as we overwrote a value
	end
	
end

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

function replaySubjectWithCloseSupportMethods:onNext(...)

	checkDisposed(self)

	if not self._isStopped then

		local value = { n = select('#', ...), ... }
		
		assert( value.n == 1, 'Only single value supported' )

		self:_enqueueValue(value, self._scheduler:now())
		self:_trimQueuedValues(self._scheduler:now())

		local os = tableShallowCopy(self._observers)

		for i = 1, #os do

			local observer = os[i]

			observer:onNext(...)

			observer:ensureActive()

		end

	end

end

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

local WFRxReplaySubjectWithCloseSupport = Rx.Class( 

	'WFRxReplaySubjectWithCloseSupport', 
	
	replaySubjectWithCloseSupportMethods 
		
)

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

return function( source, bufferSize )

	assert( 
		bufferSize == nil, 
		"As opposed to the standard :shareReplay() we enforce a buffer size of 1" 
	)
	
	return source:multicast(function()
		return WFRxReplaySubjectWithCloseSupport( 1 )
	end)
	:refCount()
	
end
