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

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 WFArguments = require 'WFArguments'

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

local Observable = Rx.Observable
local SchedulerArgument = Rx.internals.SchedulerArgument
local Subject = Rx.Subject

local notificationCreateOnCompleted = Rx.Notification.createOnCompleted
local notificationCreateOnError = Rx.Notification.createOnError
local notificationCreateOnNext = Rx.Notification.createOnNext

local disposableCreate = Rx.Disposable.create

local check = WFArguments.check

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

--[[

	WFRxControlledObservableWithBuffer diverges from the standard 
	Rx.Observable:controlled() operator in that it maintains a bounded queue as 
	opposed to the unbounded one in the :controlled() operator.
	However onError() and onCompleted() notifications are not counting to the
	buffer size, but are getting still queued up.
	
]]

local ControlledSubject = Rx.Class 'WFRxControlledObservableWithBuffer.ControlledSubject' {

	Subject,
		-- superclass

	_bufferSize = false,
		-- (number)

	_queue = false,
		-- (table)
		
	_subject = false,
		-- (Subject)

	_requestedCount = false,
		-- (number)

	_requestedDisposable = false,
		-- (Disposable)

	_error = false,
		-- (table)

	_hasFailed = false,
		-- (boolean)

	_hasCompleted = false,
		-- (boolean)

	_scheduler = false,
		-- (Scheduler)

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

	__init = function( self, bufferSize, requestCount, scheduler )

		self._bufferSize = bufferSize
		self._queue = {}
		self._subject = Subject()
		self._requestedCount = requestCount or math.huge
		self._requestedDisposable = nil
		self._error = nil
		self._scheduler = scheduler

	end,

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

	_subscribe = function( self, observer )
	
		return self._subject:subscribe( observer )
		
	end,
	
	--------------------------------------------------------------------------------

	_reduceQueue = function( self )
	
		if self._bufferSize <= 0 then
			return
		end
		
		while #self._queue >= self._bufferSize  do
			table.remove( self._queue, 1 )
		end
		
	end,

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

	onCompleted = function( self )

		self._hasCompleted = true

		if #self._queue == 0 then

			self._subject:onCompleted()
			self:disposeCurrentRequest()

		else

			table.insert( self._queue, notificationCreateOnCompleted() )

		end

	end,

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

	onError = function( self, ... )

		self._hasFailed = true

		self._error = { n = select( '#', ... ), ... }

		if #self._queue == 0 then

			self._subject:onError( ... )
			self:disposeCurrentRequest()

		else
			
			table.insert( self._queue, notificationCreateOnError(...) )

		end

	end,

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

	onNext = function( self, ... )

		if self._requestedCount <= 0 then

			self:_reduceQueue()
			
			table.insert( self._queue, notificationCreateOnNext(...) )
			
		else

			if self._requestedCount == 1 then
				self:disposeCurrentRequest()
			end

			self._requestedCount = self._requestedCount - 1

			self._subject:onNext( ... )

		end

	end,

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

	_processRequest = function( self, numberOfItems )

		while #self._queue > 0 and ( numberOfItems > 0
			or self._queue[ 1 ]:kind() ~= 'N' ) do
		
			local first = table.remove( self._queue, 1 )
			first:accept( self._subject )
		
			if first:kind() == 'N' then
				numberOfItems = numberOfItems - 1
			else
				self:disposeCurrentRequest()
				self._queue = {}
			end
		
		end

		return numberOfItems

	end,

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

	request = function( self, number )

		self:disposeCurrentRequest()

		self._requestedDisposable = self._scheduler:schedule(number,
			function(_, i)

				local numberOfItemsRemaining = self:_processRequest( i )

				local stopped = self._hasCompleted or self._hasFailed

				if not stopped and numberOfItemsRemaining > 0 then

					self._requestedCount = numberOfItemsRemaining

					return disposableCreate( function()
						self._requestedCount = 0
					end )
						-- Scheduled item is still in progress. Return a new
						-- disposable to allow the request to be interrupted
						-- via dispose.

				end

			end)

		return self._requestedDisposable

	end,

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

	disposeCurrentRequest = function( self )

		if self._requestedDisposable then

			self._requestedDisposable:dispose()
			self._requestedDisposable = nil

		end

	end,
	
	--------------------------------------------------------------------------------

	dispose = function( self )
	
		self:disposeCurrentRequest()
		
		self:Rx_Subject_dispose()
		
	end

}

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

local ControlledObservable = Rx.Class 'WFRxControlledObservableWithBuffer' {

	Observable,
		-- superclass
		
	_source = false,
		-- (Observable)

	_subject = false,
		-- (ControlledSubject)

	_multicast = false,
		-- (ConnectableObservable)
		
	----------------------------------------------------------------------------

	__init = function( self, source, bufferSize, requestCount, scheduler )

		self._subject = ControlledSubject( bufferSize, requestCount, scheduler )

		local multicast = source:multicast( self._subject )
		self._multicast = multicast
		self._source = multicast:refCount()
		
	end,

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

	_subscribe = function( self, observer )

		return self._source:subscribe( observer )

	end,

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

	request = function(self, numberOfItems)

		return self._subject:request( numberOfItems or -1 )

	end,
	
}

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

return function( self, bufferSize, requestCount, scheduler )

	check( 'table', 'number', 'number|nil', 'table|nil' )
	
	scheduler = SchedulerArgument.validateWithDefault( 'Rx.Observable:controlled',
													  'currentThread', scheduler )

	assert( bufferSize > 0, "WFRxControlledObservableWithBuffer: bufferSize has to be greater then zero" )
	
	return ControlledObservable( self, bufferSize, requestCount, scheduler )

end
