BT411/engine/MUNGA/EVENT.cpp

#include "munga.h"
#pragma hdrstop

#include "event.h"
#include "registry.h"
#include "app.h"
#include "entity.h"
#include "evtstat.h"

#if defined(TRACE_PROCESS_EVENT)
	static BitTrace Process_Event("Process Event");
#	define SET_PROCESS_EVENT() Process_Event.Set()
#	define CLEAR_PROCESS_EVENT() Process_Event.Clear()
#else
#	define SET_PROCESS_EVENT()
#	define CLEAR_PROCESS_EVENT()
#endif

//#############################################################################
//###########################    AbstractEvent    #############################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
AbstractEvent::AbstractEvent(
	Receiver::Message *message,
	Time time,
	ClassID class_id
):
	Node(class_id)
{
	Verify(message);
	alarmTime = time;

	//
	//--------------------------------------------------------
	// Allocate some memory to store the message parameters in
	//--------------------------------------------------------
	//
	size_t long_size = (message->messageLength+3)>>2;
	messageToSend = (Receiver::Message*)new long[long_size];
	Check_Pointer(messageToSend);
	Register_Pointer(messageToSend);

	Mem_Copy(
		messageToSend,
		message,
		message->messageLength,
		long_size*sizeof(long)
	);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
AbstractEvent::~AbstractEvent()
{
	if (messageToSend)
	{
		Unregister_Pointer(messageToSend);
		delete messageToSend;
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	AbstractEvent::Process()
{
	Fail("AbstractEvent::Process should not be called!\n");
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	AbstractEvent::DumpData()
{
	Fail("AbstractEvent::DumpData should not be called!\n");
}

//#############################################################################
//###############################    Event    #################################
//#############################################################################

MemoryBlock *Event::GetAllocatedMemory()
{
	static MemoryBlock allocatedMemory(sizeof(Event), 20, 10, "Events");
	return &allocatedMemory;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
Event::Event(
	Receiver *target,
	Receiver::Message *message,
	Time time
):
	AbstractEvent(message, time, EventClassID),
	targetReceiver(this)
{
	targetReceiver.Add(target);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
Event::~Event()
{
#if defined(TRACE_EVENT_COUNT)
	//
	//-------------------------------------------
	// Delete the link to the current event queue
	//-------------------------------------------
	//
	PlugIteratorOf<EventQueue*>
		old_queue_link(this);
	Node *node;
	while ((node = old_queue_link.GetCurrent()) != NULL)
	{
		Check(node);
		if (
			node->GetClassID() == EventQueueClassID
			 || node->GetClassID() == GeneralEventQueueClassID
		)
		{
			GeneralEventQueue *q = Cast_Object(GeneralEventQueue*, node);
			while (q->priorityLevels < 0)
			{
				--q;
				Check(q);
			}
			--q->eventCount;
			if (q->eventCountTrace)
			{
				Check(q->eventCountTrace);
				q->eventCountTrace->TakeSnapshot(q->eventCount);
			}
			break;
		}
	}
#endif
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	Event::Repost(
		EventQueue* queue,
		int priority,
		Time new_time
	)
{
	Check(this);
	Check(queue);

	//
	//-------------------------------------------
	// Delete the link to the current event queue
	//-------------------------------------------
	//
	PlugIteratorOf<EventQueue*>
		old_queue_link(this);
	Node *node;
	while ((node = old_queue_link.GetCurrent()) != NULL)
	{
		Check(node);
		if (
			node->GetClassID() == EventQueueClassID
			 || node->GetClassID() == GeneralEventQueueClassID
			 || node->GetClassID() == DeferredEventQueueClassID
		)
		{
			#if defined(TRACE_EVENT_COUNT)
				if (node->GetClassID() != DeferredEventQueueClassID)
				{
					GeneralEventQueue *q = Cast_Object(GeneralEventQueue*, node);
					while (q->priorityLevels < 0)
					{
						--q;
						Check(q);
					}
					--q->eventCount;
					if (q->eventCountTrace)
					{
						Check(q->eventCountTrace);
						q->eventCountTrace->TakeSnapshot(q->eventCount);
					}
				}
			#endif

			old_queue_link.Remove();
			break;
		}
		else
		{
			old_queue_link.Next();
		}
	}

	//
	//-----------------------------------------------------------------------
	// Now, if necessary, reset the time and stuff the event in the new event
	// queue
	//-----------------------------------------------------------------------
	//
	if (new_time.ticks)
	{
		alarmTime = new_time;
	}
	queue->Enqueue(priority,this);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	Event::Defer()
{
	//
	//-----------------------------------------------
	// See if the target has a deferred queue already
	//-----------------------------------------------
	//
	Receiver *target = targetReceiver.GetCurrent();
	Check(target);
	PlugIteratorOf<DeferredEventQueue*>
		blind_links(target);
	DeferredEventQueue
		*deferred_queue;
	while ((deferred_queue = blind_links.ReadAndNext()) != NULL)
	{
		if (deferred_queue->GetClassID() == DeferredEventQueueClassID)
		{
			break;
		}
	}

	//
	//----------------------------------------------------------------------
	// If we couldn't find a deferred queue, then a new one must be created
	// and attached to the object
	//----------------------------------------------------------------------
	//
	if (!deferred_queue)
	{
		deferred_queue = new DeferredEventQueue(target);
		Register_Object(deferred_queue);
	}

	//
	//-------------------------------------------------
	// Now, simply repost the event to this event queue
	//-------------------------------------------------
	//
	Repost(deferred_queue->deferredEvents,0);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	Event::Process()
{
	Check(this);
	targetReceiver.GetCurrent()->Receive(this);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	Event::DumpData()
{
	Check(this);

	Receiver *target = targetReceiver.GetCurrent();
	Check(target);
	Check(application);
	Registry *registry = application->GetRegistry();
	if (registry)
	{
		Check(registry);
		Receiver::SharedData *shared_data =
			registry->GetStaticData(target->GetClassID());
		if (shared_data)
		{
			Check(shared_data);
			Derivation *derivation = shared_data->derivedClasses;
			Check(derivation);
			Receiver::MessageHandlerSet *handlers =
				shared_data->activeMessageHandlers;
			Check(handlers);
			DEBUG_STREAM << handlers->GetName(messageToSend->messageID) << " message for class " << derivation->className << std::endl << std::flush;
		}
		else
		{
			goto Vanilla_Message;
		}
	}
	else
	{
Vanilla_Message:
		DEBUG_STREAM << "Message " << messageToSend->messageID << " for object of classID " << target->GetClassID() << std::endl << std::flush;
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	Event::ReleaseLinkHandler(
		Socket*,
		Plug*
	)
{
	Unregister_Object(this);
	delete this;
}

//#############################################################################
//###########################    NetworkEvent    ##############################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
NetworkEvent::NetworkEvent(
	HostID host_ID,
	NetworkManager::ClientID client_ID,
	EventStyle event_style,
	Receiver::Message *message,
	Time time
):
	AbstractEvent(message, time, NetworkEventClassID)
{
	hostID = host_ID;
	clientID = client_ID;
	eventStyle = event_style;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
NetworkEvent::~NetworkEvent()
{
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	NetworkEvent::Process()
{
	Check(this);
	switch (eventStyle)
	{
	case SendStyle:
		Check(application);
		application->GetNetworkManager()->Send(
			messageToSend,
			clientID,
			hostID
		);
		break;

	case BroadcastStyle:
		Check(application);
		application->GetNetworkManager()->Broadcast(
			messageToSend,
			clientID
		);
		break;

	case ExclusiveBroadcastStyle:
		Check(application);
		application->GetNetworkManager()->ExclusiveBroadcast(
			messageToSend,
			clientID
		);
		break;
	}

	Unregister_Object(this);
	delete this;
}

//#############################################################################
//#######################    GeneralEventQueue    ############################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
GeneralEventQueue::GeneralEventQueue(ClassID class_ID):
	Node(class_ID),
	pendingEvents(this),
	timedEvents(this)
{
	priorityLevels = -1;
	eventCount = 0;
	#if defined(TRACE_EVENT_COUNT)
		eventCountTrace = NULL;
	#endif
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
GeneralEventQueue::GeneralEventQueue():
	Node(GeneralEventQueueClassID),
	pendingEvents(this),
	timedEvents(this)
{
	priorityLevels = -1;
	eventCount = 0;
	#if defined(TRACE_EVENT_COUNT)
		eventCountTrace = NULL;
	#endif
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
GeneralEventQueue::~GeneralEventQueue()
{
	#if defined(TRACE_EVENT_COUNT)
		if (eventCountTrace)
		{
			Unregister_Object(eventCountTrace);
			delete eventCountTrace;
		}
	#endif
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
GeneralEventQueue*
	GeneralEventQueue::Make(
		int priorities,
		#if defined(TRACE_EVENT_COUNT)
			const char* trace_name
		#else
			const char*
		#endif
	)
{
	GeneralEventQueue *queue = new GeneralEventQueue[priorities];
	Register_Object(queue);

	queue->priorityLevels = priorities;
	#if defined(TRACE_EVENT_COUNT)
		if (trace_name)
		{
			Check_Pointer(trace_name);
			queue->eventCountTrace =
				new TraceOf<int>(
					trace_name,
					0,
					Trace::IntegerType,
					TraceSample::IntegerSnapshot
				);
			Register_Object(queue->eventCountTrace);
		}
	#endif
	return queue;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::Post(
		int priority,
		Receiver *target,
		Receiver::Message *message,
		Time time
	)
{
	Check(this);
	Check(target);
	Check(message);

	//
	//------------------------------------------------------------------------
	// Create a new event, then insert it at the end of the event queue.  This
	// means that timed events get put in sort of haphazardly!!!!
	//------------------------------------------------------------------------
	//
	Event *event = new Event(target, message, time);
	Register_Object(event);

	Enqueue(priority, event);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::SendEvent(
		int priority,
		HostID host_ID,
		NetworkManager::ClientID client_id,
		Receiver::Message *message,
		Time when
	)
{
	Check(this);
	Check(message);

	NetworkEvent *event =
		new NetworkEvent(
			host_ID, client_id, NetworkEvent::SendStyle, message, when
		);
	Register_Object(event);
	Enqueue(priority, event);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::BroadcastEvent(
		int priority,
		NetworkManager::ClientID client_ID,
		Receiver::Message *message,
		Time when
	)
{
	Check(this);
	Check(message);

	//
	// HACK - What is a null network address?
	//
	NetworkEvent *event =
		new NetworkEvent(
			(NetworkAddress)0,
			client_ID,
			NetworkEvent::BroadcastStyle,
			message,
			when
		);
	Register_Object(event);
	Enqueue(priority, event);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::ExclusiveBroadcastEvent(
		int priority,
		NetworkManager::ClientID client_ID,
		Receiver::Message *message,
		Time when
	)
{
	Check(this);
	Check(message);

	//
	// HACK - What is a null network address?
	//
	NetworkEvent *event =
		new NetworkEvent(
			(NetworkAddress)0,
			client_ID,
			NetworkEvent::ExclusiveBroadcastStyle,
			message,
			when
		);
	Register_Object(event);
	Enqueue(priority, event);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::Enqueue(
		int priority,
		AbstractEvent *event
	)
{
	Check(event);
	Verify((unsigned)priority < priorityLevels);

	//
	//------------------------------------------------------------
	// If this is not a timed event, stuff it in the pending queue
	//------------------------------------------------------------
	//
	if (!event->alarmTime.ticks)
	{
		this[priority].pendingEvents.Add(event);
	}

	//
	//----------------------------------------
	// Otherwise, sort it into the timed queue
	//----------------------------------------
	//
	else
	{
		ChainIteratorOf<AbstractEvent*>
			timed_events(this[priority].timedEvents);
		timed_events.Last();
		AbstractEvent* next_event = timed_events.GetCurrent();

		//
		//---------------------------------------------------------------------
		// If the timed queue is empty, or if we are later than the last entry,
		// just add the event to the end
		//---------------------------------------------------------------------
		//
		if (!next_event || next_event->alarmTime.ticks <= event->alarmTime.ticks)
		{
			this[priority].timedEvents.Add(event);
		}

		//
		//--------------------------------------------------------------------
		// Find the event to insert before.  If we run off the beginning, stop
		// checking and set the iterator to the first event in the queue
		//--------------------------------------------------------------------
		//
		else
		{
			do
			{
				timed_events.Previous();
				next_event = timed_events.GetCurrent();
				if (!next_event)
				{
					timed_events.First();
					break;
				}
			} while (next_event->alarmTime.ticks > event->alarmTime.ticks);

			//
			//--------------------------------------------------------------------
			// If we didn't run off the queue, we need to insert after the current
			// event, so bump the iterator forward one event
			//--------------------------------------------------------------------
			//
			if (next_event)
			{
				timed_events.Next();
			}

			//
			//------------------------------------------------------------------
			// Insert the new event ahead of where the iterator currently points
			//------------------------------------------------------------------
			//
			timed_events.Insert(event);
		}
	}
	#if defined(TRACE_EVENT_COUNT)
		++eventCount;
		if (eventCountTrace)
		{
			Check(eventCountTrace);
			eventCountTrace->TakeSnapshot(eventCount);
		}
	#endif
	#if defined(USE_EVENT_STATISTICS)
		event_statistics_manager.Maintain(Cast_Object(Event*, event));
	#endif
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
AbstractEvent*
	GeneralEventQueue::PeekAtNextEvent(int min_priority)
{
	Check(this);

	//
	//--------------------------------------------------------------------------
	// Start at the top of the priority queue, searching backwards until we find
	// an available event or run off the minimum queue priority
	//--------------------------------------------------------------------------
	//
	Verify(min_priority >= 0);
	int priority = priorityLevels;
	while (--priority >= min_priority)
	{
		//
		//-------------------------------------------------------------------
		// See if the first event in the timed queue is ready to go.  If not,
		// none of the other timed events will be ready to go either
		//-------------------------------------------------------------------
		//
		ChainIteratorOf<AbstractEvent*>
			timed_events(this[priority].timedEvents);
		AbstractEvent
			*event = timed_events.GetCurrent();
		if (event)
		{
			Check(event);
			if (event->alarmTime.ticks <= Now().ticks)
			{
				return event;
			}
		}

		//
		//------------------------------
		// Now look in the untimed queue
		//------------------------------
		//
		ChainIteratorOf<AbstractEvent*>
			events(this[priority].pendingEvents);
		event = events.GetCurrent();
		if (event)
		{
			Check(event);
			return event;
		}
	}
	return NULL;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
Logical
	GeneralEventQueue::IsPriorityEmpty(int priority)
{
	Check(this);

	Verify(priority >= 0 && priority < priorityLevels);
	ChainIteratorOf<AbstractEvent*> i(this[priority].pendingEvents);
	ChainIteratorOf<AbstractEvent*> j(this[priority].timedEvents);
	return !i.GetCurrent() && !j.GetCurrent();
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
Logical
	GeneralEventQueue::ProcessOneEvent(int min_priority)
{
	Check(this);
	Verify(min_priority >= 0);

	AbstractEvent *event=PeekAtNextEvent(min_priority);
	if (event)
	{
		SET_PROCESS_EVENT();
		event->Process();
		CLEAR_PROCESS_EVENT();
		return True;
	}
	return False;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::ProcessAllEvents(int min_priority)
{
	Check(this);
	Verify(min_priority >= 0);
	while (ProcessOneEvent(min_priority));
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::FlushAllEvents(int max_priority)
{
	Check(this);
	Verify(max_priority < priorityLevels);

	//
	//-----------------------------------------
	// Make sure that max_priority is set right
	//-----------------------------------------
	//
	if (max_priority < 0)
	{
		max_priority = priorityLevels - 1;
	}

	//
	//---------------------------------------------------------------------
	// Step through each of the priority queues and delete any events found
	// within
	//---------------------------------------------------------------------
	//
	for (int priority=max_priority; priority>=0; --priority)
	{

		//
		//---------------------------------------
		// Delete the events from the timed queue
		//---------------------------------------
		//
		ChainIteratorOf<AbstractEvent*> i(this[priority].timedEvents);
		AbstractEvent *event;
		while ((event = i.ReadAndNext()) != NULL)
		{
			Unregister_Object(event);
			delete event;
		}

		//
		//-----------------------------------------
		// Delete the events from the regular queue
		//-----------------------------------------
		//
		ChainIteratorOf<AbstractEvent*> j(this[priority].pendingEvents);
		while ((event = j.ReadAndNext()) != NULL)
		{
			Unregister_Object(event);
			delete event;
		}
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::FlushMatchingEvents(
		Receiver::MessageID target_message,
		int max_priority
	)
{
	Check(this);
	Verify(max_priority < priorityLevels);

	//
	//-----------------------------------------
	// Make sure that max_priority is set right
	//-----------------------------------------
	//
	if (max_priority < 0)
	{
		max_priority = priorityLevels - 1;
	}

	//
	//---------------------------------------------------------------------
	// Step through each of the priority queues and delete any events found
	// within
	//---------------------------------------------------------------------
	//
	for (int p=max_priority; p>=0; --p)
	{
		//
		//-------------------------------------------
		// Delete events from the regular queue first
		//-------------------------------------------
		//
		ChainIteratorOf<AbstractEvent*> i(this[p].pendingEvents);
		AbstractEvent *event;
		while ((event = i.ReadAndNext()) != NULL)
		{

			//
			//------------------------------
			// Make sure the message matches
			//------------------------------
			//
			if (
				target_message == Receiver::AnyMessageID
				 || target_message == event->messageToSend->messageID
			)
			{
				Unregister_Object(event);
				delete event;
			}
		}

		//
		//-----------------------------------
		// Delete events from the timed queue
		//-----------------------------------
		//
		ChainIteratorOf<AbstractEvent*> j(this[p].timedEvents);
		while ((event = j.ReadAndNext()) != NULL)
		{

			//
			//------------------------------
			// Make sure the message matches
			//------------------------------
			//
			if (
				target_message == Receiver::AnyMessageID
				 || target_message == event->messageToSend->messageID
			)
			{
				Unregister_Object(event);
				delete event;
			}
		}
	}
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	GeneralEventQueue::DumpEventQueue()
{
	Check(this);

	//
	//----------------------------------------------------------------------
	// Step through each of the priority queues and display any events found
	// within
	//----------------------------------------------------------------------
	//
	CollectionSize total_count = 0;
	for (int priority=priorityLevels - 1; priority>=0; --priority)
	{
		DEBUG_STREAM << "EventQueue priority " << priority
		 << "\n--------------------------------------------------------------\n";

		ChainIteratorOf<AbstractEvent*> i(this[priority].pendingEvents);
		AbstractEvent *event;
		size_t count=0;
		while ((event = i.ReadAndNext()) != NULL)
		{
			++count;
			event->DumpData();
		}
		DEBUG_STREAM << count << " priority " << priority << " events\n\n" << std::flush;
		count = 0;
		ChainIteratorOf<AbstractEvent*> j(this[priority].timedEvents);
		while ((event = j.ReadAndNext()) != NULL)
		{
			++count;
			event->DumpData();
		}
		DEBUG_STREAM << count << " priority " << priority << " timed events\n\n" << std::flush;
		total_count += count;

	}
	DEBUG_STREAM << total_count << " total events in queues\n\n" << std::flush;
}

//#############################################################################
//############################    EventQueue    ###############################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
EventQueue::EventQueue():
	GeneralEventQueue(EventQueueClassID)
{
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
EventQueue::~EventQueue()
{
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
EventQueue*
	EventQueue::Make(
		int priorities,
		#if defined(TRACE_EVENT_COUNT)
			const char* trace_name
		#else
			const char*
		#endif
	)
{
	EventQueue *queue = new EventQueue[priorities];
	Register_Object(queue);

	queue->priorityLevels = priorities;
	#if defined(TRACE_EVENT_COUNT)
		if (trace_name)
		{
			Check_Pointer(trace_name);
			queue->eventCountTrace =
				new TraceOf<int>(
					trace_name,
					0,
					Trace::IntegerType,
					TraceSample::IntegerSnapshot
				);
			Register_Object(queue->eventCountTrace);
		}
	#endif
	return queue;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	EventQueue::ProcessMatchingEvents(
		Receiver *receiver,
		Receiver::MessageID message_ID
	)
{
	Check(this);
	Check(receiver);

	//
	//----------------------------------------------------------------
	// Step through each of the priority queues and process any events
	// found within
	//----------------------------------------------------------------
	//
	for (int p=priorityLevels-1; p>=0; --p)
	{
		Event *event;
		AbstractEvent *abstract_event;
		ChainIteratorOf<AbstractEvent*> i(this[p].pendingEvents);
		while ((abstract_event = i.ReadAndNext()) != NULL)
		{
			Check(abstract_event);
			event = Cast_Object(Event*, abstract_event);

			//
			//------------------------------
			// Make sure the message matches
			//------------------------------
			//
			if (
				(event->targetReceiver.GetCurrent() == receiver) &&
				(message_ID == Receiver::AnyMessageID ||
				 message_ID == event->messageToSend->messageID)
			)
			{
				event->Process();
			}
		}

		ChainIteratorOf<AbstractEvent*> j(this[p].timedEvents);
		while ((abstract_event = j.ReadAndNext()) != NULL)
		{
			Check(abstract_event);
			event = Cast_Object(Event*, abstract_event);

			//
			//------------------------------
			// Make sure the message matches
			//------------------------------
			//
			if (
				(event->targetReceiver.GetCurrent() == receiver) &&
				(message_ID == Receiver::AnyMessageID ||
				 message_ID == event->messageToSend->messageID)
			)
			{
				event->Process();
			}
		}
	}
}

//#############################################################################
//########################    DeferredEventQueue    ###########################
//#############################################################################

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
DeferredEventQueue::DeferredEventQueue(Receiver* target):
	Node(DeferredEventQueueClassID),
	waitingReceiver(this)
{
	deferredEvents = new EventQueue;
	Register_Object(deferredEvents);

	waitingReceiver.Add(target);
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
DeferredEventQueue::~DeferredEventQueue()
{
	Unregister_Object(deferredEvents);
	delete deferredEvents;
}

//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
//
void
	DeferredEventQueue::ReleaseLinkHandler(
		Socket*,
		Plug*
	)
{
	Unregister_Object(this);
	delete this;
}

#if defined(TEST_CLASS)
	#include "event.tcp"
#endif