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

KSReactive.lua

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

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

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

--[[

	KSReactive: Reactive Programming Support
	
	We support two fundamental types: event streams and behaviors. All behaviors
	are also event streams that fire when the behavior changes value. The
	key differences between the two is that behaviors have a current value and
	only fire when that value actually changes whereas event streams have no
	persistent value and can fire the same value repeatedly.
	
	function KSReactive.sourceE() -> source event node
	
	function KSReactive.variableB() -> variable behavior node
	
	function KSReactive.constantB( value... ) -> constant behavior node with the
		given value.
		
	function KSReactive.constantErrorB( message... ) -> constant behavior node with the
		given value as an error.
		
	function KSReactive.anyE( ... ) -> merged event stream
		The vararg portion is a list of event streams. This combines multiple event streams.
		If any of the streams fire in a given cycle, this fires true.
		
	function KSReactive.zeroE()
		Returns an event stream that starts out (and remains) done.
		
	function KSReactive.callB( fn, ... ) -> behavior node for fn( ... )
		It will re-evaluate when the function or parameters change. If the function
		or the arguments are not behaviors, they will be run through constantB
		to get converted to behaviors.
		
	function KSReactive.argTupleB( ... ) -> behavior for ... with error detection
		If none of ... is an error, then the value is the standard argument format extension
		of ... in which the last argument can supply arbitrarily many values. If any of the
		values is an error, then the result is one of those errors.
		
	function KSReactive.dataCallB( fn, ... ) -> behavior for fn( ... )
		It will re-evaluate when the function or parameters change. If the function
		or the arguments are not behaviors, they will be run through constantB
		to get converted to behaviors. The function will not be called for errors and
		instead errors will pass through unchanged.
		
	function KSReactive.closingDataCallB( fn, ... ) -> behavior for fn( ... )
		This works just like dataCallB but it claims ownership of its input behaviors.
		
	function KSReactive.liftB( fn ) -> a function that accepts behaviors as inputs
		and returns a behavior as a result. Effectively:
			 function( ... ) return dataCallB( fn, ... ) end
			
	function KSReactive.lift( t ) -> a table in which the entries for any field with
		a name ending in a capital B has been lifted via KSReactive.liftB.
				
	function KSReactive.laterB()
		Returns a behavior that transitions from false to true when the event processor
		gets a chance to process later.
		
	function KSReactive.timerE( delayTime ) -> timer event stream
		Returns an event stream that will fire the value true after the given delay time.
		Delay times are rounded and grouped so that we can avoid creating too many
		separate timers. If one passes 'later' for delayTime, the event stream fires
		essentially at the same time laterB would fire.
		
	function KSReactive.timedSeqB( value0, delay, value1 )
		Returns a behavior that starts at value0 and transitions to value1 after the indicated
		delay.
	
	function KSReactive.falseThenTrueB( delay )
		Returns a behavior that starts at false and transitions to true after the indicated
		delay.
	
	function KSReactive.trueThenFalseB( delay )
		Returns a behavior that starts at true and transitions to false after the indicated
		delay.
	
	function KSReactive.isEventNode( v ) -> v_is_event_node, v_is_behavior
		Test whether v is a reactive event stream node and furthermore whether
		it is a behavior.
	
	function KSReactive.isBehavior( v ) -> v_is_behavior
		Test whether v is a behavior. (Short form for the second parameter from
		isEventNode.)
	
	-- FIXME [mhamburg 2011-03-14] : [scouten 2011-03-11]: Looks like this is out of date with the
	-- current set of reactive nodes.
	
	function KSReactive.makeWeakEventTable( opt_initial_value_for_behaviors )
		A class for constructing a table to manage a weak mapping to event
		nodes -- either events or behaviors. These can be managed as follows:
		
			et =  KSReactive.makeWeakEventTable( < initial value for behaviors > )
			
			local E = et:eventE( 'foo' )
			local B = et:behaviorB( 'baz' )
			
			-- All functions do nothing if the event doesn't currently exist. These
			-- return true if the event exists and false if it doesn't.
			
			et:postToEvent( 'foo', 3 )
			et:postErrorToEvent( 'foo', "Disk full" )
			et:closeEvent( 'baz' )
			
			-- But we might also want to test for event existence
			
			et:eventExists( 'bar' )
			
			-- We can get a list of the events currently in existence.
			
			et:eventKeys()
			
	We can also provide standard support for simple caches of behaviors. The
	following routine is basically an extension of WFCloseSupport.fetchCached
	but it also checks to make sure that the value is still considered valid.
	
	FIXME: This needs to do something intelligent with errors. Rework that
	when we rework some of the other protocols.
	
		KSReactive.fetchCachedBehavior( cache, key, validateValueNow )
				
	Event stream modifiers:
	
		e:holdB( value... )
			Returns a behavior that takes value... as its initial value and
			updates whenever e fires.
			
		e:closingHoldB( value... )
			Returns a behavior that takes value... as its initial value and
			updates whenever e fires. Takes close responsibility for e.
			
		e:holdValueNowB( b )
			Return a behavior that starts out with the value of b and changes
			whenever e fires.
			
		e:closingHoldValueNowB( b )
			Return a behavior that starts out with the value of b and changes
			whenever e fires. Takes close responsibility for e.
			
		e:relayE()
			Returns an event stream that supplies the exact same values as
			e. Relays can be used to protect underlying event streams from
			external modification, closure, etc.
			
		e:autoCloseRelayE()
			Auto-closing relay logic for events. See relayE. However, if this
			node is closed or simply GC'd we will check to see whether there
			are any other active successors for e and if not we will close e.
			
		b:relayB()
			Relay logic for behaviors. See relayE.
			
		b:autoCloseRelayB()
			Auto-closing relay logic for behaviors. See autoCloseRelayE.
	
		e:mapE( fn )
			Takes an event stream and runs each of the events through fn and
			returns the mapped value. Errors are passed through unchanged.
			
		e:closingMapE( fn )
			Takes an event stream and runs each of the events through fn and
			returns the mapped value. Errors are passed through unchanged.
			Sets the source as a close target.
			
		b:mapB( fn )
			Takes a behavior and runs its value through fn to generate a new
			mapped value. Errors are passed through unchanged.
			
		b:closingMapB( fn )
			Does a mapB on the behavior and sets the behavior as a close target
			of the result.
		
		b:closingKnownMapB( fn )
			Same as closingMapB, but omits calling fn whenever any of the inputs are
			WFUnknown.value()
		
		e:rawMapE( fn )
			fn takes values in the form 'data', < data > and 'error', < error  message >
			and gets to map them to similar format values.
			
		e:closingRawMapE( fn )
			fn takes values in the form 'data', < data > and 'error', < error  message >
			and gets to map them to similar format values. Sets the source as a close
			target.
			
		b:rawMapB( fn )
			fn takes values in the form 'data', < data > and 'error', < error  message >
			and gets to map them to similar format values.
			
		b:closingRawMapB( fn )
			fn takes values in the form 'data', < data > and 'error', < error  message >
			and gets to map them to similar format values. Sets b as a close target.
			
		e:filterE( fn )
			Takes an event stream and tests each value against fn and passes
			through only those values for which fn returns a true value.
			Error values pass through unchanged. (FIXME: If fn generates an
			error, it isn't clear what should happen.)
			
		e:closingFilterE( fn )
			Takes an event stream and tests each value against fn and passes
			through only those values for which fn returns a true value.
			Error values pass through unchanged. (FIXME: If fn generates an
			error, it isn't clear what should happen.) Sets the source as a
			close target.
			
		e:constantE( value... )
			Turns every value from e into value... Passes errors through
			unchanged.
			
		-- OBSOLETE: e:switchE()
			e should be an event stream valued event stream. Each time it fires,
			we change which event stream we are listening to to produce the
			output. This passes errors through unchanged. It is done when both
			e is done and the last event stream if any supplied by e is done.
			
		e:mapSwitchE( fn )
			fn should be a function that transforms the values provided by e into
			event streams. We then switch over these streams. As such, it is
			equivalent to e:mapE( fn ):switchE(). The distinction between this
			version built out of the primitive modifiers and mapSwitchE is
			that mapSwitchE takes ownership of the mapped values and will
			close them when switching.
			
		e:closingMapSwitchE( fn )
			Runs e through fn as a mapSwitchE and sets e as a close target for
			the result.
			
		b:mapSwitchB( fn )
			fn should map the values from b to behaviors. As with mapSwitchE,
			this is essentially equivalent to b:mapB( fn ):switchB() with
			the added property that it takes ownership of the values produced
			by the map and closes them when switching. (BUG: Documentation in
			terms of the now non-existent switchB is a bit of a problem.)
			
		b:closingMapSwitchB( fn )
			Does a mapSwitchB on the behavior and sets the behavior as a close
			target for the result.
			
		b:closingKnownMapSwitchB( fn )
			Same as b:closingMapSwitchB, but omits calling fn whenever any of the inputs are
			WFUnknown.value()
		
		b:zipMapSwitchB( fn )
			Like mapSwitchB but prepends the value of b (only one position if
			b is a multiple value) to the values from the behavior produced
			by fn. This is useful if you need to keep track of where a value
			came from.
			
		e:skipErrorsE()
			Returns an event stream that drops any error values from e and
			passes everything else through unchanged.
			
		e:endAfterErrorE()
			Returns an event stream that sets itself to done after the first
			error value.
			
		e:closingEndAfterErrorE()
			Returns an event stream that sets itself to done after the first
			error value. Sets e as a close target.
			
		e:takeE( n )
			Take n events and then terminate. If the source delivers fewer
			than n events, terminate early. n should be a number greater than
			zero. (We effectively take the ceiling when counting down.)
			
		e:accumulateE( fn, accum )
			For each event, we compute fn( accum, ... ). The first result is
			taken as the new accumulator value and all of the results are
			returned. Errors are passed through without affecting the accumulator.
			If you want errors to be sticky, follow or precede this with
			endAfterErrorE.
			
		e:accumulateB( fn, accum, ... )
			A behavior that starts out at accum, ... and then builds up values
			in the same manner as accumulateE.
			
		e:mergeE( otherStreamE )
			Merge the event stream e with the event stream otherStreamE. If
			both streams fire on a given cycle, then only the receiver of this
			method's value is propagated.
			
		e:calmE( interval )
			We are either active or calm. When active, an event flows through
			and sets the state to calm for the time period specified by interval.
			While calm, events just accumulate. When we exit the calm state, we
			will transmit the last value to come through -- if any -- and if there
			is a value re-enter the calm state. Errors propagate immediately and
			also take the node out of the calm state.
			
		b:calmB( interval )
			Like calmE but working with behaviors. Changes to the behavior get
			calmed.
			
		b:mcallB( methodName, optDefault )
			For data values (as opposed to errors) call a method on the values
			to get a new value. Basically a shorthand for mapB. The additional
			arguments will be passed as arguments to the method. (FUTURE: Allow
			the arguments to themselves be behaviors.) If the current value is
			nil or false, we will use the default if supplied or the current
			value if not supplied.
			
		b:mcallSwitchB( methodName, optDefault )
			Like mcallB but this works with methods that return behaviors by
			then doing a switch to drop back down to values.
			
		b:propertyB( propertyName, optDefault )
			Extracts the field [ propertyName ] from the incoming value.
			Uses optDefault if incoming value is nil or false. If optDefault is not
			supplied, then it returns the incoming value in these cases.
			
		b:closingPropertyB( propertyName, optDefault )
			Extracts the field [ propertyName ] from the incoming value.
			Uses optDefault if incoming value is nil or false. If optDefault is not
			supplied, then it returns the incoming value in these cases. Takes
			on close responsibility for b.
			
		b:toBooleanB()
			= b:mapB( function( v ) return not not v end )
			
		b:closingToBooleanB()
			= b:closingMapB( function( v ) return not not v end )
			
		b:filterB( fn, ... )
			Filters the values of b so that the value stream only contains values
			meeting the test of the predicate fn. If the initial values don't meet
			the predicate, then we use ... as a starting point. Errors pass through
			unchanged.
			
		b:closingFilterB( fn, ... )
			Filters the values of b so that the value stream only contains values
			meeting the test of the predicate fn. If the initial values don't meet
			the predicate, then we use ... as a starting point. Errors pass through
			unchanged. Sets the source as a close target.
		
		b:closingObjectRelayB()
			Automatically relays and closes old relays the values passing through b.
			The values are expected to have a method relayObject() which creates
			a new copy of the object that can be independently closed (as WFObjectProxy
			does). When the resulting behavior is closed, the currenty value is
			also closed.
			
		b:skipErrorsB( ... )
			Remove errors from the stream of values for the behavior. If the initial
			value is an error, replace it with ....
			
		b:valueNow()
			The current value of the behavior. Use with caution since the behavior
			may not have updated if called during the current event cycle. Also
			beware that behaviors also often represent values to be delivered
			eventually and as such may not have the expected value at this particular
			moment. Errors come through as packed values (KSRPackedValues). You
			probably want to use the next routine to avoid needing to deal with
			KSRPackedValues.
			
		b:isErrorNow()
			Returns true/false and the error message if any if true. This reports
			whether the behavior value is currently an error.
			
		b:assertValueNow( value, message )
			Confirm that the current value matches the given value asserting with
			the given message if it doesn't.
			
	The escape hatches. These are used to connect the network up to the outside
	world. As such they stay alive as long as their sources chains are alive even
	if they aren't explicitly referenced anywhere else. Hence, if you don't want
	this to happen, you need to be good about explicitly closing them.
	
		e:onDataE( fn )
			Runs fn when e fires with normal data passing it the data.
			
		e:onErrorE( fn )
			Runs fn when e fires an error.
			
		b:onDataB( fn )
			Runs fn on the current value of b (as of this cycle) and whenever
			b changes provided that the value of b is not an error value.
			This is also a behavior returning the same values as b (including
			errors).
			
		e:ownedOnDataE( owner, fn )
			Attaches an onDataE to an owning object closing it when the owner
			closes.
			
		b:ownedOnDataB( owner, fn )
			Attaches an onDataB to an owning object closing it when the owner
			closes.
			
		e:weakOnDataE( fn )
			Runs fn when e fires with normal data passing it the data. This will
			not keep the event chain alive on its own.
			
		b:weakOnDataB( fn )
			Runs fn when constructed and when b changes with normal data passing
			it the data. This will not keep the event chain alive on its own.
			
		e:closingOwnedOnDataE( owner, fn )
			Attaches an onDataE to an owning object closing it when the owner
			closes. Sets e as a close target.
			
		b:closingOwnedOnDataB( owner, fn )
			Attaches an onDataB to an owning object closing it when the owner
			closes. Sets b as a close target.
			
		e:ownedOnErrorE( owner, fn )
			Attaches an onErrorE to an owning object closing it when the owner
			closes.
			
		e:ownedOnDoneE( owner, fn )
			Runs fn when e is done. This attaches the node to owner so that it
			closes when owner closes.
			
		e:weakOnDoneE( fn )
			Runs fn when e announces that it is done but will not keep e
			alive.
			
		e:closingWeakOnDoneE( fn )
			Runs fn when e announces that it is done but will not keep e alive.
			Sets e as a close target.
			
	We can also cause a node to live until it is done.
		
		e:runUntilDone()
			
	We can monitor a node to see what values it sends:
	
		n:setDebugHook( [ optHookFn = debug.print ], [ optHookName ] )
			
	Close v Done. Closing a source node starts propagating a done signal through
	the network. This generally results in nodes closing as they lose all of their
	sources. Behaviors, however, need not close when done (though they will notify
	their successors that they are done).
		
	We can add support methods for working with the reactive system to other
	classes via KSReactive.addReactiveSupportMethods. The class must also
	support setOwned (generally derived from addCloseSupportMethods). This adds:
	
		obj:createReactiveVariable( key, ... )
			Checks that obj[ key ] is a false value and then sets it to
			a reactive variable with initial value ....
	
]]

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

--[[

Requests:

	b:callWhenE( value, fn, ... )
		When the b:valueNow() == value, call fn( ... ) and return the result
		(or the error) via an event stream constructed by this call.
		See undoRedoImp in OzDocStore.lua.

]]

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

-- local assertf = require 'KSLuaUtils'.assertf
-- local string_format = string.format
-- local pairs = pairs
local assert = assert
local type = type
local select = select
local rawset = rawset
local unpack = unpack
local pcall = pcall
-- local tostring = tostring
local pairs = pairs
local setmetatable = setmetatable

local WFArguments = require 'WFArguments'
local WFArguments_check = assert( WFArguments.check )

local string_format = assert( string.format )

local math_max = math.max

local KSLeaks = require 'KSLeaks'

local KSLeaks_registerObject = assert( KSLeaks.registerObject )

local WFCloseSupport = require 'WFCloseSupport'

local WFCloseSupport_hasBeenOrWillBeClosed = assert( WFCloseSupport.hasBeenOrWillBeClosed )
local WFCloseSupport_fetchCached = assert( WFCloseSupport.fetchCached )

local WFIdle = require 'WFIdle'

local WFQueueing = require 'WFQueueing'

local WFQueueing_makeQueue = assert( WFQueueing.makeQueue )

local WFVarArgs = require 'WFVarArgs'
local WFVarArgs_map = assert( WFVarArgs.map )
local WFVarArgs_findFirst = assert( WFVarArgs.findFirst )

local KSRPackedValues = require 'KSRPackedValues'

local KSRPackedValues_pack = KSRPackedValues.pack
local KSRPackedValues_packOne = KSRPackedValues.packOne
local KSRPackedValues_unpack = KSRPackedValues.unpack
local KSRPackedValues_error = KSRPackedValues.error
local KSRPackedValues_isError = KSRPackedValues.isError
local KSRPackedValues_first = KSRPackedValues.first
local KSRPackedValues_testEquals = KSRPackedValues.testEquals
local KSRPackedValues_packPCall = KSRPackedValues.packPCall

local kSameValue = { }
local kSkipValue = { }
local kDoneValue = { }
local kScheduleRequestValue = { }

local WFDebugTraceback_traceback = require 'WFDebugTraceback'.traceback

local traceback = function()
	return WFDebugTraceback_traceback { "KSReactive.lua" }
end

local DEBUG_NODE_BURSTS = false  -- should normally be false

local ALLOW_EMPTY_TUPLES_FROM_MAP = false
	-- Can mapE and mapB return empty tuples and have those propagate properly?
	-- We ought to support this but historically we have used that as an indicator
	-- to just leave the value unchanged. Try flipping the flag and see what
	-- hapepens...
if require 'WFCapabilities'.getCapability 'KSReactive_emptyTuplesFromMap' == 'allowed' then
	ALLOW_EMPTY_TUPLES_FROM_MAP = true
end
	
local AUTO_CLOSE = false
	-- Do we want to use the new auto close logic?
	
local AUTO_CLOSE_INITIAL_NODES = false
	-- Do we want to set initial nodes up for auto closure?
	
local AUTO_CLOSE_PROTECTS_STRONG_SUCCESSOR = AUTO_CLOSE_INITIAL_NODES
	-- Do we want to protect onData and friends from auto-closure? (Irrelevant
	-- without AUTO_CLOSE_INITIAL_NODES.)
	
local TRACE_NODE_CREATION = AUTO_CLOSE
	-- Do we want to capture a traceback when we create nodes?
	

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

-- FIXME: Move this elsewhere or make it a standard utility

local function lazyFunction( moduleName, functionName )

	local fn = nil

	return function( ... )
	
		if fn == nil then
			fn = require( moduleName )[ functionName ]
		end
		
		return fn( ... )
	
	end

end


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

-- Bypass grabbing all of KSLuaUtils for non-debug builds.

local KSLuaUtils, dpcall, assertft

if _G.DEBUG then

	KSLuaUtils = require 'KSLuaUtils'
	dpcall = assert( KSLuaUtils.dpcall )
	assertft = assert( KSLuaUtils.assertft )

else

	-- Minimally compatible version of above functions with no debug output.

	dpcall = function( ident, ... )
		return pcall( ... )
	end
	
	assertft = function( value )
		return assert( value )
	end

end

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

-- Bypass grabbing all of KSTableUtils just to be able to make fully weak tables.

local fullyWeakMT = { __mode = 'kv' }

local function makeFullyWeakTable()
	return setmetatable( {}, fullyWeakMT )
end


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

-- FIXME: Find a better home for this

-- makeTriggerFunction{ schedule = ..., action = ..., suppressRetrigger = ... }

local function makeTriggerFunction( args )
	
	local scheduled = false
	
	local suppressed = false
	
	local schedule = args.schedule
	local action = args.action
		
	local run
	
	if args.suppressRetrigger then
	
		run = function()
			scheduled = false
			suppressed = true
			pcall( action )
			suppressed = false
		end
	
	else
	
		run = function()
			scheduled = false
			action()
		end
	
	
	end
	
	return function()
		if not scheduled then
			scheduled = true
			schedule( run )
		end
	end
	
end

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

local function makeDeferredProcessingPool( scheduleFn, processFn )

	local pool = nil
	
	local running = false
	
	local supply = nil -- a waiting pool for use
	
	-- Run through flush pool processing values and clearing the pool. Exit
	-- when done or if we hit an error. We will handle errors via a pcall loop,
	-- but this way we only bother with one pcall if everything succeeds.
	
	local function flush_imp( flushPool )
		for obj, value in pairs( flushPool ) do
			flushPool[ obj ] = nil
			processFn( obj, value )
		end
	end
	
	local function flush()
		running = true
		while pool do
			local oldPool = pool
			pool = nil
			repeat
				local success, message = pcall( flush_imp, oldPool )
				if not success then
					debug.print( "Error draining deferred processing pool: " .. tostring( message ) )
				end
			until success
			supply = oldPool
		end
		running = false
	end
	
	local trigger = makeTriggerFunction{
		schedule = scheduleFn,
		action = flush,
		suppressRetrigger = true
	}
	
	local function add( obj, value )
		if value == nil then
			value = true
		end
		if pool then
			pool[ obj ] = value
		else
			if supply then
				pool = supply
				supply = nil
				pool[ obj ] = value
			else
				pool = { [ obj ] = value }
			end
			if not running then
				trigger()
			end
		end
		--[[
		debug.print( "--- add: begin pool contents ---" )
		for entry in pairs( pool ) do
			debug.print( tostring( entry ) )
		end
		debug.print( "--- end pool contents ---" )
		]]
	end
	
	local function remove( obj )
		if pool then
			pool[ obj ] = nil
			--[[
			debug.print( "--- remove: begin pool contents ---" )
			for entry in pairs( pool ) do
				debug.print( tostring( entry ) )
			end
			debug.print( "--- end pool contents ---" )
			]]
		end
	end
	
	return add, remove, flush

end

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

local protectedFromAutoClose = makeFullyWeakTable()

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

local function protectFromAutoClose( obj )
	protectedFromAutoClose[ obj ] = obj
end

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

local addToAutoClose, removeFromAutoClose, flushAutoClose =
	makeDeferredProcessingPool( WFIdle.doASAP, function( obj, case )
		if not protectedFromAutoClose[ obj ]
			and not WFCloseSupport_hasBeenOrWillBeClosed( obj )
		then
			obj:maybeAutoClose( case )
		end
	end )

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

local function checkForAutoClose( obj, case )

	if not AUTO_CLOSE then
		return
	end

	if not WFCloseSupport_hasBeenOrWillBeClosed( obj ) then
		addToAutoClose( obj, case )
	end

end


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

local protectStrongSuccessorFromAutoClose

if AUTO_CLOSE_PROTECTS_STRONG_SUCCESSOR then
	protectStrongSuccessorFromAutoClose = protectFromAutoClose
else
	protectStrongSuccessorFromAutoClose = function() end
end


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

--[[

Implementation Notes:

Posting to an event stream or setting the value on a reactive variable triggers
a reactive cycle. At the base level, this is handled via two mechanisms: the
delivery "queue" and the schedule queue.

THE DELIVERY QUEUE
------------------
The delivery queue is a sequence of node/value pairs for which we want to deliver
the given value to the given node. Delivery will invoke the appropriate method on the
node based on the value type:

	_receiveData( packedValue ): Receives a normal packed value as data
	_receiveError( errorValue ): Receives an error value
	_receiveDone( allSourcesGone ): This ties to the magic done value,
		but instead of receiving that value, it receives a flag indicating
		whether all of the sources for the node are done.
	
Each can return a value to be handed on to the node's successors or can
return kScheduleRequestValue which will queue the node to run at the appropriate
time in the topological sort at which point its _run method will be invoked
and this can in turn return a packed value to pass on up the chain.

node_deliverPackedValue( node, packedValue ) is the dispatch function that
runs this logic by deciding which method to invoke and how to handle the result.

node_deliverPackedValue is fed from the delivery queue which contains pairs of
nodes and packed values to deliver.

node_sendPackedValue is the routine called with a new value to be fed into
the system. Generally, it enqueues the value onto the delivery queue for all
of the successors for the node. If the value is the done value, it will also
initiate a tear down of the node including running any close targets and other
pieces of the close notification system. This will occur before the next node
gets a chance to run.

If node_deliverPackedValue fails, we will invoke node_deliveryFailed which will
______. FIXME

The optional table at valuesSentThisCycle records the values delivered by
node_sendPackedValue. Any given node should only have node_sendPackedValue
called once per cycle.

THE SCHEDULE QUEUE
------------------
The schedule queue implements topological sorting on the reactive graph. If
while processing a delivered message the node returns kScheduleRequestValue,
then the node will be placed on the schedule queue. The schedule queue is
a priority queue sorted by node depth at the time of insertion. When we remove
nodes from the queue, we double-check their depth since it is possible that the
depth will have increased while they were queued. (Depth never decreases.) If
it has increased, we requeue and try again. If the depth matches, then we invoke
the _run method on the node which like the various delivery methods above can
return values to pass to the node's successors via node_sendPackedValue which
in turn puts values onto the delivery queue.

THE CORE REACTIVE CYCLE
-----------------------
We pull from the delivery queue until it is empty and then pull from and process
nodes on the schedule queue until the delivery queue is again non-empty. The
core reactive cycle is done when both of these queues are empty. When both queues
are empty, we discard our knowledge of values sent during this reactive cycle
(stored in valuesSentThisCycle). This logic is encoded in runScheduledQueue.

OUTER CYCLES
------------
We have a series of levels of outer cycles where we will accumulate node/value
pairs for later delivery. These will only accumulate one (packed) value per
node so multiple accumulations for the same node will only keep the last value.
Nodes using these levels are known as "delayed" nodes and are implemented through
the delayE constructor with appropriate parameter values to select the delay
level. These nodes break cycles in the topological sorting by always having a depth
of zero since they don't fire until at least one cycle after they receive a value.

These levels are:

next tick: Used by nodes constructed by delayE. These values will be sent
immediately after we complete a core reactive cycle and will trigger the next
reactive cycle.

stable: These values will be sent when the input queue (see below) is empty.

THE INPUT QUEUE
---------------
Sitting between the next tick level and the stable level is the input queue.
This is a sequence of node/value pairs (possibly with multiple entries per
node). This is where outside events feed in. As we drain the input queue
(runInputQueue), we run core reactive cycle and next tick cycles as appropriate.

]]

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

--[[

KSRDirectedAcyclicGraph:

	This module provides a routine for creating a table full of functions for
	manipulating a DAG built out of tables with private keys. The DAG recognizes
	both strong and weak nodes where weak nodes are not kept alive by their
	predecessors but nodes always keep their predecessors alive.
	
	When we create the table of functions, we supply it with a table of
	customization options including:
	
		nodeName: A function to take a node table and return a string for debugging
			-- GRUNGE [mhamburg 2011-01-12]: This could just use __tostring.
			
		gainedFirstSuccessor: A function taking a node. This will be called
			when a node gains its first successor.
			
		lostSuccessor: A function taking a node to be called if the node loses
			a successor.
			
	The functions created in the table include:
	
		function allocNode( optStrongSuccessor )
			This creates a table with various fields filled in. The strong flag
			indicates whether this will be a strong node. The node is not
			actually ready for use until you also call:
			
		function initNode( node, nodeKind, isBase, ...predecessors... )
			This finishes the node initialization and wires it up appropriately.
			The varargs section lists its predecessors. This returns the node.
			
			So the allocation pattern is:
			
				node = allocNode( optStrongSuccessor )
				... set values before initialization completes ...
				initNode( node, nodeKind, isBase, ... predecessors... )
				
			A short version with no extra initialization is:
			
				return initNode( allocNode( strong ), nodeKind, isBase, ...predecessors... )
			
		function addLink( predecessor, successor )
			Add a link from predecessor to successor.
			
		function removeLink( predecessor, successor )
			Remove the link if any from predecessor to successor.
			
		function iterSuccessors( node )
			Returns an iterator for the successors of a node. It is an error
			(undetected at present) to add successors while iterating successors.
			
		function getNodeDepth( node )
			Get the current depth for node.
			
		function isNode( v )
			Returns true iff v is a node. (Note done nodes are still nodes.)
			
	It also includes the following debugging functions. This is where the node
	name actually matters.
	
		function assertNode( node [ , optExtraMessage ] )
			Check that node is a valid node for this DAG. Throw if it is not.
			
		function checkDepths( rootNode )
			Makes sure that the depth values are all properly increasing as
			we move from predecessors to successors.
			
		function checkSuccessors( node, ...successors... )
			Check that the successors to node are exactly the listed nodes.

]]

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

-- FIXME [mhamburg 2011-01-12]: We could optimize the predecessors case for
-- single predecessors I haven't done that yet, because it isn't critical
-- in order to get this working.

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

-- FIXME [mhamburg 2011-01-12]: Move to KSRIterators

local function iterNone()
	return nil
end

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

-- Add element to a set at t[ k ] creating the set if need be. Returns true
-- if the set didn't exist before (which is akin to it being empty though
-- not quite the same).

local function addSetFieldElement( t, k, element )
	local s = rawget( t, k )
	local result = false
	if not s then
		s = {}
		rawset( t, k, s )
		result = true
	end
	s[ element ] = element
	return result
end

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

-- Remove an element from an optional set at t[ k ]. Removes the set if it is
-- now empty and returns true.

local function removeSetFieldElement( t, k, element )
	local s = rawget( t, k )
	if s and s[ element ] then
		s[ element ] = nil
		if next( s ) == nil then
			rawset( t, k, false )
			return true
		end
	end
	return false
end

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

-- Iterate a set field if not empty.

local function iterSetField( t, k )
	local s = rawget( t, k )
	if not s then
		return iterNone
	else
		return pairs( s )
	end
end

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

-- We need to track whether or not something is a node for debugging purposes.
-- We store the node kind as the value.

local WFNodeWatcher = require 'WFNodeWatcher'
local nodeWatcher = WFNodeWatcher( 'reactive', DEBUG_NODE_BURSTS, 100 )

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

local function isNode( node )
	return nodeWatcher:isNode( node )
end

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

local function assertNode( node, msg )
	if nodeWatcher:isNode( node ) then
		return node
	end
	local tback = traceback()
	local errMsg
	if msg then
		errMsg = string_format( "Expected to be passed a node (%s): %s", tostring( msg ), tback )
	else
		errMsg = string_format( "Expected to be passed a node: %s", tback )
	end
	error( errMsg, 0 )
end

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

-- We iterate successors using the optional set field logic. Note, however,
-- that because we apply different rules to successors, we do not use those
-- routines for adding and removing successors.

-- FIXME [mhamburg 2011-03-14] : [scouten 2011-03-11]: Scratching my head over this comment. Perhaps
-- you could explain what is different about successors and how that impacts
-- the add/remove logic.

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

local function node_iterSuccessors( node )
	return iterSetField( node, 'private:allSuccessors' )
end

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

local function node_iterPredecessors( node )
	return iterSetField( node, 'private:predecessors' )
end

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

-- Notification and update for a node achieving strong status by gaining its first
-- strong successor.
	
local function node_becameStrong( node )
	for pred in node_iterPredecessors( node ) do
		if addSetFieldElement( pred, 'private:strongSuccessors', node ) then
			if not rawget( pred, 'private:nodeIsStrongSuccessor' ) then
				node_becameStrong( pred )
			end
		end
	end
end

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

-- Notification and update for a node losing strong status by losing all of its
-- strong successors.
	
local function node_becameWeak( node )
	for pred in node_iterPredecessors( node ) do
		if removeSetFieldElement( pred, 'private:strongSuccessors', node ) then
			if not rawget( pred, 'private:nodeIsStrongSuccessor' ) then
				node_becameWeak( pred )
			end
		end
	end
end

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

-- Does the node have a predecessor? This relies on the clearing logic dropping
-- us down to false instead of a table.

local function node_hasPredecessor( node )
	return not not rawget( node, 'private:predecessors' )
end

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

-- Does the node have a successor?

local function node_hasSuccessor( node )

	local allSuccessors = rawget( node, 'private:allSuccessors' )
	
	if not allSuccessors then
		return false
	end
	
	if next( allSuccessors ) ~= nil then
		return true
	end

	rawset( node, 'private:allSuccessors', false )
	
	return false

end

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

-- We use the depth field to mark nodes as done by setting it to -1. Since we won't
-- allow done nodes in links, this shouldn't matter.

local function node_isDone( node )
	return rawget( node, 'private:depth' ) == -1
end

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

-- We can force a node to be a base node. When this is the case, it has depth zero.

-- FIXME [mhamburg 2011-03-14] : [scouten 2011-03-11]: What does it mean to be a base node? Is this an init-time
-- operation, or can you mark a node as base at any time? If init-time, perhaps this
-- condition should be tested/verified?

local function node_markAsBase( node )
	assert( not node_isDone( node ), "Cannot change base status for a done node." )
	rawset( node, 'private:depth', 0 )
end

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

local function node_isBase( node )
	return rawget( node, 'private:depth' ) == 0
end

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

-- Recursively walk the successor links from a node making sure that the depth
-- is at least the supplied depth and increases by at least one each time we
-- follow a link. This will run forever if the graph has cycles, but we use it
-- in addLink to prevent that.

local function node_raiseDepth( node, newDepth )

	if node_isBase( node ) then
		return
	end

	if newDepth <= rawget( node, 'private:depth' ) then
		return
	end
	
	rawset( node, 'private:depth', newDepth )
	
	local nextDepth = newDepth + 1
	
	for successor in node_iterSuccessors( node ) do
		node_raiseDepth( successor, nextDepth )
	end
	
end

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

-- Add a link from pred to succ.

local function nodes_addLink( pred, succ )

	assert( not WFCloseSupport_hasBeenOrWillBeClosed( pred ),
			"addLink: predecessor should not have been closed" )

	assert( not WFCloseSupport_hasBeenOrWillBeClosed( succ ),
			"addLink: successor should not have been closed" )

	-- The successor should not be done.
	
	assert( not node_isDone( succ ), "addLink: successor should not be done" )
	
	-- If the predecessor is done, we simply fail to construct the link.
	
	if node_isDone( pred ) then
		return
	end
	
	-- Make sure the depth is right (this also detects cycles though in doing so it leaves the depths incremented)

	local predDepth = rawget( pred, 'private:depth' )
	
	node_raiseDepth( succ, predDepth + 1 )

	assert( rawget( pred, 'private:depth' ) == predDepth, "Cannot add cyclic link" )
	
	-- The predecessor won't auto close in this case
	
	removeFromAutoClose( pred )
	
	-- Check the strong status for each node. We may be changing the status for the
	-- predecessor node, so we only check whether the local flag would block
	-- propagation.

	local succIsStrong = rawget( succ, 'private:nodeIsStrongSuccessor' ) or rawget( succ, 'private:strongSuccessors' )
	local predNodeIsStrong = rawget( pred, 'private:nodeIsStrongSuccessor' )

	local allSuccessors = rawget( pred, 'private:allSuccessors' )
	
	if not allSuccessors then
		allSuccessors = makeFullyWeakTable()
		rawset( pred, 'private:allSuccessors', allSuccessors )
		-- FIXME? gainedFirstSuccessor( pred )
	end
	
	allSuccessors[ succ ] = succ
	
	addSetFieldElement( succ, 'private:predecessors', pred )

	-- If strong, then we need to add it to the set of strong successors and
	-- possibly run up the tree adding strong successors.
	
	if succIsStrong then
		if addSetFieldElement( pred, 'private:strongSuccessors', succ ) then
			if not predNodeIsStrong then
				node_becameStrong( pred )
			end
		end
	end
	
end

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

-- Remove the link if any from pred to succ.

local function nodes_removeLink( pred, succ )

	-- We exit early if there isn't a link from pred to succ
	
	local allSuccessors = rawget( pred, 'private:allSuccessors' )
	
	if not allSuccessors then
		return
	end
	
	if not allSuccessors[ succ ] then
		return
	end
	
	-- Queue predecessor for auto close
	
	checkForAutoClose( pred, 'lost-successor' )
	
	-- Deal with strong status first.
	
	if removeSetFieldElement( pred, 'private:strongSuccessors', succ ) then
		if not rawget( pred, 'private:nodeIsStrongSuccessor' ) then
			node_becameWeak( pred )
		end
	end
	
	-- Now break the weak link.
	
	allSuccessors[ succ ] = nil
	
	-- Remove the predecessor link.
	
	removeSetFieldElement( succ, 'private:predecessors', pred )
	
	-- And do any necessary notification.

	-- FIXME?
	-- if lostSuccessor then
	--		lostSuccessor( succ )
	-- end

end

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

-- Get the depth for a node.

local function getNodeDepth( node )
	assertNode( node, "getNodeDepth" )
	return rawget( node, 'private:depth' )
end

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

-- Get the kind for a node

local function getNodeKind( node )
	assertNode( node, "getNodeKind" )
	return nodeWatcher:getNodeKind( node )
end

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

-- A utility function for making a pre-initialized node table. This needs to be
-- handed to initNode to finish the initialization.

local function nodeDAG_allocNode( optStrongSuccessor )

	local node = {
		[ 'private:depth' ] = 1,
		[ 'private:predecessors' ] = false,
		[ 'private:allSuccessors' ] = false,
		[ 'private:strongSuccessors' ] = false,
		[ 'private:nodeIsStrongSuccessor' ] = not not optStrongSuccessor,
	}
	
	if TRACE_NODE_CREATION then
		node[ 'private:traceback' ] = traceback()
	end
	
	return node

end

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

-- Finish node construction potentially wiring the node into the graph based on the
-- list of predecessors.

local function nodeDAG_initNode( node, nodeKind, isBase, ... )

	assert( rawget( node, 'private:depth' ), "Table does not appear to have come from allocNode." )
	assert( not nodeWatcher:isNode( node ), "Node table appears to have already been used." )

	-- The varargs section provides us with a list of predecessors. We use those
	-- to determine the initial depth and after construction we wire them up as
	-- predecessors.
	
	local predDepth = 0
	
	local predCount = select( '#', ... )
	
	for i = 1, predCount do
		local pred = select( i, ... )
		if pred then
			assertNode( pred, "initNode: predecessor" )
			if not node_isDone( pred ) then
				predDepth = math_max( predDepth, rawget( pred, 'private:depth' ) )
			end
		end
	end
	
	-- We don't actually raise the depth for base nodes. We leave in the computation
	-- because it also does some validation.
	
	if isBase then
		node_markAsBase( node )
	else
		rawset( node, 'private:depth', predDepth + 1 )
	end
	
	-- Mark it as a node

	nodeWatcher:watchNode( node, nodeKind )
	
	-- Handle counting
	
	KSLeaks_registerObject( nodeKind, node )
	
	-- Wire up the predecessors
	
	for i = 1, predCount do
		local pred = select( i, ... )
		if pred and not node_isDone( pred ) then
			-- assertNode( pred, "nodeDAG_initNode:addLink: predecessor" )
			-- assertNode( node, "nodeDAG_initNode:addLink: successor" )
			nodes_addLink( pred, node )
		end
	end
	
	-- Return it for chaining
	
	return node

end

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

-- Mark the node as done. This disconnects it in both directions and adjusts
-- the depth.
	
local function disconnectNodeAndMarkAsDone( node )

	assertNode( node, "disconnectNodeAndMarkAsDone" )
	
	if node_isDone( node ) then
		return
	end
	
	-- Remove successor links
	
	for succ in node_iterSuccessors( node ) do
		nodes_removeLink( node, succ )
	end
	
	-- Remove predecessor links
	
	for pred in node_iterPredecessors( node ) do
		nodes_removeLink( pred, node )
	end
	
	-- Set the depth to mark the node as done

	rawset( node, 'private:depth', -1 )
	
end

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

local function wrapAssertNode( msg, fn )
	return function( node )
		assertNode( node, msg )
		return fn( node )
	end
end

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

local nodeDAG_nodeHasPredecessor = wrapAssertNode( "nodeHasPredecessor", node_hasPredecessor )
local nodeDAG_nodeIsBase = wrapAssertNode( "nodeIsBase", node_isBase )
local nodeDAG_iterSuccessors = wrapAssertNode( "iterSuccessors", node_iterSuccessors )
local nodeDAG_nodeIsDone = wrapAssertNode( "nodeIsDone", node_isDone )

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

-- FIXME [mhamburg 2011-01-18]: Move to generic utilities.

local function makeMetatable_populateFrom( source )
	
	return {
		__index = function( t, k )
			local v = source[ k ]
			rawset( t, k, v )
			return v
		end,
		__metatable = "< metatable: populate from >"
	}

end

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

local function makeNodePackedValueQueue()

	local enqueue1, dequeue1 = WFQueueing_makeQueue()
	
	local function enqueue( node, packedValue )
		enqueue1( node )
		enqueue1( packedValue )
	end
	
	local function dequeue()
		return dequeue1(), dequeue1()
	end
	
	return enqueue, dequeue

end

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

-- The delivery queue contains the messages to be delivered asap this cycle
-- without regard to further scheduling constraints. These are messages sent
-- by sources and via this queue arriving at destinations.

local deliveryQueue_pushNodePackedValue, deliveryQueue_popNodePackedValue =
			makeNodePackedValueQueue()

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

-- The input queue represents inputs from the public API. Post operations
-- and close operations from outside get encoded here. delay nodes run
-- before the input queue. stable nodes run after the input queue. The values
-- here represent values to be provided to the node send routine.

local inputQueue_pushNodePackedValue, inputQueue_popNodePackedValue =
			makeNodePackedValueQueue()
			
--------------------------------------------------------------------------------

-- We run the scheduler based on negative node depths so that higher node
-- depths are processed later. We manage this through a priority queue.

local scheduleQueue_enqueue, scheduleQueue_dequeue = WFQueueing.makePriorityQueue()

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

local function node_addToScheduleQueue( node )

	if rawget( node, 'private:isInScheduleQueue' ) then
		return
	end

	assert( not nodeDAG_nodeIsBase( node ),
		"Base nodes should not need to be scheduled" )

	local depth = getNodeDepth( node )
	
	scheduleQueue_enqueue( -depth, node )
	
	rawset( node, 'private:isInScheduleQueue', true )

end

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

local function nextScheduledNode()

	while true do
	
		local priority, node = scheduleQueue_dequeue()
		
		-- If the queue comes back empty, we're done.

		if not priority then
			return nil
		end

		-- It's no longer in the scheduled queue.

		rawset( node, 'private:isInScheduleQueue', false )
		
		-- We can only return it, however, if the depth is right.
		
		local depth = getNodeDepth( node )
		
		if depth <= -priority then
			return node
		end
		
		-- Otherwise, we need to reschedule.
		
		scheduleQueue_enqueue( -depth, node )
		
		rawset( node, 'private:isInScheduleQueue', true )
		
	end

end

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

local valuesSentThisCycle = nil
	-- A table mapping nodes to values for everything sent this cycle. Cleared
	-- after each cycle. Used when linking to nodes that have already fired.
	-- A cycle is defined as the end of running the propagation for one event
	-- in the network.

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

local function node_enqueuePackedValueForDelivery( node, packedValue )

	if valuesSentThisCycle == nil then
		valuesSentThisCycle = { }
	end
	
	assert( valuesSentThisCycle[ node ] == nil, "We can only send one value per node per cycle" )

	valuesSentThisCycle[ node ] = packedValue
	
	for successor in nodeDAG_iterSuccessors( node ) do
		deliveryQueue_pushNodePackedValue( successor, packedValue )
	end
	
end

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

local function node_sendPackedValue( node, packedValue )

	if nodeDAG_nodeIsDone( node ) then
		return
	end
	
	node_enqueuePackedValueForDelivery( node, packedValue )
		-- All of the sinks will receive the packed value via node_deliverPackedValue.
		
	local debugHook = rawget( node, 'private:debugHook' )
	
	if debugHook then
		debugHook( node, packedValue )
	end
		
	-- If the value is kDoneValue, then this node is done and we need to
	-- disconnect it from the network and mark it as done.
	
	if packedValue == kDoneValue then
		disconnectNodeAndMarkAsDone( node )
		if node[ 'private:closeWhenDone' ] then
			node:close()
		end
	end

end

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

-- FIXME [mhamburg 2011-05-05]: Failure here should probably be treated as fatal
-- since it signals a problem putting things onto queues.

local function node_finishDeliveryProcessing( node, value )

	if value == kSkipValue or value == nil then
		return
	end
	
	if value == kScheduleRequestValue then
		node_addToScheduleQueue( node )
		return
	end
	
	assert( not node._delayed, "Delayed nodes should not return values from their receive functions" )
	
	node_sendPackedValue( node, value )

end

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

local function node_deliverPackedValue( node, packedValue )

	-- Don't deliver to done nodes.

	if nodeDAG_nodeIsDone( node ) then
		return
	end
	
	-- Don't deliver to nodes that are queued to close since we've been told
	-- that they are no longer interesting.
	
	if WFCloseSupport_hasBeenOrWillBeClosed( node ) then
		return
	end
	
	-- We don't pcall here because we expect the delivery queue to handle the
	-- case where delivery fails.

	local resultValue
	
	-- Dispatch based on the type of value to deliver.
	
	if packedValue == kDoneValue then
		resultValue = node:_receiveDone( not nodeDAG_nodeHasPredecessor( node ) )
	elseif KSRPackedValues_isError( packedValue ) then
		resultValue = node:_receiveError( packedValue )
	else
		resultValue = node:_receiveData( packedValue )
	end
	
	-- Process the results of the delivery.
	
	node_finishDeliveryProcessing( node, resultValue )

end

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

local function resendPackedValueForThisCycleFromTo( source, sink )

	if nodeDAG_nodeIsDone( source ) then
		node_deliverPackedValue( sink, kDoneValue)
		return
	end
	
	if not valuesSentThisCycle then
		return
	end
	
	local sentValue = valuesSentThisCycle[ source ]
	
	if not sentValue then
		return
	end
	
	-- FIXME [mhamburg 2011-03-11]: Could we avoid error handling logic issues
	-- here by instead just putting the sent value on the delivery queue?
	
	node_deliverPackedValue( sink, sentValue )
	
end

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

local function node_deliveryFailed( node, packedValue, errorMessage )

	debug.print( "Delivery failed for node: ", node, packedValue, errorMessage,
				 "\nin event processor "
					.. tostring( require 'WFCurrentThread'.DEBUG_getThreadName() ),
				 "\n\nvia " .. tostring( traceback() ) .. "\n\n" )
	
	-- FIXME [mhamburg 2011-05-05]: What if the node is a delayed node? Can
	-- we say those just don't fail?
	
	-- FIXME [mhamburg 2011-05-05]: Can we somehow delay this until it's time
	-- for the node to run so that it doesn't get a chance to fire again this
	-- cycle if it has multiple inputs?
	
	pcall( node_sendPackedValue, node, KSRPackedValues_error( errorMessage ) )
		-- The pcall just represents paranoia around not bringing down the
		-- system when we're in error handling code.

end

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

local prevDeliveryNode = nil
local prevDeliveryValue = nil
	-- The last node for which we attempted delivery. Used in error handling.

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

-- Run the delivery queue as far as we can until we encounter an error or
-- finish the queue. Record the node being currently processed as we go so
-- that we can clean up if need be. Note that this logic is not re-entrant.

local function runDeliveryQueue_core()

	for node, packedValue in deliveryQueue_popNodePackedValue do
		prevDeliveryNode = node
		prevDeliveryValue = packedValue
		node_deliverPackedValue( node, packedValue )
	end
	
	prevDeliveryNode = nil
	prevDeliveryValue = nil

end

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

-- Call runDeliveryQueue_core repeatedly cleaning up from errors each time.
-- When it returns without error, the queue must be done.

local function runDeliveryQueue()

	while true do
		local success, message = pcall( runDeliveryQueue_core )
		if success then
			return
		end
		if prevDeliveryNode then
			node_deliveryFailed( prevDeliveryNode, prevDeliveryValue, message )
		end
		prevDeliveryNode = nil
		prevDeliveryValue = nil
	end

end

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

local function node_runFailed( node, errorMessage )

	debug.print( "Run failed for node: ", node, errorMessage )
	
	-- Send the error message on as the value.

	pcall( node_sendPackedValue, node, KSRPackedValues_error( errorMessage ) )

end

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

local function runScheduledQueue()

	while true do
		runDeliveryQueue()
		local node = nextScheduledNode()
		if not node then
			break
		end
		local success, value = pcall( node._run, node )
		if success then
			if value then
				node_sendPackedValue( node, value )
			end
		else
			node_runFailed( node, value )
		end
	end
	
	valuesSentThisCycle = nil
		-- The cycle ends when we finish running all of the scheduled nodes

end

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

local nextTickNodePackedValuePairs = { }

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

local sendingNode = nil
local sendingValue = nil
	-- Used for error handling
	
local didSend = false

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

local function sendDelayedNodePackedValuePairs_core( nodePackedValuePairs )

	for node, packedValue in pairs( nodePackedValuePairs ) do
		nodePackedValuePairs[ node ] = nil
		sendingNode = node
		sendingValue = packedValue
		if packedValue then
			didSend = true
			node_sendPackedValue( node, packedValue )
		end
	end
	
	sendingNode = nil
	sendingValue = nil

end

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

local function node_delayedSendFailed( node, packedValue, errorMessage )

	debug.print( "Delayed send failed for node: ", errorMessage )

	-- FIXME [mhamburg 2011-02-15]

end

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

local function sendDelayedNodePackedValuePairs( nodePackedValuePairs )

	-- GRUNGE [scouten 2011-03-11]: Seems like this batching-pcall loop metaphor
	-- appears frequently. Opportunity for consolidation of code?
	--
	-- [mhamburg 2011-03-11]: Maybe, but it gets tricky to do so without generating
	-- closure creation overhead.
	
	didSend = false

	while true do
	
		local success, message =
			pcall( sendDelayedNodePackedValuePairs_core, nodePackedValuePairs )

		if success then
			break
		end

		-- assert( false, "Send delayed queue experienced failure: " .. tostring( message ) )
		
		node_delayedSendFailed( sendingNode, sendingValue, message )

		sendingNode = nil
		sendingValue = nil

	end
	
	return didSend

end

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

local function node_sendDelayedPackedValue( accumulator, node, packedValue )
	accumulator[ node ] = packedValue
end

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

local function node_sendDoneOnNextTick( node )
	node_sendDelayedPackedValue( nextTickNodePackedValuePairs, node, kDoneValue )
end

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

local runningScheduledAndNextTick = false

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

local function runScheduledAndNextTick()

	if runningScheduledAndNextTick then
		return
	end
	
	runningScheduledAndNextTick = true

	repeat
		runScheduledQueue()
		local didSend = sendDelayedNodePackedValuePairs( nextTickNodePackedValuePairs )
	until not didSend
	
	runningScheduledAndNextTick = false

end

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

-- We avoid reentrancy on the code to run the input queue since once it's
-- running, it will run until drained.

local runningInputQueue = false

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

-- Flush values out of the input queue running the scheduled and next tick
-- actions between them. Once the input queue goes empty, send the stable
-- values if any. We are done when everything except the idle values are done.

local function runInputQueue()

	if runningInputQueue then
		return
	end

	runningInputQueue = true

	while true do

		runScheduledAndNextTick() -- Generally does nothing at first.

		local node, packedValue = inputQueue_popNodePackedValue()
		
		-- If we don't have a node, then we've exhausted the input. Send out
		-- any stable values.
		
		if node then
		
			node_sendPackedValue( node, packedValue )
				-- FIXME [mhamburg 2011-02-02]: We should ensure that this doesn't result in errors
		
		else
		
			-- The input queue is empty, so send the stable values. If those send
			-- anything, it's possible that we will have more work to do and hence
			-- we cycle around again.
			
			break
			
		end
		
	end
	
	runningInputQueue = false

end


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

-- @mark Node allocation and initialization

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

local function allocNode( strongSuccessor )

	local node = nodeDAG_allocNode( strongSuccessor )
	
	if strongSuccessor then
		protectStrongSuccessorFromAutoClose( strongSuccessor )
	end
	
	return node

end

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

local function initNodeWithMetatable( node, nodeKind, mt, ... )

	local sourceCount = select( '#', ... )
	
	for i = 1, sourceCount do
		local source = select( i, ... )
		if source then
			assertNode( source )
		end
	end
	
	setmetatable( node, mt )
	
	local isBase = node._delayed
	
	if node._delayed then
		isBase = true
	end

	nodeDAG_initNode( node, nodeKind, isBase, ... )
	
	-- Queue this node up to see whether we fail to connect to it (or otherwise
	-- protect it).
	
	if AUTO_CLOSE_INITIAL_NODES then
		checkForAutoClose( node, 'initial' )
	end
	
	rawset( node, 'private:debugHook', false )
	
	-- If we've already sent values this cycle, then we need to send to
	-- the newly created node.
	
	-- FIXME [mhamburg 2011-03-11]: If any of these resends fail, we should
	-- potentially tear the node back down. Or we need to handle this with
	-- appropriate pcall logic around the delivery.
	
	for i = 1, sourceCount do
		local source = select( i, ... )
		if source then
			resendPackedValueForThisCycleFromTo( source, node )
		end
	end
	
	-- Either we aren't in mid-cycle in which case no values will have been
	-- delivered or we are in which case we will pick up the need to process
	-- this node as part of the cycle logic.
	
	return node

end

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

-- @mark Link adjustment

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

local function removeLink( pred, succ )

	assertNode( pred, "removeLink: predecessor" )
	assertNode( succ, "removeLink: successor" )
	
	return nodes_removeLink( pred, succ )
	
end

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

local function addLink( source, sink )

	-- Check that we've got nodes
	
	assertNode( source, "addLink: source" )
	assertNode( sink, "addLink: sink" )
	
	nodes_addLink( source, sink )

	-- FIXME [mhamburg 2011-03-11]: If this fails, we should potentially tear the
	-- link back down. Or we need to handle this with appropriate pcall logic around the delivery.

	resendPackedValueForThisCycleFromTo( source, sink )

end


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

-- @mark Debugging

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

local function eventNodeToString( node )

	return "KSReactive: " .. getNodeKind( node )

end


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

-- @mark closing method support

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

local addAutoCloseTarget

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

local function makeClosingMethod( rawMethod )

	if AUTO_CLOSE then
		return rawMethod
	else
		return function( self, ... )
			local result = rawMethod( self, ... )
			addAutoCloseTarget( result, self )
			return result
		end
	end

end

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

if AUTO_CLOSE then

	addAutoCloseTarget = function() end

else

	addAutoCloseTarget = function( self, target )
		self:addCloseTarget( target )
	end
	
end


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

-- @mark Event nodes

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

local EventNodeMethods = {

	[ 'private:closeWhenDone' ] = true

}

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

local populateEventNodeMethodsMT = makeMetatable_populateFrom( EventNodeMethods )

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

local function makeEventNodeMetatableForMethods( methods )

	setmetatable( methods, populateEventNodeMethodsMT )
	
	local mt = {
		__index = methods,
		__tostring = eventNodeToString,
		__metatable = "< reactive event node metatable >"
	}
	
	return mt

end

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

local function makeInitEventNodeForMethods( nodeKind )

	WFArguments_check( 'string' )

	return function ( methods )
	
		WFArguments_check( 'table' )
	
		local mt = makeEventNodeMetatableForMethods( methods )
		
		return function( node, ... )
			return initNodeWithMetatable( node, nodeKind, mt, ... )
		end
		
	end

end

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

EventNodeMethods.isDone = nodeDAG_nodeIsDone

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

function EventNodeMethods:close()

	-- If we're already marked as done, then we should have severed links and
	-- sent the done value, so there is no other work to do.

	if not nodeDAG_nodeIsDone( self ) then

		inputQueue_pushNodePackedValue( self, kDoneValue )

		runInputQueue()

	end

end

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

WFCloseSupport.addCloseSupportMethods( EventNodeMethods )
	-- This will wrap the preceding close logic.

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

function EventNodeMethods:maybeAutoClose( case )

	if not node_hasSuccessor( self ) then
		if case == 'initial' then
			local tback = rawget( self, 'private:traceback' )
			if tback then
				debug.print( "closing node that was never used. created @ " .. tostring( tback ) )
			else
				debug.print( "closing node that was never used" )
			end
		end
		self:close()
	end

end

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

local owned2owner = makeFullyWeakTable()

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

function EventNodeMethods:setOwnedBy( owner, optKey, optStorage )

	WFArguments_check( '_', 'table', '?string', '?table' )
	
	local prevOwner = owned2owner[ self ]
	
	if not prevOwner then
	
		-- first owner
		
	elseif prevOwner ~= owner then
	
		error( "Attempt to change owner for node" )
		
	else
	
		assert( not optKey, "Cannot set owner repeatedly when using a key" )
		assert( not optStorage, "Cannot set owner repeatedly when using storage" )
		
		return self
	
	end
	
	if WFCloseSupport_hasBeenOrWillBeClosed( owner ) then
		debug.print( "Attempted to attach a node to a closed owner" )
		WFIdle.doASAP( function() self:close() end )
			-- Close it ASAP. We don't close right away because we need
			-- to return self, but we just established that it should be closed.
		return self
	end
	
	protectFromAutoClose( self ) -- no auto closure since we're owned
	
	local storage
	
	if optStorage then
		storage = optStorage
	elseif not optKey then
		storage = rawget( owner, 'private:keyOwnedSet' )
		if not storage then
			storage = { }
			rawset( owner, 'private:keyOwnedSet', storage )
		end
	else
		storage = owner
	end
	
	if optKey then
		local oldValue = storage[ optKey ]
		if oldValue == self then
			-- do nothing (this might have gotten set as part of initialization)
			oldValue = nil
		end
		storage[ optKey ] = self
		owner:addCloseTarget( self )
		owned2owner[ self ] = owner
		if oldValue then
			oldValue:close()
			-- Don't bother removing as a close target
		end
	else
		storage[ self ] = true
		owner:addCloseTarget( self )
		owned2owner[ self ] = owner
	end
	
	return self

end

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

EventNodeMethods._delayed = false

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

function EventNodeMethods:_receiveData( packedValue )

	return packedValue

end

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

function EventNodeMethods:_receiveError( errorValue )

	return errorValue

end

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

function EventNodeMethods:_receiveDone( allSourcesGone )

	if allSourcesGone then
		return kDoneValue
	end

end

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

-- @mark Behavior Nodes

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

local behaviors = makeFullyWeakTable()
	-- A set to track all of the behaviors for type testing

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

local BehaviorNodeMethods = setmetatable( { }, { __index = EventNodeMethods } )

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

BehaviorNodeMethods[ 'private:closeWhenDone' ] = false
	-- Behavior nodes do not close by default when done.
	-- FIXME [mhamburg 2011-04-26]: The behavior prior to today was that
	-- behaviors actually ate the done signal which seems overly extreme.
	-- So, now we transmit done but eat close. Is this what we want to do?

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

local populateBehaviorNodeMethodsMT = makeMetatable_populateFrom( BehaviorNodeMethods )

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

local function makeBehaviorNodeMetatableForMethods( methods )

	setmetatable( methods, populateBehaviorNodeMethodsMT )
	
	local mt = {
		__index = methods,
		__tostring = eventNodeToString,
		__metatable = "< reactive behavior node metatable >"
	}
	
	return mt

end

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

local function makeInitBehaviorNodeForMethods( nodeKind )

	WFArguments_check( 'string' )

	return function( methods )
	
		WFArguments_check( 'table' )
	
		local mt = makeBehaviorNodeMetatableForMethods( methods )
		
		return function( node, initialPackedValue, ... )
			behaviors[ node ] = node
			rawset( node, 'private:behaviorCurrentValue', initialPackedValue )
			return initNodeWithMetatable( node, nodeKind, mt, ... )
		end
		
	end

end

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

local function BehaviorNode_packedValueNow( self )

	return rawget( self, 'private:behaviorCurrentValue' )

end

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

local function BehaviorNode_valueNow( self )

	return KSRPackedValues_unpack( BehaviorNode_packedValueNow( self ) )

end

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

BehaviorNodeMethods.valueNow = BehaviorNode_valueNow

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

function BehaviorNodeMethods:assertValueNow( value, message )

	assert( select( '#',  BehaviorNode_valueNow( self ) ) <= 1,
		"Too many values to match with assertValueNow" )

	assert( BehaviorNode_valueNow( self ) == value, message )

end

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

local function BehaviorNode_isErrorNow( self )

	return KSRPackedValues_isError( BehaviorNode_packedValueNow( self ) )

end

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

BehaviorNodeMethods.isErrorNow = BehaviorNode_isErrorNow

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

local function BehaviorNode_setPackedValue( self, newPackedValue )

	local oldPackedValue = rawget( self, 'private:behaviorCurrentValue' )
	
	if KSRPackedValues_testEquals( oldPackedValue, newPackedValue ) then
		return
	end
	
	rawset( self, 'private:behaviorCurrentValue', newPackedValue )
	
	return newPackedValue

end

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

function BehaviorNodeMethods:_receiveDone( allSourcesGone )
	-- We still have a value, but we're done changing so let the rest of
	-- the world know.
	if allSourcesGone then
		return kDoneValue
	end
end

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

local function isEventNode( v )

	-- Note that nodeDAG_isNode will report true even if v is closed. This is
	-- important because what we often care about is whether it has torn down
	-- the event graph for that node not whether it is externally closed.
	
	-- FIXME [mhamburg 2011-04-26]: This used to report false if the node
	-- was done but it's still an event node in that case. Does this change
	-- anything?
	
	local vIsEventNode = isNode( v )
		and not WFCloseSupport_hasBeenOrWillBeClosed( v )
	
	local vIsBehavior = vIsEventNode and behaviors[ v ]
	
	return vIsEventNode, vIsBehavior

end

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

local function isBehavior( v )

	local isEventNode, isBehavior = isEventNode( v )
	
	return isEventNode and isBehavior

end

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

-- @mark Debug hooks

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

local function formatValues( ... )

	local n = select( '#', ... )
	
	if n == 0 then
		return '< none >'
	end
	
	if n == 1 then
		return tostring( ( ... ) )
	end
	
	if n == 2 then
		local x, y = ...
		return string_format( "%s, %s", tostring( x ), tostring( y ) )
	end

	if n == 3 then
		local x, y, z = ...
		return string_format( "%s, %s, %s", tostring( x ), tostring( y ), tostring( z ) )
	end

	local x, y, z = ...
	return string_format( "%s, %s, %s,...", tostring( x ), tostring( y ), tostring( z ) )

end

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

function EventNodeMethods:setDebugHook( hookFn, hookName )

	if not hookFn then
		rawset( self, 'private:debugHook', false )
		return self
	end
	
	if type( hookFn ) == 'string' then
		assert( hookName == nil, "When we get a string as the first parameter to setDebugHook, there should be no second parameter." )
		hookName = hookFn
		hookFn = debug.print
	end

	rawset( self, 'private:debugHook', function( sender, packedValue )
	
		local nameStr
		
		if hookName then
			assert( type( hookName ) == 'string', "Hook name should be a string" )
			local nameFmtStr = behaviors[ sender ] and "B< %s : %s >-> " or "E< %s : %s >-> "
			nameStr = string_format( nameFmtStr, hookName, tostring( sender ) )
		else
			local nameFmtStr = behaviors[ sender ] and "B< %s >-> " or "E< %s >-> "
			nameStr = string_format( nameFmtStr, tostring( sender ) )
		end

		if packedValue == kDoneValue then
			hookFn( string_format( "%s DONE", nameStr ) )
			return
		end
		
		local isError, message = KSRPackedValues_isError( packedValue )
					
		if isError  then
			hookFn( string_format( "%s ERROR: %s", nameStr, tostring( message ) ) )
			return
		end
		
		hookFn( string_format( "%s DATA: %s", nameStr, formatValues( KSRPackedValues.unpack( packedValue ) ) ) )
		
	end )
	
	return self

end


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

-- @mark constantB, constantErrorB

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

-- FIXME [mhamburg 2011-05-04]: I was preventing constant nodes from closing
-- or being close targets but this interferes with optimizing things down
-- to constant nodes. The result of removing this is that we can't memoize
-- them but that also fought with using them as optimization points. We could
-- perhaps memoize just the public entry point but even that gets risky.

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

local constantB_initNode = makeInitBehaviorNodeForMethods 'constantB' {

	_receiveData = function()
		error( "Constant nodes should not receive values" )
	end,
	
	_receiveError = function()
		error( "Constant nodes should not receive error values" )
	end,
	
}

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

local constantB_packedValue = function( packedValue )

	local node = constantB_initNode( allocNode( false ), packedValue )
	
	disconnectNodeAndMarkAsDone( node )
		-- A constant node should not generate any permanent successors
		-- since it won't ever change its value.
		
	protectFromAutoClose( node )
	
	return node

end

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

local function constantB( ... )

	return constantB_packedValue( KSRPackedValues_pack( ... ) )

end

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

local function constantErrorB( ... )

	return constantB_packedValue( KSRPackedValues_error( ... ) )

end


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

local function EventNode_useDonePathForSuccessors( self )

	if nodeDAG_nodeIsDone( self ) then
		if not self[ 'private:closeWhenDone' ] and WFCloseSupport_hasBeenOrWillBeClosed( self ) then
			error( "Reactive node has already been closed: " .. tostring( self ) )
		end
		return true
	end
	
	return false

end

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

-- @mark holdB

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

local holdB_initNode = makeInitBehaviorNodeForMethods 'holdB' {

	_receiveData = BehaviorNode_setPackedValue,
	
	_receiveError = BehaviorNode_setPackedValue,
		
}

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

local function EventNode_holdB_packed( self, packedInitialValue, closesSelf )

	if EventNode_useDonePathForSuccessors( self ) then
		return constantB_packedValue( packedInitialValue )
	end

	local holdB = holdB_initNode( allocNode( false ), packedInitialValue, self )
	
	if closesSelf then
		addAutoCloseTarget( holdB, self )
	end
	
	return holdB

end

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

function EventNodeMethods:holdB( ... )
	
	return EventNode_holdB_packed( self, KSRPackedValues_pack( ... ), false )

end
	
--------------------------------------------------------------------------------

function EventNodeMethods:closingHoldB( ... )
	
	return EventNode_holdB_packed( self, KSRPackedValues_pack( ... ), true )

end
	
--------------------------------------------------------------------------------

function EventNodeMethods:holdValueNowB( b )

	assert( behaviors[ b ], "Argument is not a behavior." )
	
	return EventNode_holdB_packed( self, BehaviorNode_packedValueNow( b ), false )

end
	
--------------------------------------------------------------------------------

function EventNodeMethods:closingHoldValueNowB( b )
	
	assert( behaviors[ b ], "Argument is not a behavior." )
	
	return EventNode_holdB_packed( self, BehaviorNode_packedValueNow( b ), true )

end
	

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

-- @mark zeroE

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

-- PERF [mhamburg 2011-04-27]? We could optimize a variety of cases to
-- return zeroE if they are already done when constructed.

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

local zeroE -- forward declaration

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

do

	--------------------------------------------------------------------------------
	
	local function zeroE_noInputs()
		debug.print( "zeroE should never receive data; eating value" )
	end
	
	--------------------------------------------------------------------------------
	
	local zeroE_initNode = makeInitEventNodeForMethods 'zeroE' {
		_receiveData = zeroE_noInputs,
		_receiveError = zeroE_noInputs,
		_receiveDone = zeroE_noInputs,
	}
	
	--------------------------------------------------------------------------------
	
	zeroE = function()
	
		local node = zeroE_initNode( allocNode( false ) )
		
		disconnectNodeAndMarkAsDone( node )
		
		assert( nodeDAG_nodeIsDone( node ), "Zero node is not done" )
		
		protectFromAutoClose( node )
			-- We don't need to auto close zeroE nodes since they have neither
			-- predecessors nor successors.
		
		return node
	
	end
	
	--------------------------------------------------------------------------------

end


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

-- @mark filterE, closingFilterE

--------------------------------------------------------------------------------
	
function EventNodeMethods:filterE( predicate )

	WFArguments_check( '_', 'callable' )
	
	return self:mapE( function( ... )
		if predicate( ... ) then
			return kSameValue
		else
			return kSkipValue
		end
	end )

end

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

EventNodeMethods.closingFilterE = makeClosingMethod( EventNodeMethods.filterE )


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

-- @mark skipErrorsE

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

do

	--------------------------------------------------------------------------------
	
	local skipErrorsE_initNode = makeInitEventNodeForMethods 'skipErrorE' {
	
		_receiveError = function() end
			-- Just eat error values
	
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:skipErrorsE()
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
	
		return skipErrorsE_initNode( allocNode( false ), self )
	
	end

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

end


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

-- @mark mapE, mapB

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

do

	--------------------------------------------------------------------------------
	
	local function mapE_nodeReceivePackedValue_finish( oldPackedValue, ... )
	
		local n = select( '#', ... )
		
		if ( not ALLOW_EMPTY_TUPLES_FROM_MAP ) and ( n == 0 ) then
			return kSkipValue
		end
		
		if n == 1 then
			local v = ( ... )
			if kSameValue == v then
				return oldPackedValue
			elseif kSkipValue == v then
				return kSkipValue
			elseif kDoneValue == v then
				return kDoneValue
			else
				return KSRPackedValues_packOne( v )
			end
		end
		
		return KSRPackedValues_pack( ... )
	
	end
	
	--------------------------------------------------------------------------------
	
	local mapE_initNode = makeInitEventNodeForMethods 'mapE' {
	
		_receiveData = function( self, packedValue )
		
			local fn = rawget( self, 'private:mapFunction' )
			
			return mapE_nodeReceivePackedValue_finish(
						packedValue,
						fn( KSRPackedValues_unpack( packedValue ) ) )
		
		end
	
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:mapE( fn )
	
		WFArguments_check( '_', 'callable' )
		
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
	
		local node = allocNode( false )
		
		rawset( node, 'private:mapFunction', fn )
		
		return mapE_initNode( node, self )
	
	end
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingMapE = makeClosingMethod( EventNodeMethods.mapE )
	
	--------------------------------------------------------------------------------
	
	local function mapB_computeInitialValue_finish(
			oldPackedValue,
			success,
			... )
			
		if not success then
		
			return KSRPackedValues_error( ( ... ) or "Error in behavior computation" )
			
		end
		
		return mapE_nodeReceivePackedValue_finish( oldPackedValue, ... )
			
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:mapB( fn )
	
		WFArguments_check( '_', 'callable' )

		local selfPackedValue = BehaviorNode_packedValueNow( self )
		
		local initialPackedValue
		
		if KSRPackedValues_isError( selfPackedValue ) then
			
			initialPackedValue = selfPackedValue
			
		else
		
			initialPackedValue = mapB_computeInitialValue_finish(
				selfPackedValue,
				pcall( fn, KSRPackedValues_unpack( selfPackedValue ) ) )
		
		end
					
		-- Default handling if the map took us to the done state.
					
		if initialPackedValue == kDoneValue then
			if ALLOW_EMPTY_TUPLES_FROM_MAP then
				return constantB_packedValue()
			else
				return constantB_packedValue( nil )
			end
		end
		
		-- We need to come up with a reasonable value in the skip case.
		
		if initialPackedValue == kSkipValue then
			if ALLOW_EMPTY_TUPLES_FROM_MAP then
				initialPackedValue = KSRPackedValues_pack( )
			else
				initialPackedValue = KSRPackedValues_pack( nil )
			end
		end
					
		if EventNode_useDonePathForSuccessors( self ) then
			return constantB_packedValue( initialPackedValue )
		end
		
		return EventNode_holdB_packed( self:mapE( fn ), initialPackedValue, true )
	
	end
	
	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingMapB = makeClosingMethod( BehaviorNodeMethods.mapB )
	
	--------------------------------------------------------------------------------

end


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

-- @mark rawMapE, rawMapB

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

do

	--------------------------------------------------------------------------------
	
	local function rawMapE_nodeReceivePackedValue_finish( oldPackedValue, ... )
	
		local n = select( '#', ... )
		
		-- In the raw case, we can't process the nothing case so we exit early.
		
		if n == 0 then
			return kSkipValue
		end
		
		local status, firstValue = ...
		
		if status == kSameValue or status == 'same' then
			assert( n == 1, "Can only use same when returning no other values" )
			return oldPackedValue
		end
		
		if status == kSkipValue or status == 'skip' then
			assert( n == 1, "Can only use skip when returning no other values" )
			return kSkipValue
		end
		
		if status == kDoneValue or status == 'done' then
			assert( n == 1, "Can only use done when returning no other values" )
			return kDoneValue
		end
		
		if status == 'data' then
			if n == 2 then
				return KSRPackedValues_packOne( firstValue ) -- common case
			else
				return KSRPackedValues_pack( select( 2, ... ) )
			end
		elseif status == 'error' then
			return KSRPackedValues_error( firstValue )
		else
			debug.print( "Bad status for rawMap: ", status )
		end
	
	end
	
	--------------------------------------------------------------------------------
	
	local function callRawMap( fn, packedValue )
	
		local isError, errorMessage = KSRPackedValues_isError( packedValue )
		
		if isError then
			return fn( 'error', errorMessage )
		else
			return fn( 'data', KSRPackedValues_unpack( packedValue ) )
		end
	
	end
	
	--------------------------------------------------------------------------------
	
	local function rawMapE_receiveNonDone( self, packedValue )
	
		return rawMapE_nodeReceivePackedValue_finish(
					packedValue,
					callRawMap( rawget( self, 'private:rawMapFunction' ), packedValue ) )
	
	end
	
	--------------------------------------------------------------------------------
	
	local rawMapE_initNode = makeInitEventNodeForMethods 'rawMapE' {
	
		_receiveData = rawMapE_receiveNonDone,
		_receiveError = rawMapE_receiveNonDone,
	
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:rawMapE( fn )
	
		WFArguments_check( '_', 'callable' )
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
	
		local node = allocNode( false )
		
		rawset( node, 'private:rawMapFunction', fn )
		
		return rawMapE_initNode( node, self )
	
	end
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingRawMapE = makeClosingMethod( EventNodeMethods.rawMapE )

	--------------------------------------------------------------------------------
	
	local function rawMapB_computeInitialValue_finish(
		oldPackedValue,
		success,
		... )

		if not success then
		
			return KSRPackedValues_error( ( ... ) or "Error in behavior computation" )
			
		else
		
			return rawMapE_nodeReceivePackedValue_finish( oldPackedValue, ... )
		
		end
		
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:rawMapB( fn )
	
		WFArguments_check( '_', 'callable' )
		
		local selfPackedValue = BehaviorNode_packedValueNow( self )
		
		local initialPackedValue = rawMapB_computeInitialValue_finish(
				selfPackedValue,
				pcall( callRawMap, fn, selfPackedValue ) )
					
		-- Default handling if the map took us to the done state.
					
		if initialPackedValue == kDoneValue then
			return constantB_packedValue()
		end
		
		-- We need to come up with a reasonable value in the skip case.
		
		if initialPackedValue == kSkipValue then
			initialPackedValue = KSRPackedValues_pack( )
		end
					
		if EventNode_useDonePathForSuccessors( self ) then
			return constantB_packedValue( initialPackedValue )
		end
		
		return EventNode_holdB_packed( self:rawMapE( fn ), initialPackedValue, true )
	
	end
	
	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingRawMapB = makeClosingMethod( BehaviorNodeMethods.rawMapB )

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

end


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

-- @mark mcallB, mcallSwitchB

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

do

	--------------------------------------------------------------------------------
	
	local defaultToMethodFunctionCacheWantValues = { }
	local defaultToMethodFunctionCacheWantBehaviors = { }
	
	local kNil = { }
	local kSame = { }

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

	local function methodFunction( wantBehaviors, methodName, ... )
	
		WFArguments_check( 'boolean', 'string' )
		
		local cacheCache = wantBehaviors
				and defaultToMethodFunctionCacheWantBehaviors
				 or defaultToMethodFunctionCacheWantValues
		
		local default = ( ... )
		local defaultKey = default
		
		if default == nil then
			if select( '#', ... ) == 0 then
				default = kSame
				defaultKey = kSame
			else
				defaultKey = kNil
			end
		end
		
		local cache = cacheCache[ defaultKey ]
		
		if not cache then
			cache = makeFullyWeakTable()
			cacheCache[ defaultKey ] = cache
		end
		
		local cached = cache[ methodName ]
		
		if cached then
			return cached
		end
		
		local result
		
		if wantBehaviors then
			if default == kSame then
				result = function( obj )
					if not obj then return constantB( obj ) end
					return obj[ methodName ]( obj )
				end
			else
				result = function( obj )
					if not obj then return constantB( default ) end
					return obj[ methodName ]( obj )
				end
			end
		else
			if default == kSame then
				result = function( obj )
					if not obj then return obj end
					return obj[ methodName ]( obj )
				end
			else
				result = function( obj )
					if not obj then return default end
					return obj[ methodName ]( obj )
				end
			end
		end
		
		cache[ methodName ] = result
		
		return result

	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:mcallB( methodName, ... )
		return self:mapB( methodFunction( false, methodName, ... ) )
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:closingMcallB( methodName, ... )
		return self:closingMapB( methodFunction( false, methodName, ... ) )
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:mcallSwitchB( methodName, ... )
		return self:mapSwitchB( methodFunction( true, methodName, ... ) )
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:closingMcallSwitchB( methodName, ... )
		return self:closingMapSwitchB( methodFunction( true, methodName, ... ) )
	end

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

end


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

-- @mark propertyB, closingPropertyB

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

do

	--------------------------------------------------------------------------------
	
	local defaultToPropertyFunctionCache = { }
	
	local kNil = { }
	local kSame = { }

	--------------------------------------------------------------------------------
	
	local function propertyFunction( propertyName, ... )
	
		local default = ( ... )
		local defaultKey = default
		
		if default == nil then
			if select( '#', ... ) == 0 then
				default = kSame
				defaultKey = kSame
			else
				defaultKey = kNil
			end
		end
		
		local cache = defaultToPropertyFunctionCache[ defaultKey ]
		
		if not cache then
			cache = makeFullyWeakTable()
			defaultToPropertyFunctionCache[ defaultKey ] = cache
		end
		
		local cached = cache[ propertyName ]
		
		if cached then
			return cached
		end
		
		local result
		
		if default == kSame then
			result = function( obj )
				if not obj then return obj end
				return obj[ propertyName ]
			end
		else
			result = function( obj )
				if not obj then return default end
				return obj[ propertyName ]
			end
		end
		
		cache[ propertyName ] = result
		
		return result
	
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:propertyB( propertyName, ... )
		WFArguments_check( '_', 'string' )
		return self:mapB( propertyFunction( propertyName, ... ) )
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingPropertyB = makeClosingMethod( BehaviorNodeMethods.propertyB )

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

end


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

-- @mark toBooleanB, closingToBooleanB

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

do

	--------------------------------------------------------------------------------
	
	local function toBoolean( v )
		return not not v
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:toBooleanB()
		return self:mapB( toBoolean )
	end

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

	BehaviorNodeMethods.closingToBooleanB = makeClosingMethod( BehaviorNodeMethods.toBooleanB )

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

end


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

-- @mark filterB, closingFilterB

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

do

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:filterB( fn, ... )
		local filteredE = self:filterE( fn )
		if self:isErrorNow() or fn( self:valueNow() ) then
			return filteredE:holdValueNowB( self )
		else
			return filteredE:holdB( ... )
		end
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingFilterB = makeClosingMethod( BehaviorNodeMethods.filterB )
	
	--------------------------------------------------------------------------------

end


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

-- @mark skipErrorsB

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

do

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:skipErrorsB( ... )
		local skippedE = self:skipErrorsE()
		if not self:isErrorNow() then
			return skippedE:closingHoldValueNowB( self )
		else
			return skippedE:closingHoldB( ... )
		end
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingSkipErrorsB =
		makeClosingMethod( BehaviorNodeMethods.skipErrorsB )

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

end


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

-- @mark Ownership support

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

local function checkOwner( owner )

	local towner = type( owner )
	
	assert( towner == 'table' or towner == 'userdata',
		"Owner is neither table nor userdata." )
	
	assert( not WFCloseSupport_hasBeenOrWillBeClosed( owner ),
		"Owner has already been closed." )
		
end

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

local function ownerAttachOwned( owner, owned )

	local owner_type = type( owner )
	
	if owner_type == 'table' then
	
		owner:addCloseTarget( owned )
		
		rawset( owner, owned, owned )
	
	elseif owner_type == 'userdata' then
	
		owner:attachOwned( owned )
		
	else
	
		error( "Bad type for owning object: " .. tostring( owner_type ) )
	
	end
	
	protectFromAutoClose( owned )

end

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

--[[

local function makeOwnedMethod( method )

	return function( self, owner, ... )
	
		checkOwner( owner )
		
		local result = method( self, ... )
		
		ownerAttachOwned( owner, result )
		
		return result
	
	end

end

]]


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

-- @mark relayE, relayB

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

do

	--------------------------------------------------------------------------------
	
	-- We provide support for hanging a set of auto close relays off of a behavior
	-- or event such that when the last relay closes we will close the behavior.
	-- This will probably get replaced with a check for whether we have a strong
	-- dependent on a node.
	
	local function setupAutoCloseRelay( base, relay )
	
		-- FIXME_AUTO_CLOSE: This should be handled by the auto close logic?
	
		-- Add the result to the relay set for the behavior
		
		local relaySet = rawget( base, 'private:relaySet' )
		
		if not relaySet then
			relaySet = makeFullyWeakTable()
			rawset( base, 'private:relaySet', relaySet )
		end
		
		relaySet[ relay ] = relay
		
		-- When we close the relay, we clear it from the relay set and if it was
		-- the last item in the set, we close the source behavior.
		
		relay:addCloseAction( function()
			relaySet[ relay ] = nil
			if next( relaySet ) == nil then
				base:close()
			end
		end )
		
		-- NOTE: We do not use mustClose on these because that will result on
		-- the finalizer running at an arbitrary time which could close the
		-- underlying relayed object just as we were reviving it from a weak
		-- cache. [mhamburg 2011-08-24]
		
		return relay
	
	end
	
	--------------------------------------------------------------------------------
	
	local relayE_initNode = makeInitEventNodeForMethods 'relayE' { }
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:relayE()
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
		
		return relayE_initNode( allocNode( false ), self )
	
	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:autoCloseRelayE()
	
		-- If the node has already been closed, then closing this node won't close
		-- it again. We could make this an error if it weren't for the fact that
		-- event nodes currently close themselves when they reach the done point.
	
		if WFCloseSupport_hasBeenOrWillBeClosed( self ) then
			return zeroE()
		end
	
		return setupAutoCloseRelay(
					self,
					relayE_initNode( allocNode( false ), self ) )
	
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:relayB()
	
		return EventNode_holdB_packed(
					self,
					BehaviorNode_packedValueNow( self ),
					false )
	
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:autoCloseRelayB()
	
		return setupAutoCloseRelay(
					self,
					self:relayB() )
	
	end
	
	--------------------------------------------------------------------------------
	
	--[[
	
	function EventNodeMethods:noCloseRelayE()
	
		return self:relayE():bounceClosure()
	
	end
	
	]]

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:noCloseRelayB()
	
		return self:relayB():bounceClosure()
	
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:relayForAutoCloseB()
	
		return self
		
	end

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:closingObjectRelayB()
	
		local closer = {}
		WFCloseSupport.addCloseSupportMethods( closer )
		 
		local resultB = self:closingRawMapB( function( status, nextObject, ... )
			local relay = nextObject
			if type( nextObject ) == 'table' and nextObject.relayObject then
				relay = nextObject:relayObject()
				closer:setOwned {
					currentValue = relay,
				}
			else
				local oldValue = closer.currentValue
				if oldValue and not WFCloseSupport.hasBeenOrWillBeClosed( oldValue ) and oldValue.close then
					oldValue:close()
				end
				closer:setOwned {
					currentValue = false,
				}
			end
			return status, relay, ...
		end )
		
		resultB:addCloseTarget( closer )
		
		return resultB
		
	end

	--------------------------------------------------------------------------------
	
	--[[
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.ownedRelayE = makeOwnedMethod( EventNodeMethods.relayE )

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

	EventNodeMethods.closingOwnedRelayE = makeClosingMethod( EventNodeMethods.ownedRelayE )
	
	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.ownedRelayB = makeOwnedMethod( BehaviorNodeMethods.relayB )

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

	BehaviorNodeMethods.closingOwnedRelayB = makeClosingMethod( BehaviorNodeMethods.ownedRelayB )
	
	--------------------------------------------------------------------------------
	
	]]
	
	--------------------------------------------------------------------------------

end


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

-- @mark onDataE, onErrorE, onDoneE, onDataB, ownedOnDataB, ownedOnDataE, ...

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

do

	--------------------------------------------------------------------------------
	
	local function onX_receiveDone( self )
	
		rawset( self, 'private:action', false )
		
		return kDoneValue
	
	end
	
	--------------------------------------------------------------------------------
	
	local onDataE_initNode = makeInitEventNodeForMethods 'onDataE' {
	
		_receiveData = function( self, packedValue )
			local actionFn = rawget( self, 'private:action' )
			dpcall( 'onDataE', actionFn, KSRPackedValues_unpack( packedValue ) )
			return packedValue
		end,
		
		_receiveDone = onX_receiveDone
	
	}
	
	--------------------------------------------------------------------------------
	
	local function EventNodeMethods_makeOnDataE( self, strongSuccessor, fn )
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
	
		local node = allocNode( strongSuccessor )
		
		rawset( node, 'private:action', fn )
		
		onDataE_initNode( node, self )
		
		return node

	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:onDataE( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return EventNodeMethods_makeOnDataE( self, true, fn )
	
	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:weakOnDataE( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return EventNodeMethods_makeOnDataE( self, false, fn )
	
	end
	
	--------------------------------------------------------------------------------
	
	local onErrorE_initNode = makeInitEventNodeForMethods 'onErrorE' {
	
		_receiveError = function( self, errorValue )
			local actionFn = rawget( self, 'private:action' )
			local isError, errorMessage = KSRPackedValues_isError( errorValue )
			assert( isError, "Expected to be called with an error value" )
			dpcall( 'onErrorE', actionFn, errorMessage )
			return errorValue
		end,
	
		_receiveDone = onX_receiveDone
	
	}
	
	--------------------------------------------------------------------------------
	
	local function EventNodeMethods_makeOnErrorE( self, strongSuccessor, fn )
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end
	
		local node = allocNode( strongSuccessor )
		
		rawset( node, 'private:action', fn )
		
		onErrorE_initNode( node, self )
	
		return node
	
	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:onErrorE( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return EventNodeMethods_makeOnErrorE( self, true, fn )
	
	end
	
	--------------------------------------------------------------------------------
	
	local onDoneE_initNode = makeInitEventNodeForMethods 'onDoneE' {
	
		_receiveDone = function( self, allSourcesGone )
			if allSourcesGone then
				local actionFn = rawget( self, 'private:action' )
				rawset( self, 'private:action', false )
				dpcall( 'onDoneE', actionFn )
				return kDoneValue
			end
		end
	
	}
	
	--------------------------------------------------------------------------------
	
	local function EventNodeMethods_makeOnDoneE( self, fn )
	
		-- We will trigger on done even on nodes that are already done
	
		local node = allocNode( false )
		
		rawset( node, 'private:action', fn )
		
		onDoneE_initNode( node, self )
		
		return node
	
	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:weakOnDoneE( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return EventNodeMethods_makeOnDoneE( self, fn )
	
	end
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingWeakOnDoneE = makeClosingMethod( EventNodeMethods.weakOnDoneE )
	
	--------------------------------------------------------------------------------
	
	local function BehaviorNodeMethods_makeOnDataB( self, strongSuccessor, fn )
	
		local pv = BehaviorNode_packedValueNow( self )
		
		if not KSRPackedValues_isError( pv ) then
			pcall( fn, KSRPackedValues_unpack( pv ) )
		end
		
		local onDataE = EventNodeMethods_makeOnDataE( self, strongSuccessor, fn )
		
		local result = EventNode_holdB_packed( onDataE, pv, true )
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:onDataB( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return BehaviorNodeMethods_makeOnDataB( self, true, fn )
	
	end
	
	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:weakOnDataB( fn )
	
		WFArguments_check( '_', 'callable' )
		
		return BehaviorNodeMethods_makeOnDataB( self, false, fn )
	
	end
	
	--------------------------------------------------------------------------------

	function EventNodeMethods:ownedOnDataE( owner, fn )
	
		WFArguments_check( '_', '_', 'callable' )
		
		checkOwner( owner )
		
		local result = EventNodeMethods_makeOnDataE( self, false, fn )
		
		ownerAttachOwned( owner, result )
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingOwnedOnDataE = makeClosingMethod( EventNodeMethods.ownedOnDataE )

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

	function EventNodeMethods:ownedOnErrorE( owner, fn )
	
		WFArguments_check( '_', '_', 'callable' )
		
		checkOwner( owner )

		local result = EventNodeMethods_makeOnErrorE( self, false, fn )
		
		ownerAttachOwned( owner, result )
		
		return result
	
	end
	
	--------------------------------------------------------------------------------

	function EventNodeMethods:ownedOnDoneE( owner, fn )
	
		WFArguments_check( '_', '_', 'callable' )
		
		checkOwner( owner )

		local result = EventNodeMethods_makeOnDoneE( self, fn )
		
		ownerAttachOwned( owner, result )
		
		return result
	
	end
	
	--------------------------------------------------------------------------------

	function BehaviorNodeMethods:ownedOnDataB( owner, fn )
	
		WFArguments_check( '_', '_', 'callable' )
		
		checkOwner( owner )

		local result = BehaviorNodeMethods_makeOnDataB( self, false, fn )
		
		ownerAttachOwned( owner, result )
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingOwnedOnDataB
		= makeClosingMethod( BehaviorNodeMethods.ownedOnDataB )

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

end


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

-- @mark protectFromAutoCloseE, protectFromAutoCloseB

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

function EventNodeMethods:protectFromAutoCloseE()

	protectFromAutoClose( self )
	
	return self

end

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

function BehaviorNodeMethods:protectFromAutoCloseB()

	protectFromAutoClose( self )
	
	return self

end


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

-- @mark runUntilDone, autoCloseRunUntilDoneE, autoCloseRunUntilDoneB

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

do

	--------------------------------------------------------------------------------
	
	local runUntilDoneSet = { }
	
	function EventNodeMethods:runUntilDone()
	
		if node_isDone( self ) then
			return
		end
	
		local doneE
		
		local function doneFn()
			runUntilDoneSet[ doneE ] = nil
			self:close() -- We actually close because we may want to close behaviors
		end
		
		doneE = self:weakOnDoneE( doneFn ):protectFromAutoCloseE()
			-- We are explicitly keeping it alive in the set so we can go
			-- with weak.
			
		runUntilDoneSet[ doneE ] = true
	
	end

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

	function EventNodeMethods:autoCloseRunUntilDoneE()
	
		if AUTO_CLOSE then
			self:runUntilDone()
		end
		
		return self
	
	end
	
	--------------------------------------------------------------------------------

	function BehaviorNodeMethods:autoCloseRunUntilDoneB()
	
		if AUTO_CLOSE then
			self:runUntilDone()
		end
		
		return self
	
	end
	
	--------------------------------------------------------------------------------

end


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

-- @mark switchE, mapSwitchE, mapSwitchB

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

do

	--------------------------------------------------------------------------------
	
	local function switchE_output_receiveNonDone( self, packedValue )
		rawset( self, 'private:switchE_output_pendingValue', packedValue )
		return kScheduleRequestValue
	end
	
	--------------------------------------------------------------------------------
	
	local switchE_output_initNode = makeInitEventNodeForMethods 'switchE_output' {
	
		_receiveData = switchE_output_receiveNonDone,
		_receiveError = switchE_output_receiveNonDone,
		
		_receiveDone = function( self, allSourcesGone )
		
			-- Check to see whether the link stream just terminated
			-- and if so close it now (probably superfluous) if appropriate
			-- and unlink.
			
			local linkStream = rawget( self, 'private:switchE_output_linkStream' )
			
			if linkStream and nodeDAG_nodeIsDone( linkStream ) then
				rawset( self, 'private:switchE_output_linkStream', false )
				if not AUTO_CLOSE then
					linkStream:close()
				end
			end

			-- If this is the end, then terminate.
		
			if allSourcesGone then
				return kDoneValue
			end
			
		end,
		
		_run = function( self )
			local value = rawget( self, 'private:switchE_output_pendingValue' )
			rawset( self, 'private:switchE_output_pendingValue', false )
			if value then
				return value
			end
		end,
	
	}
	
	--------------------------------------------------------------------------------
	
	local function switchE_output_updateLinkStream( outputNode, newStream )
	
		local oldStream = rawget( outputNode, 'private:switchE_output_linkStream' )
		
		if oldStream == newStream then
			return
		end
		
		if oldStream then
			rawset( outputNode, 'private:switchE_output_linkStream', false )
			removeLink( oldStream, outputNode )
			if not AUTO_CLOSE then
				oldStream:close()
				-- Having closed the stream, we don't need to remove it as a close target
				-- which is what removeAutoCloseTarget does
				-- removeAutoCloseTarget( outputNode, oldStream )
			else
				-- remove auto close target does nothing when AUTO_CLOSE
				-- removeAutoCloseTarget( outputNode, oldStream )
			end
		end
		
		if newStream then
			addAutoCloseTarget( outputNode, newStream )
			addLink( newStream, outputNode )
			rawset( outputNode, 'private:switchE_output_linkStream', newStream )
		end
		
	end
	
	--------------------------------------------------------------------------------
	
	local switchE_input_initNode = makeInitEventNodeForMethods 'switchE_input' {
	
		_receiveData = function( self, packedValue )
		
			local outputNode = rawget( self, 'private:switchE_input_outputNode' )
			
			local newStream = KSRPackedValues_first( packedValue ) or false
				-- Note that we force nil to false.
			
			-- Don't switch for duplicate values.
			
			-- GRUNGE [mhamburg 2011-05-05]: Does this conflict with the logic
			-- that sends values from behaviors?
			
			if rawget( self, 'private:switchE_input_previousEventStream' ) == newStream then
				return
			end
			
			rawset( self, 'private:switchE_input_previousEventStream', newStream )
			
			-- If we don't have a new stream, then we tear things down.
			
			if not newStream then
				rawset( outputNode, 'private:switchE_output_pendingValue', false )
				switchE_output_updateLinkStream( outputNode, false )
				return
			end
			
			-- We need to set the value from the switch. If this is a behavior, we treat the current
			-- value as having fired (subject to getting updated later in this event cycle). If not,
			-- we specify no value and look to either the link operation or a later firing to set
			-- any value. In the former case, we need to schedule the output node to run since it
			-- may need to produce a value.
			
			local nsIsEventStream, nsIsBehavior = isEventNode( newStream )
			
			if not nsIsEventStream then
				debug.print( "switchE expects to receive event nodes; got ", tostring( newStream ) )
				return self:_receiveError( KSRPackedValues_error( "Expected to receive event node" ) )
			end
			
			if nsIsBehavior then
				rawset( outputNode, 'private:switchE_output_pendingValue', BehaviorNode_packedValueNow( newStream ) )
				node_addToScheduleQueue( outputNode )
			else
				rawset( outputNode, 'private:switchE_output_pendingValue', false )
			end
			
			-- Link to receive events from the stream if it isn't done. This
			-- also handles close ownership.
			
			local linkStream = newStream
			
			if nodeDAG_nodeIsDone( linkStream ) then
				linkStream = false
			end
			
			switchE_output_updateLinkStream( outputNode, linkStream )
			
		end,
		
		_receivedError = function( self, errorValue )
		
			local outputNode = rawget( self, 'private:switchE_input_outputNode' )
			
			rawset( self, 'private:switchE_input_previousEventStream', false )
			
			switchE_output_updateLinkStream( outputNode, false )
	
			rawset( outputNode, 'private:switchE_output_pendingValue', errorValue )
			
			node_addToScheduleQueue( outputNode )
		
		end,
	
	}
	
	--------------------------------------------------------------------------------
	
	-- Construct a switch. We can optionally have the switch close the streams
	-- it receives. We can also specify an initial stream that it holds ahead
	-- of receiving any events. We will not look to this stream for values the
	-- way we would if it were a behavior since we may not be in the midst of
	-- processing event logic. Any use of initial values would be in the context
	-- of behaviors and would be handled via a holdB as a later stage.
	
	local function EventNode_constructSwitchE( source, initialOutputStream )
	
		-- Initial output streams are only interesting if they haven't finished
		-- firing.
	
		if initialOutputStream and EventNode_useDonePathForSuccessors( initialOutputStream ) then
			initialOutputStream = false
		end
	
		initialOutputStream = initialOutputStream or false
		
		--  If the source is done and there is no initial output stream, then
		-- we will generate nothing.
	
		if EventNode_useDonePathForSuccessors( source ) and not initialOutputStream then
			return zeroE()
		end
		
		local inputNode = allocNode( false )
		local outputNode = allocNode( false )
		
		rawset( inputNode, 'private:switchE_input_previousEventStream', initialOutputStream )
			-- Debounce the input.
			
		rawset( inputNode, 'private:switchE_input_outputNode', outputNode )
		
		rawset( outputNode, 'private:switchE_output_pendingValue', false )
		rawset( outputNode, 'private:switchE_output_linkStream', false )
		
		-- Start hooking things up. We wait to hook up the source because we
		--  want the internal wiring in place before we allow events to
		-- come in from the source.
		
		switchE_input_initNode( inputNode )
		switchE_output_initNode( outputNode, inputNode )
		
		addAutoCloseTarget( outputNode, inputNode )
		
		-- If we have an initial stream to drive the output, hook it up now.
		-- Note that the addLink here may result in the output node receiving
		-- a value.
		
		if initialOutputStream then
			switchE_output_updateLinkStream( outputNode, initialOutputStream )
		end
		
		-- Now we can hook up the source.
		
		addLink( source, inputNode )

		return outputNode
	
	end
	
	--------------------------------------------------------------------------------
	
	--[[ switchE is no more...
	
	function EventNodeMethods:switchE()
	
		return EventNode_constructSwitchE( self, false, false )
			The middle false is no longer supported
	
	end
	
	]]
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:mapSwitchE( fnE )
	
		local mappedE = self:mapE( fnE )
		
		local switchedE = EventNode_constructSwitchE( mappedE, false )
		
		addAutoCloseTarget( switchedE, mappedE )
		
		return switchedE
	
	end
	
	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingMapSwitchE = makeClosingMethod( EventNodeMethods.mapSwitchE )
	
	--------------------------------------------------------------------------------
	
	local function BehaviorNode_computeInitialStreamAndValue( self, requireBehavior )
	
		-- The initial stream is the initial stream from which we get events
		-- for the output. The initial packed value is the initial output
		-- value. If nothing else were changing, our naive answer would be
		-- initialStream:holdB( initialPackedValue ).
		
		local initialStream, initialPackedValue
		
		local selfPackedValue = BehaviorNode_packedValueNow( self )
		
		if KSRPackedValues_isError( selfPackedValue ) then
			initialStream = false
			initialPackedValue = selfPackedValue
		elseif not selfPackedValue then -- FIXME [mhamburg 2011-05-05]: Allow nil?
			initialStream = false
			initialPackedValue = nil
		else
			local selfValue = KSRPackedValues_first( selfPackedValue )
			local selfValueIsEventNode, selfValueIsBehavior = isEventNode( selfValue )
			assertft( selfValueIsEventNode, "Value should have been an event node." )
			assertft( selfValueIsBehavior or not requireBehavior, "Value should have been a behavior." )
			initialStream = selfValue
			if selfValueIsBehavior then
				initialPackedValue = BehaviorNode_packedValueNow( selfValue )
			end
		end
		
		-- If the initial stream won't be firing, we don't want to assume ownership
		-- for it and we will be able to just use a constant behavior if it never
		-- fires.
		
		if initialStream and EventNode_useDonePathForSuccessors( initialStream ) then
			initialStream = false
		end
		
		return initialStream, initialPackedValue

	end
	
	--------------------------------------------------------------------------------
	
	local function BehaviorNode_constructSwitchB( self )
	
		-- The initial stream is the initial stream from which we get events
		-- for the output. The initial packed value is the initial output
		-- value. If nothing else were changing, our naive answer would be
		-- initialStream:holdB( initialPackedValue ).
		
		local initialStream, initialPackedValue =
			BehaviorNode_computeInitialStreamAndValue( self, true )
		
		-- The standard case. self isn't done, so we want to switch on the
		-- streams coming in from self. We utilize the behavior that switchE
		-- treats switching to a behavior as firing the current value of the
		-- behavior if the behavior doesn't fire in some other way during the
		-- cycle. We initialize the switch as listening to the initial stream
		-- if any as the source for its output so that we have a value even
		-- before self fires.
		
		if not EventNode_useDonePathForSuccessors( self ) then
			local switchE = EventNode_constructSwitchE( self, initialStream )
			return EventNode_holdB_packed( switchE, initialPackedValue, true )
		end
		
		-- If self is done but has a value that isn't finished, then we can
		-- listen to that value and optionally close it if we close our input.
		
		if initialStream then
			return EventNode_holdB_packed( initialStream, initialPackedValue, true )
		end
		
		-- Otherwise, we just have a constant node.
		
		return constantB_packedValue( initialPackedValue )

	end
	
	--------------------------------------------------------------------------------
	
	-- mapSwitchB maps a value to a behavior valued behavior and in turn to the value
	-- of the current behavior. It owns the intermediate behaviors and will close
	-- them when switching.
	
	function BehaviorNodeMethods:mapSwitchB( fnB )
	
		local mappedB = self:mapB( fnB )
		
		local switchedB = BehaviorNode_constructSwitchB( mappedB )
		
		addAutoCloseTarget( switchedB, mappedB )
		
		return switchedB
	
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingMapSwitchB
		= makeClosingMethod( BehaviorNodeMethods.mapSwitchB )
	
	--------------------------------------------------------------------------------
	
	--[[ UNUSED
	
	local function BehaviorNode_constructSwitchE( self, closesStreams )
	
		-- The initial stream is the initial stream from which we get events
		-- for the output. The initial packed value is the initial output
		-- value. If nothing else were changing, our naive answer would be
		-- initialStream:holdB( initialPackedValue ).
		
		local initialStream, _ = BehaviorNode_computeInitialStreamAndValue( self, false )
		
		-- The standard case. self isn't done, so we want to switch on the
		-- streams coming in from self. We utilize the behavior that switchE
		-- treats switching to a behavior as firing the current value of the
		-- behavior if the behavior doesn't fire in some other way during the
		-- cycle. We initialize the switch as listening to the initial stream
		-- if any as the source for its output so that we have a value even
		-- before self fires.
		
		if not EventNode_useDonePathForSuccessors( self ) then
			local switchE = EventNode_constructSwitchE( self, closesStreams, initialStream )
			return switchE
		end
		
		-- If self is done but has a value that isn't finished, then we can
		-- listen to that value and optionally close it if we close our input.
		
		if initialStream and not EventNode_useDonePathForSuccessors( initialStream ) then
			return initialStream
		end
		
		-- Otherwise, we just have a constant node.
		
		return zeroE()

	end
	
	]]
	
	--------------------------------------------------------------------------------
	
	--[[ switchE is no more...
	
	-- switchE for behaviors needs to support an initial event stream unlike
	-- switchE for event nodes
	
	function BehaviorNodeMethods:switchE()
	
		return BehaviorNode_constructSwitchE( self, false )
	
	end
	
	]]
	
	--------------------------------------------------------------------------------

end

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

-- @mark closingKnownMapB and closingKnownMapSwitchB

do
	local WFUnknown = require 'WFUnknown'

	--------------------------------------------------------------------------------
	
	local function anyAreUnknown( ... )
		for i = 1, select( '#', ... ) do
			if WFUnknown.isUnknown( select( i, ... ) ) then
				return true
			end
		end
		return false
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingKnownMapB = 	function( self, fn )
		return BehaviorNodeMethods.closingMapB( self, function( ... )
			if not anyAreUnknown( ... ) then
				return fn( ... )
			end
			return WFUnknown.value()
		end )
	end

	--------------------------------------------------------------------------------
	
	BehaviorNodeMethods.closingKnownMapSwitchB = function( self, fn )
		return BehaviorNodeMethods.closingMapSwitchB( self, function( ... )
			if not anyAreUnknown( ... ) then
				return fn( ... )
			end
			return constantB( WFUnknown.value() )
		end )
	end

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

end


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

-- @mark constantE

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

do

	--------------------------------------------------------------------------------
	
	local constantE_initNode = makeInitEventNodeForMethods 'constantE' {
	
		_receiveData = function( self )
		
			return rawget( self, 'private:constantE_value' )
		
		end
	
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:constantE( ... )
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end

		local node = allocNode( false )
		
		rawset( node, 'private:constantE_value', KSRPackedValues_pack( ... ) )
		
		return constantE_initNode( node, self )
	
	end

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

end


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

-- @mark takeE

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

do

	--------------------------------------------------------------------------------
	
	local function takeE_receive( self, packedValue )
		local counter = rawget( self, 'private:takeE_counter' )
		counter = counter - 1
		if counter <= 0 then
			node_sendDoneOnNextTick( self )
		else
			rawset( self, 'private:takeE_counter', counter )
		end
		return packedValue
	end
	
	--------------------------------------------------------------------------------
	
	local takeE_initNode = makeInitEventNodeForMethods 'takeE' {
		_receiveData = takeE_receive,
		_receiveError = takeE_receive
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:takeE( n )
	
		WFArguments_check( '_', 'number' )
	
		assert( 0 < n, "Expected positive numeric parameter for takeE." )
		
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end

		local node = allocNode( false )
		
		rawset( node, 'private:takeE_counter', n )
		
		return takeE_initNode( node, self )
	
	end

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

end


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

-- @mark endAfterErrorE

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

do

	--------------------------------------------------------------------------------
	
	local endAfterErrrorE_initNode = makeInitEventNodeForMethods 'endAfterErrorE' {
		_receiveError = function( self, packedValue )
			if rawget( self, 'private:endAfterErrorE_blocked' ) then
				return
			end
			node_sendDoneOnNextTick( self )
			rawset( self, 'private:endAfterErrorE_blocked', true )
			return packedValue
		end
	}
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:endAfterErrorE()
	
		WFArguments_check( '_' )
	
		if EventNode_useDonePathForSuccessors( self ) then
			return zeroE()
		end

		local node = allocNode( false )
		
		rawset( node, 'private:endAfterErrorE_blocked', false )
		
		return endAfterErrrorE_initNode( node, self )
	
	end

	--------------------------------------------------------------------------------
	
	EventNodeMethods.closingEndAfterErrorE =
		makeClosingMethod( EventNodeMethods.endAfterErrorE )

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

end


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

-- @mark accumulateE, accumulateB

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

function EventNodeMethods:accumulateE( fn, accumulator )

	WFArguments_check( '_', 'callable' )
	
	local function finish( ... )
		accumulator = ( ... )
		return ...
	end
	
	return self:mapE( function( ... )
		return finish( fn( accumulator, ... ) )
	end )

end

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

function EventNodeMethods:accumulateB( fn, accumulator, ... )

	local e = self:accumulateE( fn, accumulator )
		-- Generate an event stream through accumulation
	
	local r = e:holdB( accumulator, ... )
		-- Start with the initial accumulator
		
	addAutoCloseTarget( r, e )
		-- Closing the result closes the underlying event stream
	
	return r

end


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

-- @mark sourceE

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

local function sourceE_finishPost( self, packedValue )
	inputQueue_pushNodePackedValue( self, packedValue )
	runInputQueue()
	return self
end

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

local sourceE_initNode = makeInitEventNodeForMethods 'sourceE' {

	_receiveData = function()
		error( "Source nodes should not receive values" )
	end,
	
	post = function( self, ... )
		return sourceE_finishPost( self, KSRPackedValues_pack( ... ) )
	end,
	
	postError = function( self, message )
		return sourceE_finishPost( self, KSRPackedValues_error( message ) )
	end,
	
	postDone = function( self )
		return sourceE_finishPost( self, kDoneValue )
	end,
	
}

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

local function sourceE()

	local node = sourceE_initNode( allocNode( false ) )

	return node
	
end


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

-- @mark variableB

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

local function variableB_setPacked( self, packedValue )
	local setResult = BehaviorNode_setPackedValue( self, packedValue )
	if setResult ~= nil then
		inputQueue_pushNodePackedValue( self, setResult )
		runInputQueue()
	end
	return self
end

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

local variableB_initNode = makeInitBehaviorNodeForMethods 'variableB' {

	_receiveData = function()
		error( "Variable nodes should not receive values" )
	end,
	
	_receiveError = function()
		error( "Variable nodes should not receive error values" )
	end,
	
	set = function( self, ... )
		assert( not nodeDAG_nodeIsDone( self ), "Cannot set a variable that is done" )
		return variableB_setPacked( self, KSRPackedValues_pack( ... ) )
	end,
	
	setError = function( self, message )
		assert( not nodeDAG_nodeIsDone( self ), "Cannot set an error on a variable that is done" )
		return variableB_setPacked( self, KSRPackedValues_error( message ) )
	end,
	
	setDone = function( self )
		assert( not nodeDAG_nodeIsDone( self ), "Node is already done" )
		inputQueue_pushNodePackedValue( self, kDoneValue )
		runInputQueue()
		return self
	end,
	
	--[[ UNUSED
	
	bind = function( self, v )
	
		assert( not isRunning(), "We cannot rebind a variable while the network is running." )
	
		local oldBindSource = self[ keyBindSource ]
		
		if oldBindSource == v then
			return
		end

		-- We also allow for unbinding
	
		if v == nil then
			if oldBindSource then
				removeLink( oldBindSource, self )
				self[ keyBindSource ] = nil
				-- Go back to the standard methods
				self._receiveData = nil
				self.set = nil
			end
			return
		end
		
		-- Otherwise, this needs to be an event node
	
		local vIsEventNode, vIsBehavior = isEventNode( v )
		
		assert( vIsEventNode, "Expected a behavior to be passed to variableB:bind" )
		
		-- Rebind. Note that we don't unbind until we've successfully bound.
		
		-- Check that we've got nodes
		
		-- assertNode( v, "variableB:bind:addLink: predecessor" )
		-- assertNode( self, "variableB:bind:addLink: successor" )
	
		nodes_addLink( v, self )
		
		if oldBindSource then
			removeLink( oldBindSource, self )
		end
		
		self[ keyBindSource ] = v
		
		-- Pick up receive packed from the node methods and block set locally.
		
		self._receiveData = BehaviorNode_setPackedValue
		self._receiveError = BehaviorNode_setPackedValue
		self.set = variableB_setWhileBoundError
		
		-- Handle values from behaviors and event nodes that have already fired
		
		if vIsBehavior then
			variableB_setPacked( BehaviorNode_packedValueNow( v ) )
		end
		
		return self
	
	end
	
	]]

}

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

local function variableB( ... )

	local node = variableB_initNode( allocNode( false ), KSRPackedValues_pack( ... ) )
	
	return node

end


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

-- @mark mergeE, anyE

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

local anyE, closingAnyE -- forward declarations

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

do

	--------------------------------------------------------------------------------
	
	-- We build a structure in which the output node directly handles the left side
	-- of the merge while the input node handles the right side of the merge by
	-- setting the right side of the output node in response to any events and
	-- queueing the output node to run. We need two nodes because we have two
	-- different actions.
	
	--------------------------------------------------------------------------------
	
	local function merge2E_output_receiveNonDone( self, packedValue )
		rawset( self, 'private:merge2E_output_leftValue', packedValue )
		return kScheduleRequestValue
	end
	
	--------------------------------------------------------------------------------
	
	local merge2E_output_initNode = makeInitEventNodeForMethods 'merge2E_output' {
	
		_receiveData = merge2E_output_receiveNonDone,
		_receiveError = merge2E_output_receiveNonDone,
	
		_run = function( self )
			local leftValue = rawget( self, 'private:merge2E_output_leftValue' )
			local rightValue = rawget( self, 'private:merge2E_output_rightValue' )
			rawset( self, 'private:merge2E_output_leftValue', false )
			rawset( self, 'private:merge2E_output_rightValue', false )
			return leftValue or rightValue
		end
	
	}
	
	--------------------------------------------------------------------------------
	
	local function merge2E_input_receiveNonDone( self, packedValue )
		local outputNode = rawget( self, 'private:merge2E_input_outputNode' )
		rawset( outputNode, 'private:merge2E_output_rightValue', packedValue )
		node_addToScheduleQueue( outputNode )
			-- No values ever flow out directly
	end
	
	--------------------------------------------------------------------------------
	
	local merge2E_input_initNode = makeInitEventNodeForMethods 'merge2E_input' {
		
		_receiveData = merge2E_input_receiveNonDone,
		_receiveError = merge2E_input_receiveNonDone,
	
	}
	
	--------------------------------------------------------------------------------
	
	-- Construct a merge for two event streams. Handles nil for the streams and
	-- returns nil if both streams are nil or done.
	
	local function merge2E_imp( leftE, rightE )
	
		-- Nodes that are done have nothing to contribute
	
		if leftE and EventNode_useDonePathForSuccessors( leftE ) then
			leftE = nil
		end
	
		if rightE and EventNode_useDonePathForSuccessors( rightE ) then
			rightE = nil
		end
		
		-- If we are missing a stream, then return the other stream
	
		if not leftE then
			return rightE
		end
		
		if not rightE then
			return leftE
		end
		
		-- Construct an appropriate node
	
		local inputNode = allocNode( false )
		local outputNode = allocNode( false )
		
		rawset( inputNode, 'private:merge2E_input_outputNode', outputNode )
		
		rawset( outputNode, 'private:merge2E_output_leftValue', false )
		rawset( outputNode, 'private:merge2E_output_rightValue', false )
		
		merge2E_input_initNode( inputNode, rightE )
		
		merge2E_output_initNode( outputNode, leftE, inputNode )
		
		addAutoCloseTarget( outputNode, inputNode )
		
		return outputNode
	
	end
	
	--------------------------------------------------------------------------------
	
	-- Concatenate a sequence of streams using merge2E_imp in right associative
	-- fashion.
		
	local function mergeE_imp( ... )
	
		local n = select( '#', ... )
		
		if n == 0 then
			return nil
		end
		
		if n == 1 then
			return (...)
		end
		
		return merge2E_imp( (...), mergeE_imp( select( 2, ... ) ) )
	
	end
	
	--------------------------------------------------------------------------------
	
	anyE = function( ... )
	
		local mergedE = mergeE_imp( ... ) or zeroE()
		
		local resultE = mergedE:rawMapE( function() return 'data', true end )
			-- We want to map errors to true as well.
			
		-- Check to see whether mergedE is amongst the input nodes. If not, we
		-- want to close it when the result closes.
		
		local mergedFromInputE = WFVarArgs_findFirst(
				function( x ) return x == mergedE end,
				... )
				
		if not mergedFromInputE then
			addAutoCloseTarget( resultE, mergedE )
		end
		
		return resultE
	
	end
	
	--------------------------------------------------------------------------------
	
	function EventNodeMethods:mergeE( otherStreamE )
	
		return merge2E_imp( self, otherStreamE ) or zeroE()
	
	end

	--------------------------------------------------------------------------------
	
	local function addAutoCloseTarget_all( result, target, ... )
		addAutoCloseTarget( result, target )
		if ( ... ) then
			return addAutoCloseTarget_all( result, ... )
		end
	end
	
	closingAnyE = function( ... )
		local result = anyE( ... )
		addAutoCloseTarget_all( result, ... )
		return result
	end
	
	function EventNodeMethods:closingMergeE( ... )
		local result = self:mergeE( ... )
		addAutoCloseTarget_all( result, self, ... )
		return result
	end

end


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

local function liftIfNeededB( v )

	if behaviors[ v ] then
		return v
	end
	
	return constantB( v )

end


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

-- @mark callB

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

local callB -- forward

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

do

	--------------------------------------------------------------------------------
	
	local function callB_constructFunctionAndArguments( node, ... )
	
		local n = select( '#', ... )
		
		rawset( node, 'private:callB_unpackLimit', n )
		
		for i = 1, n do
			node[ i ] = liftIfNeededB( select( i, ... ) )
		end
	
	end
	
	--------------------------------------------------------------------------------
	
	local function callB_receiveNonDone()
		return kScheduleRequestValue
	end
	
	--------------------------------------------------------------------------------
	
	local function callB_evalToPacked( self )
		return KSRPackedValues_packPCall(
					WFVarArgs_map(
						BehaviorNode_valueNow,
						unpack( self, 1, rawget( self, 'private:callB_unpackLimit' ) ) ) )
	end
	
	--------------------------------------------------------------------------------
	
	local callB_initNode = makeInitBehaviorNodeForMethods 'callB' {
	
		_receiveData = callB_receiveNonDone,
		_receiveError = callB_receiveNonDone,
		
		_run = function( self )
			return BehaviorNode_setPackedValue( self, callB_evalToPacked( self ) )
		end,
	
	}
	
	--------------------------------------------------------------------------------
	
	callB = function( fn, ... )
	
		local node = allocNode( false )
		
		callB_constructFunctionAndArguments( node, fn, ... )
		
		local initialValue = callB_evalToPacked( node )
		
		return callB_initNode( node, initialValue,
			unpack( node, 1, rawget( node, 'private:callB_unpackLimit' ) ) )
	
	end
	
	--------------------------------------------------------------------------------
	
end


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

-- @mark argTupleB

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

local argTupleB_imp -- forward

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

do

	--------------------------------------------------------------------------------
	
	local function argTupleB_constructInputs( ... )
	
		local n = select( '#', ... )
		
		local inputs = { n = n, ... }
		
		for i = 1, n do
			inputs[ i ] = liftIfNeededB( inputs[ i ] )
		end
		
		return inputs
	
	end
	
	--------------------------------------------------------------------------------
	
	local function argTupleB_receiveNonDone()
		return kScheduleRequestValue
	end
	
	--------------------------------------------------------------------------------
	
	local function argTupleB_evalToPacked_expanded( ... )
	
		local hasErrorB = WFVarArgs_findFirst( BehaviorNode_isErrorNow, ... )
		
		if hasErrorB then
			return BehaviorNode_packedValueNow( hasErrorB )
		end
		
		return KSRPackedValues_pack( WFVarArgs_map( BehaviorNode_valueNow, ... ) )
	
	end
	
	--------------------------------------------------------------------------------
	
	local function argTupleB_evalToPacked( self )
	
		local inputs = rawget( self, 'private:argTupleB_inputs' )
	
		return argTupleB_evalToPacked_expanded( unpack( inputs, 1, inputs.n ) )
		
	end
	
	--------------------------------------------------------------------------------
	
	local argTupleB_initNode = makeInitBehaviorNodeForMethods 'callB' {
	
		_receiveData = argTupleB_receiveNonDone,
		_receiveError = argTupleB_receiveNonDone,
		
		_run = function( self )
			return BehaviorNode_setPackedValue( self, argTupleB_evalToPacked( self ) )
		end,
	
	}
	
	--------------------------------------------------------------------------------
	
	-- Find a non-constant behavior
	
	local function isNonConstantB( x )
		return behaviors[ x ] and not EventNode_useDonePathForSuccessors( x )
	end
	
	--------------------------------------------------------------------------------
	
	local function isConstantErrorB( x )
		return behaviors[ x ]
			and EventNode_useDonePathForSuccessors( x )
			and BehaviorNode_isErrorNow( x )
	end
	
	--------------------------------------------------------------------------------
	
	local function optBehavior_valueNow( x )
	
		if behaviors[ x ] then
			return BehaviorNode_valueNow( x )
		else
			return x
		end
	
	end
	
	--------------------------------------------------------------------------------
	
	local function optBehavior_close( x )
	
		if x and behaviors[ x ] then
			x:close()
		end
	
	end
	
	--------------------------------------------------------------------------------
	
	argTupleB_imp = function( closeInputs, ... )
	
		local n = select( '#', ... )
		
		if n == 0 then
			return constantB()
		end
		
		if n == 1 then
			local x = ( ... )
			if behaviors[ x ] then
				if closeInputs then
					return x:autoCloseRelayB()
				else
					return x:relayB()
				end
			else
				return constantB( x )
			end
		end
		
		-- Check for a constant error.
		
		local constantErrorB = WFVarArgs_findFirst( isConstantErrorB, ... )
		
		if constantErrorB then
			-- We can't just return the original. Well, maybe we could in the
			-- auto-close case but only if we knew that it wasn't being used
			-- elsewhere.
			return constantErrorB:relayB()
		end
	
		-- Check to find something non-constant.
	
		local nonConstantB = WFVarArgs_findFirst( isNonConstantB, ... )
		
		if nonConstantB then
		
			local node = allocNode( false )
			
			local inputs = argTupleB_constructInputs( ... )
			
			rawset( node, 'private:argTupleB_inputs', inputs )
			
			local initialValue = argTupleB_evalToPacked( node )
			
			local resultB = argTupleB_initNode( node, initialValue, unpack( inputs, 1, inputs.n ) )
			
			if closeInputs then
				local closeTargets = { }
				for i = 1, inputs.n do
					local inp = inputs[ i ]
					if not closeTargets[ inp ] then
						closeTargets[ inp ] = true
						addAutoCloseTarget( resultB, inp )
					end
				end
			end
			
			return resultB
		
		end
		
		-- Everything is constant and not an error. Just return a constantB node.
		
		local resultB = constantB( WFVarArgs_map( optBehavior_valueNow, ... ) )
		
		if closeInputs and not AUTO_CLOSE then
			WFVarArgs_map( optBehavior_close, ... )
		end
		
		return resultB
	
	end
	
	--------------------------------------------------------------------------------
	
end

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

local function argTupleB( ... )
	return argTupleB_imp( false, ... )
end


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

-- @mark dataCallB

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

local function dataCallB( fn, ... )

	assert( fn, "No function provided." )
	
	if behaviors[ fn ] then
		local argsB = argTupleB( fn, ... )
		return argsB:closingMapSwitchB( function( f, ... )
			return argTupleB_imp( true, f( ... ) )
		end )
	else
		WFArguments_check( 'callable' )
		local argsB = argTupleB( ... )
		return argsB:closingMapSwitchB( function( ... )
			return argTupleB_imp( true, fn( ... ) )
		end )
	end

end

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

local function closingDataCallB( ... )

	local resultB = dataCallB( ... )
	
	local closeTargets = { }
		-- We can only add a node as a close target once, so watch out for
		-- duplicate inputs.
	
	for i = 1, select( '#', ... ) do
		local inputB = select( i, ... )
		if isEventNode( inputB ) and not closeTargets[ inputB ] then
			closeTargets[ inputB ] = true
			addAutoCloseTarget( resultB, inputB )
		end
	end
	
	return resultB

end


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

-- @mark liftB

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

local liftB, lift

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

do

	--------------------------------------------------------------------------------
	
	local liftBCache = makeFullyWeakTable()
		-- functions to their lifted forms. This makes it safer to lift repeatedly.
	
	--------------------------------------------------------------------------------
	
	liftB = function( fn )
	
		local cached = liftBCache[ fn ]
		
		if cached then
			return cached
		end
		
		local result = function( ... )
			return dataCallB( fn, ... )
		end
		
		liftBCache[ fn ] = result
		liftBCache[ result ] = result -- No sense in lifting twice
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	local string_sub = string.sub
	
	--------------------------------------------------------------------------------
	
	local function getLifter( k, lifters )
	
		if type( k ) ~= 'string' or k == '' then
			return nil
		end
		
		return lifters[ string_sub( k, -1 ) ]
	
	end
	
	--------------------------------------------------------------------------------
	
	local function liftTable( t, lifters )
	
		local result = { }
		
		for k, v in pairs( t ) do
		
			local lifter = getLifter( k, lifters )
			
			if lifter then
				v = lifter( v )
			end
			
			result[ k ] = v
		
		end
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	local functionLifters = {
		[ 'B' ] = liftB
	}
	
	--------------------------------------------------------------------------------
	
	lift = function( t )
	
		return liftTable( t, functionLifters )
	
	end
	
	--------------------------------------------------------------------------------
	
end


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

-- @mark tableB

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

local tableB

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

do

	--------------------------------------------------------------------------------
	
	-- We use special designators for inputs to the template. This is a collection
	-- we build as need be.
	
	local inputObjects = { }
	
	-- We maintain a mapping from input objects to the index they represent
	
	local inputObjectToIndex = { }
	
	--------------------------------------------------------------------------------
	
	local function getInputObject( idx )
	
		local obj = inputObjects[ idx ]
		
		if not obj then
		
			obj = { }
			
			inputObjects[ idx ] = obj
			
			inputObjectToIndex[ obj ] = idx
		
		end
		
		return obj
	
	end
	
	--------------------------------------------------------------------------------
	
	local isSource = isBehavior
	
	--------------------------------------------------------------------------------
	
	local KSPromises_isPromise_lazy = nil
	
	--------------------------------------------------------------------------------
	
	local function checkValidTableForTemplate( v )
	
		local is_event_node, is_behavior = isEventNode( v )
	
		if is_event_node and not is_behavior then
			error "We do not support event nodes in templates for tableB. holdB?"
		end
		
		if KSPromises_isPromise_lazy( v ) then
			error "We do not support promises in templates for tableB. untilResolvedB?"
		end
		
	end
	
	--------------------------------------------------------------------------------
	
	local function convertToTemplate_imp( t, inputList, inputMap )
	
		local tt = type( t )
		
		if tt ~= 'table' then
			return t
		end
		
		checkValidTableForTemplate( t )
		
		if isSource( t ) then
			local idx = inputMap[ t ]
			if not idx then
				idx = #inputList + 1
				inputList[ idx ] = t
				inputMap[ t ] = idx
			end
			return getInputObject( idx )
		end
		
		local result = { }
		
		for k, v in pairs( t ) do
			result[ k ] = convertToTemplate_imp( v, inputList, inputMap )
		end
		
		return result
	
	end

	--------------------------------------------------------------------------------
	
	local function convertToTemplate( t )
	
		if KSPromises_isPromise_lazy == nil then
			KSPromises_isPromise_lazy = require 'KSPromises' . isPromise
		end
		
		local inputList = { }
		local inputMap = { }
		
		local template = convertToTemplate_imp( t, inputList, inputMap )
		
		return template, inputList
	
	end
	
	--------------------------------------------------------------------------------
	
	local function instantiateTemplate( t, inputs )
	
		local tt = type( t )
		
		if tt ~= 'table' then
			return t
		end
		
		local idx = inputObjectToIndex[ t ]
		
		if idx then
			return inputs[ idx ]
		end
		
		local result = { }
		
		for k, v in pairs( t ) do
			result[ k ] = instantiateTemplate( v, inputs )
		end
		
		return result
	
	end
	
	--------------------------------------------------------------------------------
	
	tableB = function( t )
	
		local template, inputList = convertToTemplate( t )
		
		-- OPTIMIZATION: We could special case specific lengths for the source list
		
		local function eval( ... )
		
			return instantiateTemplate( template, { ... } )
		
		end

		return closingDataCallB( eval, unpack( inputList ) )
	
	end
	
	--------------------------------------------------------------------------------

end


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

-- @mark zipMapSwitchB

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

do

	--------------------------------------------------------------------------------
	
	function BehaviorNodeMethods:zipMapSwitchB( fn )
		return argTupleB( self, self:mapSwitchB( fn ) )
	end

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

end


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

local KSReactiveTimer_laterB = lazyFunction( 'KSReactiveTimer', 'laterB' )
local KSReactiveTimer_timerE = lazyFunction( 'KSReactiveTimer', 'timerE' )
local KSReactiveTimer_timedSeqB = lazyFunction( 'KSReactiveTimer', 'timedSeqB' )
local KSReactiveTimer_falseThenTrueB = lazyFunction( 'KSReactiveTimer', 'falseThenTrueB' )
local KSReactiveTimer_trueThenFalseB = lazyFunction( 'KSReactiveTimer', 'trueThenFalseB' )
local KSReactiveTimer_calmE = lazyFunction( 'KSReactiveTimer', 'calmE' )
local KSReactiveTimer_calmB = lazyFunction( 'KSReactiveTimer', 'calmB' )


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

-- @mark calmE, calmB, closingCalmE, closingCalmB

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

function EventNodeMethods:calmE( interval, startTimer )
	return KSReactiveTimer_calmE( self, interval, startTimer )
end

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

function BehaviorNodeMethods:calmB( interval, startTimer )
	return KSReactiveTimer_calmB( self, interval, startTimer )
end

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

EventNodeMethods.closingCalmE = makeClosingMethod( EventNodeMethods.calmE )

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

BehaviorNodeMethods.closingCalmB = makeClosingMethod( BehaviorNodeMethods.calmB )


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

local function fetchCachedBehavior( cache, key, validateValueNow )

	local resultB = WFCloseSupport_fetchCached( cache, key )

	if not resultB then
		return nil
	end
	
	local value = resultB:valueNow()
	
	-- If the value has been closed then we definitely wanted a cache miss.
	
	if value and WFCloseSupport_hasBeenOrWillBeClosed( value ) then
		cache[ key ] = nil
		return nil
	end
	
	-- If we have a validator to apply, then we want to apply it.
	
	if validateValueNow and not validateValueNow( value ) then
		cache[ key ] = nil
		return nil
	end
	
	-- Everything looks okay (at least from a quick inspection).
	
	return resultB

end

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

local function createReactiveVariable( self, key, ... )

	if self[ key ] then
		error( "Already had reactive variable: " .. tostring( key ) )
	end
	
	self:setOwned { [ key ] = variableB( ... ) }

	return self

end

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

--[[!
	\brief Adds utility methods to a class for use of the reactive system.

	\param methodTable A table to which to add the function pointers.
]]--

local function addReactiveSupportMethods( methodTable )

	assert( not methodTable.createReactiveVariable,
		"Already had createReactiveVariable method" )
		
	assert( methodTable.setOwned,
		"Must support the setOwned method to add reactive support." )
		
	methodTable.createReactiveVariable = createReactiveVariable
	
	return methodTable

end

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

return {
	sourceE = sourceE,
	variableB = variableB,
	constantB = constantB,
	constantErrorB = constantErrorB,
	zeroE = zeroE,
	anyE = anyE,
	closingAnyE = closingAnyE,
	callB = callB,
	argTupleB = argTupleB,
	dataCallB = dataCallB,
	closingDataCallB = closingDataCallB,
	tableB = tableB,
	liftB = liftB,
	lift = lift,
	isEventNode = isEventNode,
	isBehavior = isBehavior,
	timerE = KSReactiveTimer_timerE,
	timedSeqB = KSReactiveTimer_timedSeqB,
	falseThenTrueB = KSReactiveTimer_falseThenTrueB,
	trueThenFalseB = KSReactiveTimer_trueThenFalseB,
	laterB = KSReactiveTimer_laterB,
	makeWeakEventTable = lazyFunction( 'KSWeakEventTable', 'makeWeakEventTable' ),
	fetchCachedBehavior = fetchCachedBehavior,
	SAME = kSameValue,
	SKIP = kSkipValue,
	DONE = kDoneValue,
	
	addReactiveSupportMethods = addReactiveSupportMethods,
	
	testing_flushAutoClose = flushAutoClose
}
