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

WFRxArrayDiffer.lua

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

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

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

WFRxArrayDiffer consumes a sequence of array-valued tables and reports the differences
in set membership from each value to the next.

For instance:

	local logger = WFLogger( 'sample' )
		:setLevel( 'debug' )

	local source = Rx.Subject()

	source
		:call( WFRxArrayDiffer )
		:subscribe(
			function( additions, removals )
				logger:debugf( "\n>> additions = %s\n>> removals = %s\n",
							   additions, removals )
			end,
			function( message )
				logger:errorf( "\n>> ERROR: %s\n\n", message )
			end )

	source:onNext( {} )

	-- Nothing happens.

	source:onNext( { 1, 3, 5, 7 } )

	>> additions = { 1, 3, 5, 7 }
	>> removals = {}

	source:onNext( { 1, 4, 7, 11 } )

	>> additions = { 4, 11 }
	>> removals = { 3, 5 }
		-- Items appear in order of appearance in either
		-- the current or previous array.

	source:onNext( "not a table" )

	>> ERROR: (message TBD)

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

	-- RxLua
local Rx = require 'Rx'
local CompositeDisposable = Rx.CompositeDisposable
local Disposable = Rx.Disposable
local ObservableBase = Rx.ObservableBase
local Observer = Rx.Observer


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

local ArrayDiffObserver = Rx.Class 'Rx.internals.ArrayDiffObserver' {

	Disposable,
	Observer,
		-- superclasses

	_observer = false,
		-- (Observer)

	_source = false,
		-- (Observable)

	_contentsAsSet = false,
		-- (table) contents of array expressed as a set

	_contentsAsArray = false,
		-- (table) contents of array from last received observation

	_isStopped = false,
		-- (boolean)

	__init = function( self, observer, source )

		self._observer = observer
		self._source = source
		self._isStopped = false

		self._contentsAsSet = {}
		self._contentsAsArray = {}

		return function()
			self._isStopped = true
		end

	end,

	onNext = function( self, newValue )

		if not self._isStopped then

			local function calculateDiffs( newValue )

				if type( newValue ) ~= 'table' then
					error "WFRxArrayDiffer: values must be of type table"
				end

				local oldContentsAsSet = self._contentsAsSet
				local newContentsAsSet = {}
				local newContentsAsArray = {}

				local additions = {}
				local nAdditions = 0

				for i, item in ipairs( newValue ) do

					newContentsAsSet[ item ] = true
					newContentsAsArray[ i ] = item

					if not oldContentsAsSet[ item ] then
						nAdditions = nAdditions + 1
						additions[ nAdditions ] = item
					end

				end

				local removals = {}
				local nRemovals = 0

				for _, item in ipairs( self._contentsAsArray ) do

					if not newContentsAsSet[ item ] then
						nRemovals = nRemovals + 1
						removals[ nRemovals ] = item
					end

				end

				self._contentsAsSet = newContentsAsSet
				self._contentsAsArray = newContentsAsArray

				return additions, removals

			end

			local function handlePredicateResult( success, additions, removals )
				if success then
					if #additions > 0 or #removals > 0 then
						self._observer:onNext( additions, removals )
					end
				else
					self._observer:onError( additions )
						-- In this case, it will be an error message.
				end
			end

			handlePredicateResult( pcall( calculateDiffs, newValue ) )

		end

	end,

	onError = function( self, ... )

		if not self._isStopped then
			self._isStopped = true
			self._observer:onError( ... )
		end

	end,

	onCompleted = function( self )

		if not self._isStopped then
			self._isStopped = true
			self._observer:onCompleted()
		end

	end,

	fail = function( self, e ) error( "ArrayDiffObserver fail called, update code in ArrayDiffObserver" ) end
		-- Not possible to reach the above line because only AutoDetachObserver
		-- has its fail method called.

}

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

local ArrayDiffObservable = Rx.Class 'WF.internals.ArrayDiffObservable' {

	ObservableBase,
		-- superclass

	_source = false,
		-- (Observable)

	__init = function( self, source )
		self._source = source
	end,

	subscribeCore = function( self, observer )

		local arrayDiffObserver = ArrayDiffObserver( observer, self )

		local success, subscriptionOrError = pcall( self._source.subscribe,
													self._source,
													arrayDiffObserver )

		if success then
			return CompositeDisposable( subscriptionOrError, arrayDiffObserver )
		else
			ArrayDiffObserver:dispose()
			error( subscriptionOrError )
		end

	end,

}

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

return function( self )

	return ArrayDiffObservable( self )

end
