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

WFRxEventBridgeTest.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.

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

	-- Rx
local Rx = require 'Rx'
local TestScheduler = Rx.testing.TestScheduler

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

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'


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

local onNext = Rx.testing.ReactiveTest.onNext
local onError = Rx.testing.ReactiveTest.onError
local onCompleted = Rx.testing.ReactiveTest.onCompleted
local subscribe = Rx.testing.ReactiveTest.subscribe

local asKSEvent = WFRxEventBridge


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

local WFRxEventBridgeTest = {}

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

local function rigEvent( test, event )

	local events = {}

	event:ownedOnDataE( test, function( ... )
		events[ #events + 1 ] = { n = select( '#', ... ) , 'data', ... }
	end )

	event:ownedOnErrorE( test, function( msg )
		events[ #events + 1 ] = { 'error', msg }
	end )

	event:ownedOnDoneE( test, function( )
		events[ #events + 1 ] = 'done'
	end )

	return events

end

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

function WFRxEventBridgeTest.testEmpty( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onCompleted( 390 )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 390 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testOneValue( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onNext( 245, "hello", "and", "more" ),
		onCompleted( 390 )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ n = 3, 'data', "hello", "and", "more" },
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 390 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testReceivesMultipleValues( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onNext( 245, "hello" ),
		onNext( 252, "blah" ),
		onNext( 300, "and more" ),
		onNext( 344, "foo" ),
		onCompleted( 390 )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ n = 1, 'data', "hello" },
		{ n = 1, 'data', "blah" },
		{ n = 1, 'data', "and more" },
		{ n = 1, 'data', "foo" },
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 390 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testReceivesErrorAfterFirstValue( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onNext( 245, "hello" ),
		onNext( 252, "blah" ),
		onNext( 300, "and more" ),
		onError( 344, "foo" )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ n = 1, 'data', "hello" },
		{ n = 1, 'data', "blah" },
		{ n = 1, 'data', "and more" },
		{ 'error', "foo" },
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 344 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testOneErrorValue( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onError( 245, "totally failed, dude" )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ 'error', "totally failed, dude" },
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 245 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testErrorValueTuple( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onError( 245, "hello", "and more", 42 )
	)

	local result = xs:call( asKSEvent, scheduler )

	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ 'error', "hello" },
					-- NOTE: KSReactive only supports a single message for
					-- errors!	So the bridge passes in the tuple it
					-- will be trimmed to a single value.
		"done"
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 245 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testSkipsSubscribeWhenForgotten( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onError( 245, "hello", "and more", 42 )
	)

	do
		local e = xs:call( asKSEvent, scheduler )
		e = e and nil
	end

	collectgarbage()
	collectgarbage()
	collectgarbage()

	scheduler:start()

	assert( #xs._subscriptions == 0, 'got no subscription' )

end

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

function WFRxEventBridgeTest.testDisposesWhenForgotten( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onNext( 245, 'message1' ),
		onNext( 250, 'forget' ),
		onNext( 255, 'message2' )
	)

	local owner = {
	}
	require 'WFCloseSupport'.addCloseSupportMethods( owner )
	local e
	e = xs
		:tap( function( message )
			if message == 'forget' then
				e = nil
				owner:close()

				collectgarbage()
				collectgarbage()
				collectgarbage()
			end
		end )
		:call( asKSEvent, scheduler )

	local events = rigEvent( owner, e )

	scheduler:start()

	owner = nil

	assert( KSTableUtils.deepEquals( events, {
		{ n = 1, 'data', "message1" },
	} ), 'expected event stream' )

	assert( #xs._subscriptions == 1, 'got one subscription' )
	assert( xs._subscriptions[ 1 ] == subscribe( 1, 250 ), 'subscription started and ended right' )

end

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

function WFRxEventBridgeTest.testFiresFirstEventForReplay( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onNext( 245, 'first' ),
		onNext( 255, 'second' ),
		onCompleted( 450 )
	)

	local replay = xs:replay( nil, 2 )
	replay:connect()

	local result = replay:call( asKSEvent, scheduler )
	local events = rigEvent( test, result )

	scheduler:start()

	assert( KSTableUtils.deepEquals( events, {
		{ n = 1, 'data', "first" },
		{ n = 1, 'data', "second" },
		'done'
	} ), 'expected event stream' )

end

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

function WFRxEventBridgeTest.testGCSafety( test )

	local scheduler = TestScheduler()

	local xs = scheduler:createHotObservable(
		onCompleted( 390 )
	)

	-- this unit test was added due to an anchoring bug where the event stream
	-- could be collected before it was returned, so run the GC aggressively to
	-- confirm this no longer occurs
	local pause = collectgarbage( "setpause", 0 )
	local mul = collectgarbage( "setstepmul", 10000 )
	local success, e = pcall( xs.call, xs, asKSEvent, scheduler )
	collectgarbage( "setpause", pause )
	collectgarbage( "setstepmul", mul )

	assert( success and e and e.ownedOnDataE, "expected event stream" )

end

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

return WFRxEventBridgeTest
