eventdispatcher.cpp

.net 方面的开发说明资料。

// ========================================================
// Event Dispatching
//
// Design and Implementation by Floris van den Berg
// ========================================================

#pragma warning (disable : 4275)
#pragma warning (disable : 4786)
#pragma warning (disable : 4200)

#include <stdio.h>
#include <assert.h>

#include <boost/thread/xtime.hpp>

#include "Utilities.h"
#include "Agent.h"
#include "OpenNet.h"
#include "EventDispatcher.h"
#include "Transport.h"

// --------------------------------------------------------

extern IAgent *s_agent_container;

// --------------------------------------------------------
// Main thread function for event dispatching
// --------------------------------------------------------

DispatchFunctor::DispatchFunctor(EventDispatcher *entry) :
m_entry(entry) {
}

bool
DispatchFunctor::operator()() {
	EpEventEx *event = NULL;

	// grab the next event from the queue

	{
		boost::mutex::scoped_lock scoped_lock(m_entry->m_mutex);

		if (!m_entry->m_event_list.empty()) {
			event = m_entry->m_event_list.front();
			m_entry->m_event_list.pop_front();
		}
	}

	// process the event

	if (event) {
		TransportEntry *plug_entry = s_agent_container->findPlugEntry((ITransport *)event->transport);

		if (plug_entry) {
			// send the event to the protocols that registered for it

			bool do_dispatch = DispatchEvent(plug_entry, event->event->protocol, event->event);

			// if the sessions didn't block this event, send it to user callback
			// note that system events can't be blocked...

			if (plug_entry->callback) {
				if (IsEqualGUID(event->event->protocol, CLSID_SYSTEM_PROTOCOL)) {
					if ((event->event->msg != SYSTEM_DATA_IN) && (event->event->msg != SYSTEM_DATA_OUT)) {
						plug_entry->callback(event->transport, event->event, plug_entry->data);
					} else if (plug_entry->m_enable_debug_events) {
						plug_entry->callback(event->transport, event->event, plug_entry->data);
					}
				} else if (do_dispatch) {
					plug_entry->callback(event->transport, event->event, plug_entry->data);
				}
			}
		}

		m_entry->destroyEventEntry(event);
		return true;
	}

	return false;	
}

bool
DispatchFunctor::DispatchEvent(TransportEntry *plug_entry, GUID &guid, EpEvent *event) {
	bool do_dispatch = true;

	// system event dispatching

	if (IsEqualGUID(event->protocol, CLSID_SYSTEM_PROTOCOL)) {
		switch(event->msg) {
			case SYSTEM_CONNECTED :
				EpCancelTimeout(plug_entry->transport, CLSID_SYSTEM_PROTOCOL, SYSTEM_CONNECT_TIMEOUT);
				break;

			case SYSTEM_DATA_IN :
			{
				if ((plug_entry->protocols_it = plug_entry->protocols.rbegin()) != plug_entry->protocols.rend())
					(*plug_entry->protocols_it)->protocol->receive((*plug_entry->protocols_it)->self, event->data, event->size);

				if (((!plug_entry->m_enable_debug_events) && (!plug_entry->protocols.empty())))
					return false;

				break;
			}

			case SYSTEM_DATA_OUT :
			{
				if (!plug_entry->m_enable_debug_events)
					return false;

				break;
			}
		};
	}

	// return true if this event may be sent to the general callback

	return do_dispatch;
}

// --------------------------------------------------------
// Event Dispatch Class
// --------------------------------------------------------

EventDispatcher::EventDispatcher() :
m_dispatch_functor(NULL),
m_mutex(),
m_event_list(),
m_time_stamp(EpGetTickCount()),
m_delay(0) {
	m_dispatch_functor = new DispatchFunctor(this);
}

EventDispatcher::~EventDispatcher() {
	delete m_dispatch_functor;
}

// --------------------------------------------------------

EpEventEx *
EventDispatcher::createEventEntry(TRANSPORT_HANDLE transport, EpEvent *event) {
	assert(event != NULL);

	EpEvent *clone = new EpEvent;
	clone->reference_id = event->reference_id;
	clone->protocol = event->protocol;
	clone->msg = event->msg;

	if ((event->data) && (event->size > 0)) {
		clone->size = event->size;
		clone->data = new unsigned char[event->size];
		memcpy(clone->data, event->data, event->size);
	} else {
		clone->size = 0;
		clone->data = NULL;
	}

	EpEventEx *entry = new EpEventEx;
	entry->transport = transport;
	entry->event = clone;
	return entry;
}

void
EventDispatcher::destroyEventEntry(EpEventEx *event) {
	if (event->event->data)
		delete [] event->event->data;

	delete event->event;
	delete event;
}

void
EventDispatcher::pushEventEntry(EpEventEx *event) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	m_event_list.push_back(event);
}

bool
EventDispatcher::heartBeat() {
	DWORD tick = EpGetTickCount();

	// correct tick when gettickcount wraps to 0

	if (m_time_stamp <= tick)
		tick -= m_time_stamp;
	else
		tick = (0xFFFFFFFF - m_time_stamp) + tick;

	// first check for changed connections

	if (tick >= m_delay) {
		m_time_stamp = EpGetTickCount();

		// dispatch the next events on the queue
		// the functor returns true if it processed an event,
		// or false if there weren't any events on the queue

		m_delay = (*m_dispatch_functor)() ? 0 : 10;
		return (m_delay == 0);
	}

	return false;
}

void
EventDispatcher::clearPendingEventEntries(EventList &list) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	EventList::iterator i = list.begin();

	while (i != list.end()) {
		destroyEventEntry(*i);
		i = list.erase(i);
	}
}

void
EventDispatcher::clearPendingEventEntries() {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	clearPendingEventEntries(m_event_list);
}

void
EventDispatcher::clearPendingEventEntries(TRANSPORT_HANDLE transport, EventList &list) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	EventList::iterator i = list.begin();

	while (i != list.end()) {
		if ((*i)->transport == transport) {
			destroyEventEntry(*i);

			i = list.erase(i);

			continue;
		}

		++i;
	}
}

void
EventDispatcher::clearPendingEventEntries(TRANSPORT_HANDLE transport) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	clearPendingEventEntries(transport, m_event_list);
}

void
EventDispatcher::clearPendingEventEntries(TRANSPORT_HANDLE transport, CLSID protocol, EventList &list) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	EventList::iterator i = list.begin();

	while (i != list.end()) {
		if ((*i)->transport == transport) {
			if (IsEqualGUID((*i)->event->protocol, protocol)) {
				destroyEventEntry(*i);

				i = list.erase(i);

				continue;
			}
		}

		++i;
	}
}

void
EventDispatcher::clearPendingEventEntries(TRANSPORT_HANDLE transport, CLSID protocol) {
	boost::mutex::scoped_lock scoped_lock(m_mutex);

	clearPendingEventEntries(transport, protocol, m_event_list);
}

// --------------------------------------------------------
// Singleton implementation
// --------------------------------------------------------

EventDispatcher *EventDispatcher::instance = NULL;

EventDispatcher *
EventDispatcher::getInstance() {
	if (!isInstantiated())
		EventDispatcher::instance = new EventDispatcher;

	return EventDispatcher::instance;
}

void
EventDispatcher::destroyInstance() {
	if (isInstantiated()) {
		delete EventDispatcher::instance;

		EventDispatcher::instance = NULL;
	}
}

bool
EventDispatcher::isInstantiated() {
	return (EventDispatcher::instance != NULL);
}