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

WFCatModelForwarderTest.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains 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 are protected by all applicable
intellectual property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

local Rx = require 'Rx'

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

	-- WF.oz_client.managed_catalog.model_forwarding
local WFCatModelForwarder = require 'WFCatModelForwarder'
local WFLogger = require 'WFLogger'
local WFRxScheduler = require 'WFRxScheduler'

local WFRxTestCase = require 'WFRxTestCase'
local WFRxModelsTest_Server = require 'WFRxModelsTest_Server'

local Observable = Rx.Observable

local WFCatModelForwarderTest = {}

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

local logger = WFLogger 'WFCatModelForwarderTest'

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

local function modelWayBackImp( a, b, deliverAllValues )

	local eventTracker = Rx.BehaviorSubject( nil )

	return {

		deliverAllValues = deliverAllValues,

		value = {
			constant = 1,
			behavior = Observable.timer( 0, 0.02, WFRxScheduler.sharedScheduler() )
				:take( 2 )
				:map( function( n )
					if n == 1 then
						return 'not yet'
					else
						return 'okay'
					end
				end ),
			a = a,
			b = b,
			lastEvent = eventTracker,
		},

		models = {

			fooModel = function( self, a, b )
				return WFCatModelForwarderTest.modelWayBack( a, b )
			end,

			streamOfValuesModel = function( self, initialValue, laterValues )
				return {
					deliverAllValues = deliverAllValues,
					-- deliver initial value followed by the later values in "bursty" mode after a delay
					value = Rx.Observable.concat( 
								Rx.Observable.just( initialValue ),
								Rx.Observable.from( laterValues )
									:delay( 0.1, WFRxScheduler.sharedScheduler() ) )
				}
			end,

			errorModel = function( self, a, b )
				return {
					value = Rx.Observable.throw( 'expected error' ),
				}
			end,

			sayGoodnightGracie = function( self )
				return {
					value = {
						gracieSays = "Goodnight, Gracie",
					}
				}
			end,

			loudElidedStream = WFRxModelsTest_Server.openLoudElidedStream,
			loudNonElidedStream = WFRxModelsTest_Server.openLoudNonElidedStream,

		},

		events = {
			fooEvent = function( self, ... )
				eventTracker:onNext( ... )
			end,
		},
	}

end

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

function WFCatModelForwarderTest.modelWayBack( a, b )

	return modelWayBackImp( a, b, false )

end

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

function WFCatModelForwarderTest.modelWayBack_deliverAllValues( a, b )

	return modelWayBackImp( a, b, true )

end

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

local defaultDoLater = function()
	return Observable.timer( 0.001, WFRxScheduler.sharedScheduler() )
end
local later = defaultDoLater

local function runCommand( params )
	return later()
	:flatMapLatest( function()
		return Observable.create( function( observer )
			local daemonthread = WFWorkerThread( "WFCatModelForwarderTest.lua :: wayback" )
			local result = daemonthread:request( 'runCommand', params )
			observer:onNext( result )
			return function()
				daemonthread:close()
			end
		end )
		:take( 1 )
		:switch()
	end )
end

local callSubmodel

function WFCatModelForwarderTest.initMessageProcessor( name )

	logger:infof( "initing message processor: %s", name )

	if name == 'wayback' then
		WFRequestDispatcher.onMessage( 'runCommand', function( params )
			return require( params.n )[ params.c ]( unpack( params.a or {}, 1, params['#'] ) )
		end )
	end

	if name == 'forwarder' then
		WFRequestDispatcher.onMessage( 'callSubmodel', function( ... )
			return callSubmodel( ... ).value
		end )
		WFRequestDispatcher.onMessage( 'rebootDaemon', function( params )
			return WFCatModelForwarder.debug_simulateReboot( runCommand )
		end )
		local nextSlowOneHasBeenQueued
		local timeToRunSlowCommand

		WFRequestDispatcher.onMessage( 'makeNextRunCommandSlowToRespond', function( params )

			assert( later == defaultDoLater, "makeNextRunCommandSlowToRespond must be followed by a call to finallyRunThatSlowCommand" )
			timeToRunSlowCommand = Rx.Subject()
			nextSlowOneHasBeenQueued = Rx.Subject()

			later = function()

				nextSlowOneHasBeenQueued:onCompleted( )
				later = defaultDoLater
				return timeToRunSlowCommand

			end
			return true

		end )

		WFRequestDispatcher.onMessage( 'waitForSlowCommandToQueueUp', function( params )
			return nextSlowOneHasBeenQueued
		end )

		WFRequestDispatcher.onMessage( 'finallyRunThatSlowCommand', function( params )
			assert( nextSlowOneHasBeenQueued, "finallyRunThatSlowCommand may only be called after makeNextRunCommandSlowToRespond" )
			timeToRunSlowCommand:onNext( true )
			timeToRunSlowCommand:onCompleted()
			timeToRunSlowCommand = nil
			nextSlowOneHasBeenQueued = nil
			return true
		end )
	end

	WFRequestDispatcher.onMessage( 'DEBUG_dumpCoverageP', function( path )
		require 'SetupCoverage'.dumpCoverage( path )
		return require 'Rx'.Observable.just( true )
	end )

end

function WFCatModelForwarderTest.forwardedModel( namespace, func )

	local model
	model, callSubmodel = WFCatModelForwarder.makeForwardedModel( runCommand, namespace, func )

	return model

end

local kCoveragePath

function WFCatModelForwarderTest.setUpClass( coveragePath )
	kCoveragePath = coveragePath
end

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

local function rxTestBasicImp( rxTest, deliverAllValues, rebootBetweenTests )

	rxTest:setTimeLimit( 10 )

	local WFTestModelClient = require 'WFTestModelClient'

	local thread = WFWorkerThread( "WFCatModelForwarderTest.lua :: forwarder" )
	local daemonthread = WFWorkerThread( "WFCatModelForwarderTest.lua :: wayback" )

	local factory = WFTestModelClient.makeModelFactory( thread )

	local model = factory:openModel( "WFCatModelForwarderTest.forwardedModel",
									 'WFCatModelForwarderTest',
									 deliverAllValues
									 	and 'modelWayBack_deliverAllValues'
										or 'modelWayBack' )

	local submodel

	local value = model:value()
	local observedValue = rxTest:observe( value )

	rxTest:finally( function()
		model:close()
		thread:close()
		daemonthread:close()
	end )

	observedValue:expectDataWithinTime( 5 ) {
			behavior = 'okay',
			constant = 1,
		}

	local function testEvent()

		model:postEvent( 'fooEvent', "Hi Last Event" )

		observedValue:expectDataWithinTime( 5 ) {
			behavior = 'okay',
			constant = 1,
			lastEvent = 'Hi Last Event',
		}

	end

	local function testSubModel()

		submodel = model:openModel( 'fooModel', "A", 'b' )
		local observedSubValue = rxTest:observe( submodel:value() )

		observedSubValue:expectDataWithinTime( 5 ) {
			constant = 1,
			behavior = 'okay',
			a = 'A',
			b = 'b',
		}

		submodel:close()
		submodel = nil

	end

	local function testStreamOfValuesModel()

		local initialValue = 'uno'

		local laterValues = { 'zwei', 'three' }

		submodel = model:openModel( 'streamOfValuesModel', initialValue, laterValues )
		local observedSubValue = 
				rxTest:observe( 
					submodel
						:value() 
						:map(
							function( v )

								if v == initialValue then

									rxTest:logf( "begin busy wait on main thread%s observable after initial value '%s' to cause backlog on client main thread",
													deliverAllValues and " deliverAllValues" or "", initialValue )

									local i = 1
									while i <  1000000 do
										i = i + 1
									end 

								end

								return v

							end )

					)

		rxTest:logf( "streamOfValuesModel with%s deliverAllValues checking initial value == %s",
						deliverAllValues and "" or " OUT", initialValue )

		observedSubValue:expectDataWithinTime( 5, initialValue )

		local expectedValues = deliverAllValues and laterValues or { laterValues[ #laterValues ] }

		for i, expectedValue in ipairs( expectedValues ) do

			rxTest:logf( "streamOfValuesModel with%s deliverAllValues checking expected value %s == %s",
							deliverAllValues and "" or " OUT", i, expectedValue )

			observedSubValue:expectDataWithinTime( 5, expectedValue )

		end

		observedSubValue:expectCompleted()

		submodel:close()
		submodel = nil

	end

	local function testErrorModel()

		submodel = model:openModel( 'errorModel' )
		local observedSubValue = rxTest:observe( submodel:value() )

		observedSubValue:expectError( 'expected error' )

		submodel:close()
		submodel = nil

	end

	local function testSubModelStream()

		local observedSubValue = rxTest:observe( thread:request( 'callSubmodel', 'sayGoodnightGracie' ) )

		observedSubValue:expectDataWithinTime( 5 ) {
			gracieSays = "Goodnight, Gracie",
		}

	end

	local function testElidedObservableStream()

		submodel = model:openModel( 'loudElidedStream' )
		local smValue = rxTest:observe( submodel:value() )

		submodel:postEvent( 'postSomeDataToStream' )
		smValue:expectDataWithinTime( 3 ) { streamValue = 3 }

		submodel:postEvent( 'closeSubject' )
		smValue:expectCompleted()
	
	end
	
	local function testLoudObservableStream()
	
		submodel = model:openModel( 'loudNonElidedStream' )
		local smValue = rxTest:observe( submodel:value() )

		submodel:postEvent( 'postSomeDataToStream' )
		smValue:expectData{ streamValue = 1 }
		smValue:expectData{ streamValue = 2 }
		smValue:expectData{ streamValue = 3 }

		submodel:postEvent( 'closeSubject' )
		smValue:expectCompleted()
	
	end

	local function testFastModelClose()

		rxTest:expectSingleValue(
			thread:request( 'makeNextRunCommandSlowToRespond' ),
			true )

		submodel = model:openModel( 'fooModel' )

		rxTest:expectOnlyCompleted(
			thread:request( 'waitForSlowCommandToQueueUp' )
		)

		submodel:close()

		rxTest:expectSingleValue(
			thread:request( 'finallyRunThatSlowCommand' ),
				-- this simply tests that when runCommand returns a result
				-- after the interest in the model has been lost that no error
				-- is generated.
			true
		)

	end

	local function rebootDaemon()

		rxTest:expectOnlyCompleted(
			thread:request( 'rebootDaemon' )
		)

	end

	local testSequence = {
		testEvent,
		testSubModel,
		testStreamOfValuesModel,
		testErrorModel,
		testSubModelStream,
		testElidedObservableStream,
		testLoudObservableStream,
		testFastModelClose,
	}

	for i, f in ipairs( testSequence ) do
		if i ~= 1 and rebootBetweenTests then

			-- Repeat all the above tests with simulating the daemon having been relaunched
			-- between the call.

			rebootDaemon()
		end
		f()
	end

		-- ALL DONE.  Now collect coverage data, too.
	if kCoveragePath then
		rxTest:expectSingleValue(
			thread:request( 'DEBUG_dumpCoverageP', kCoveragePath ),
			':ignore:' )
		rxTest:expectSingleValue(
			daemonthread:request( 'DEBUG_dumpCoverageP', kCoveragePath ),
			':ignore:' )
	end

end

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

function WFCatModelForwarderTest.rxTestBasic( rxTest )

	rxTestBasicImp( rxTest, false, false )

end

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

function WFCatModelForwarderTest.rxTestBasicWithReboot( rxTest )

	rxTestBasicImp( rxTest, false, true )

end

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

function WFCatModelForwarderTest.rxTestBasic_deliverAllValues( rxTest )

	rxTestBasicImp( rxTest, true, false )

end

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

function WFCatModelForwarderTest.rxTestBasic_deliverAllValuesWithReboot( rxTest )

	rxTestBasicImp( rxTest, true, true )

end

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

local RxTestHelper = {

	_rxTest = false,

	_observer = false,
		-- rxTest observer

	_weakValue = false,
		-- fully weak table to hold value

	Rx.Observer,

}

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

function RxTestHelper:__init( rxTest, model )

	self._rxTest = rxTest
	self._weakValue = setmetatable( {}, { __mode = 'kv' } )

	local value= model:value()
	self._weakValue[ value ] = value
	self._observer = rxTest:observe( value )

end

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

function RxTestHelper:checkError( name, expected )

	logger:debugf( "Checking error value (%s) is %s", name, expected )
	self._observer:expectError( expected )
	logger:debugf( "Got my error value %s!", name )

end

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

function RxTestHelper:checkNext( name, expected )

	logger:debugf( "Checking next value (%s) is %s", name, expected )
	self._observer:expectDataWithinTime( 5, expected )
	logger:debugf( "Got my next value %s!", name )

end

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

function RxTestHelper:checkCompleted( name )

	logger:debugf( "Checking completed (%s)", name )
	self._observer:expectCompleted( )
	logger:debugf( "(%s) was completed" )

end

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

function RxTestHelper:dispose()

	self._observer:dispose()

	collectgarbage( 'collect' )
	collectgarbage( 'collect' )
	collectgarbage( 'collect' )

	self._rxTest:expectEqual( nil, next( self._weakValue ), "Value GCs when subscription ended" )

end

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

--[[
	\return (boolean) if object was GC'd after extensive retry
--]]
function RxTestHelper:waitForGC( valueFetcher )
	local WFIdle = require 'WFIdle'

	local value = valueFetcher()
	local gcWatcher = setmetatable( { [ value ] = value }, { __mode = 'kv' } )
	value = nil

	local i = 10
	local function try()

		-- The model's original value might have already been received.
		-- the scheduler queue will have a strong link to the value
		-- and thus to the model.  so we have to wait for the scheduler
		-- to process that.

		collectgarbage( 'collect' )
		collectgarbage( 'collect' )
		collectgarbage( 'collect' )

		return not not next( gcWatcher )

	end

	local wait = Rx.Subject()

	local retry
	retry = function()
		WFIdle.onIdleDo( function()
			if try() then
				i = i - 1
				if i > 0 then
					retry()
				else
					wait:onCompleted()
				end
			else
				wait:onCompleted()
			end
		end )
	end

	retry()

	self._rxTest:expectOnlyCompleted( wait )

	if next( gcWatcher ) then
		return false
	end
	return true
end

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

function RxTestHelper:checkTrue( value, details )
	return self._rxTest:expectEqual( true, value, details )
end

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

function RxTestHelper:checkFalse( value, details )
	return self._rxTest:expectEqual( false, value, details )
end

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

RxTestHelper = Rx.Class( 'RxTestHelper', RxTestHelper )

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

function WFCatModelForwarderTest.rxTestTestsOnRootModel( rxTest )

	local AsyncTestHelper = function( model )
		return RxTestHelper( rxTest, model )
	end

	local WFTestModelClient = require 'WFTestModelClient'

	local thread = WFWorkerThread( "WFCatModelForwarderTest.lua :: forwarder" )
	local daemonthread = WFWorkerThread( "WFCatModelForwarderTest.lua :: wayback" )

	local factory = WFTestModelClient.makeModelFactory( thread )

	rxTest:finally( function()
		daemonthread:close()
		factory:close()
		thread:close()
	end )

	rxTest:expectSingleValue( daemonthread:threadIsReady(), true )

	local model = factory:openModel(
					"WFCatModelForwarderTest.forwardedModel",
					'WFModelsTest_Server',
					'rootModel' )

	WFCatModelForwarderTest.runTestsOnRootModel( AsyncTestHelper, model )

end

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

function WFCatModelForwarderTest.runTestsOnRootModel( AsyncTestHelper, rootModel )

	-- A rather large set of things tested all at once, so that we don't have to 
	-- try and coordinate launching and shutting down the daemon more than once.
	--
	-- [1] This tests that model data values (onNext) are delivered
	-- [2] ... that when the client closes a model the server receives that close
	-- [3] ... that when a model isCompleted it may still make submodels
	-- [4] ... that completed models deliver onCompleted to client
	-- [5] ... that client closing a model disposes the server's subscription
	-- [6] ... that a client letting a model GC closes the server's model
	-- [7] ... that letting the model but not the model value() GC does NOT close
	--            the subscription
	-- [8] ... that errors are terminal and cause the model subscription to dispose

	local rootHelper = AsyncTestHelper( rootModel )

	--------------------------------------------------------------------------------
	-- [1] This tests that model data values (onNext) are delivered
	-- [4] that completed models deliver onCompleted to client
	--------------------------------------------------------------------------------

	rootHelper:checkNext( 'Root model got true', { root = true } )
	rootHelper:checkCompleted( 'Root model completed' )
	rootHelper:dispose()

	--------------------------------------------------------------------------------
	-- [3] that when a model isCompleted it may still make submodels
	--------------------------------------------------------------------------------

	local subModel = rootModel:subModel( 'rxFromTest', { 'basic test' } )
	local helper = AsyncTestHelper( subModel )

	helper:checkNext( 'rxFromTest got basic test', 'basic test' )
	helper:checkCompleted( 'rxFromTest model completed' )
	helper:dispose()
	subModel:close()

	--------------------------------------------------------------------------------
	-- [5] that client closing a model disposes the server's subscription
	--------------------------------------------------------------------------------

	local modelToDispose = rootModel:subModel( 'rxObserveDisposal' )
	subModel = modelToDispose:subModel( 'isParentModelDisposed' )
	helper = AsyncTestHelper( subModel )

	helper:checkNext( 'started out not disposed', 'not disposed' )


	-- We might consider disposing the server side subscription if the client
	-- disposes, but this was deemed too dangerous: If a model is kept around on
	-- the client, and UI is presented, then torn down, then presented again,
	-- (resulting in a subscribe-dispose-subscribe pattern) the second time the UI
	-- wouldn't get data, but would be an error.
	-- modelToDispose:value():subscribe():dispose()

	-- So instead, we close the model, which also will cause the server-side
	-- subscription to be disposed.

	modelToDispose:close()
	modelToDispose = nil

	helper:checkNext( 'disposed after close', 'is disposed' )
	helper:checkCompleted( 'completed after disposed' )

	helper:dispose()
	subModel:close()

	--------------------------------------------------------------------------------
	-- [6] that a client letting a model GC closes the server's model
	--------------------------------------------------------------------------------

	-- purge any garbage; makes it easier to verify __gc metamethod is firing
	-- for the correct model without spurious older models also getting __gc'd
	-- at the same time.

	collectgarbage( 'collect' )
	collectgarbage( 'collect' )
	collectgarbage( 'collect' )

	local modelToGC = rootModel:subModel( 'rxObserveClose' )
	subModel = modelToGC:subModel( 'isParentModelClosed' )
	helper = AsyncTestHelper( subModel )

	helper:checkNext( 'started out not closed', 'not closed' )

	helper:checkTrue( helper:waitForGC( function()
		local temp = modelToGC
		modelToGC = nil
		return temp
	end ), 'Model should GC if no-one is holding value' )

	helper:checkNext( 'closed after gc', 'now it is closed' )
	helper:checkCompleted( 'completed after closed' )

	helper:dispose()
	subModel:close()

	--------------------------------------------------------------------------------
	-- [7] ... that letting the model but not the model value() GC does NOT close
	--------------------------------------------------------------------------------

	modelToGC = rootModel:subModel( 'rxObserveClose' )
	helper = AsyncTestHelper( modelToGC )
	subModel = modelToGC:subModel( 'parentModelChanger' )

	helper:checkNext( 'got model initial value', 'original value' )

	local weak = setmetatable( {}, { __mode = 'kv' } )
	weak[ modelToGC ] = modelToGC

	helper:checkFalse( helper:waitForGC( function()
		local temp = modelToGC
		modelToGC = nil
		return temp
	end ), 'Model should NOT GC because helper is subscribed to model:value()' )

	subModel:subModel( 'changeGrandParentValue', 'still receive values so long as subscribed' )

	helper:checkNext( 'GC not allowed while value() subscribed', 'still receive values so long as subscribed' )

	subModel:subModel( 'changeGrandParentValue', '< special unit test value: completed >' )
	helper:checkCompleted( 'completed after model value completed' )

	modelToGC = next( weak )
		-- Create a strong link to the model again.

	helper:checkTrue( not not modelToGC, "Model should not have been GCd" )

	helper:dispose()
		-- Subscription is now disposed and allowed to GC

	helper:checkTrue( helper:waitForGC( function()
		local temp = modelToGC
		modelToGC = nil
		return temp
	end ), 'Model should GC once value is GCed even if a submodel is still open' )

	subModel:close()

	--------------------------------------------------------------------------------
	-- [8] ... that errors are terminal and cause the model subscription to dispose
	--------------------------------------------------------------------------------

	local errModel = rootModel:subModel( 'rxErrorTester' )
	local errHelper = AsyncTestHelper( errModel )
	subModel = errModel:subModel( 'countSubscriptions' )
	helper = AsyncTestHelper( subModel )

	errHelper:checkError( "Value is an error", "Testing Model Errors" )
	helper:checkNext( "Errors do not resubscribe, nor autoclose", {
		isClosed = false,
		subscriptionCount = 1,
		disposeCount = 1,
	} )

	errHelper:dispose()
	errHelper = nil
	errModel:close()
	errModel = nil

	helper:checkNext( "Models with errors still close", {
		isClosed = true,
		subscriptionCount = 1,
		disposeCount = 1,
	} )

	helper:dispose()
	subModel:close()

	--------------------------------------------------------------------------------
	-- [2] that when the client closes a model the server receives that close
	--------------------------------------------------------------------------------

	subModel = rootModel:subModel( 'rootClosedSubModel' )
	helper = AsyncTestHelper( subModel )

	helper:checkNext( 'started out not disposed', 'root is not closed' )

	rootModel:close()

	helper:checkNext( 'root was closed', 'root was closed' )
	helper:checkCompleted( 'completed after root closed' )

	helper:dispose()
	subModel:close()
end

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

return WFRxTestCase.wrapSuite( WFCatModelForwarderTest )
