XNSim/XNCore/XNEventManager.cpp

#include "XNEventManager.h"
#include "XNEventManager_p.h"
#include "XNFramework.h"

// 构造函数
XNEventManager::XNEventManager(QObject *parent)
	: XNBaseFrameObject(*new XNEventManagerPrivate(this), parent)
{
	// 设置唯一标识符
	setUniqueId(7);
	// 设置对象名称
	setObjectName("XNEventManager");
}

// 析构函数
XNEventManager::~XNEventManager()
{
}

// 保护构造函数实现
XNEventManager::XNEventManager(XNEventManagerPrivate &dd, QObject *parent)
	: XNBaseFrameObject(dd, parent)
{
}

// 添加一个用于异步执行的任务类
class EventTask : public QRunnable
{
public:
	EventTask(const QString &name, const QVariant &data,
			  std::function<void(const QVariant &)> callback, XNEventManager *manager)
		: eventName(name), eventData(data), eventCallback(callback), eventManager(manager)
	{
		setAutoDelete(true);
	}

	void run() override
	{
		try {
			eventCallback(eventData);
			if (eventManager) {
				QMetaObject::invokeMethod(eventManager, "EventProcessed", Qt::QueuedConnection,
										  Q_ARG(QString, eventName), Q_ARG(bool, true));
			}
		} catch (const std::exception &e) {
			LOG_ERROR(
				QString("Async event handler exception for %1: %2").arg(eventName).arg(e.what()));
			if (eventManager) {
				QMetaObject::invokeMethod(eventManager, "EventProcessed", Qt::QueuedConnection,
										  Q_ARG(QString, eventName), Q_ARG(bool, false));
			}
		}
	}

private:
	QString eventName;
	QVariant eventData;
	std::function<void(const QVariant &)> eventCallback;
	XNEventManager *eventManager;
};

// 修改注册事件处理器的实现
int XNEventManager::RegisterEventHandler(const QString &eventName,
										 std::function<void(const QVariant &)> callback,
										 quint32 objectId, bool async, XNEvent::Priority priority)
{
	Q_D(XNEventManager);
	if (eventName.isEmpty() || !callback) {
		LOG_WARNING("Invalid event name or callback!");
		return -1;
	}

	QMutexLocker locker(&d->eventMutex);

	// 生成新的本地ID
	d->localIdCounter = (d->localIdCounter + 1) & 0xFFFF;
	if (d->localIdCounter == 0)
		d->localIdCounter = 1; // 避免为0

	// 创建处理器信息
	EventHandlerInfo handlerInfo;
	handlerInfo.callback = callback;
	handlerInfo.objectId = objectId;
	handlerInfo.localId	 = d->localIdCounter;
	handlerInfo.isAsync	 = async;
	handlerInfo.priority = priority;
	// 计算全局处理器ID
	int handlerId = handlerInfo.GetHandlerId();

	// 添加处理器信息到事件列表
	d->eventHandlers[eventName].append(handlerInfo);
	// 添加反向映射
	d->handlerToEvent[handlerId] = eventName;

	LOG_INFO(
		QString("Registered %1 event handler for event: %2, handler ID: %3 (object: %4, local: %5)")
			.arg(async ? "async" : "sync")
			.arg(eventName)
			.arg(handlerId)
			.arg(objectId)
			.arg(d->localIdCounter));
	return handlerId;
}

// 修改移除事件处理器的实现
bool XNEventManager::RemoveEventHandler(const QString &eventName, int handlerId)
{
	Q_D(XNEventManager);
	QMutexLocker locker(&d->eventMutex);

	// 如果指定了事件名称，先验证事件是否存在
	if (!eventName.isEmpty()) {
		if (!d->eventHandlers.contains(eventName)) {
			LOG_WARNING(QString("Event %1 not found!").arg(eventName));
			return false;
		}

		// 查找并移除指定的处理器
		auto &handlers = d->eventHandlers[eventName];
		for (auto it = handlers.begin(); it != handlers.end(); ++it) {
			if (it->GetHandlerId() == handlerId) {
				handlers.erase(it);
				d->handlerToEvent.remove(handlerId);
				LOG_INFO(
					QString("Removed handler ID %1 from event: %2").arg(handlerId).arg(eventName));

				// 如果事件没有处理器了，移除整个事件
				if (handlers.isEmpty()) {
					d->eventHandlers.remove(eventName);
				}
				return true;
			}
		}
		LOG_WARNING(QString("Handler ID %1 not found in event: %2").arg(handlerId).arg(eventName));
		return false;
	}

	// 如果没有指定事件名称，使用反向映射查找
	auto eventIt = d->handlerToEvent.find(handlerId);
	if (eventIt != d->handlerToEvent.end()) {
		QString eventToRemove = eventIt.value();
		auto &handlers		  = d->eventHandlers[eventToRemove];

		for (auto it = handlers.begin(); it != handlers.end(); ++it) {
			if (it->GetHandlerId() == handlerId) {
				handlers.erase(it);
				d->handlerToEvent.remove(handlerId);
				LOG_INFO(QString("Removed handler ID %1 from event: %2")
							 .arg(handlerId)
							 .arg(eventToRemove));

				// 如果事件没有处理器了，移除整个事件
				if (handlers.isEmpty()) {
					d->eventHandlers.remove(eventToRemove);
				}
				return true;
			}
		}
	}

	LOG_WARNING(QString("Handler ID %1 not found!").arg(handlerId));
	return false;
}

// 修改触发事件的实现
void XNEventManager::TriggerEvent(const QString &eventName, const QVariant &eventData,
								  bool forceAsync, XNEvent::Priority priority)
{
	Q_D(XNEventManager);

	QList<EventHandlerInfo> handlers;
	{
		QMutexLocker locker(&d->eventMutex);
		if (!d->eventHandlers.contains(eventName)) {
			//LOG_WARNING(QString("No handlers registered for event: %1").arg(eventName));
			return;
		}
		handlers = d->eventHandlers[eventName];
	}

	for (const auto &handler : handlers) {
		if (forceAsync || handler.isAsync) {
			if (priority == XNEvent::Priority::RealTime) {
				// 创建实时任务
				RTEventTask *task = new RTEventTask(eventName, eventData, handler.callback, this);
				d->rtManager.addTask(task);
			} else {
				// 普通异步任务使用线程池
				d->normalThreadPool.start(
					new EventTask(eventName, eventData, handler.callback, this));
			}
		} else {
			// 同步执行
			try {
				handler.callback(eventData);
				emit EventProcessed(eventName, true);
			} catch (const std::exception &e) {
				LOG_ERROR(QString("Exception in handler %1 for event %2: %3")
							  .arg(handler.GetHandlerId())
							  .arg(eventName)
							  .arg(e.what()));
				emit EventProcessed(eventName, false);
			}
		}
	}

	//LOG_INFO(QString("Triggered event: %1 with %2 handlers").arg(eventName).arg(handlers.size()));
}

void XNEventManager::SetMaxThreadCount(int count)
{
	Q_D(XNEventManager);
	d->threadPool.setMaxThreadCount(count);
	LOG_INFO(QString("Set thread pool max thread count to %1").arg(count));
}

int XNEventManager::GetMaxThreadCount() const
{
	Q_D(const XNEventManager);
	return d->threadPool.maxThreadCount();
}

void XNEventManager::WaitForAsyncEvents()
{
	Q_D(XNEventManager);
	d->threadPool.waitForDone();
	LOG_INFO("All async events have been processed");
}

// 初始化事件管理器
void XNEventManager::OnInitialize()
{
	Q_D(XNEventManager);

	// 配置普通线程池
	d->threadPool.setMaxThreadCount(QThread::idealThreadCount());

	// 配置实时线程池
	SetRTThreadPoolConfig(2,								   // 最大线程数
						  sched_get_priority_min(SCHED_FIFO),  // 最小优先级
						  sched_get_priority_max(SCHED_FIFO)); // 最大优先级

	LOG_INFO("XNEventManager Initialize Success!");
	d->_status = XNFrameObjectStatus::Initialized;
	emit Initialize();
}

// 准备执行
void XNEventManager::OnPrepareForExecute()
{
	Q_D(XNEventManager);
	// 设置状态为就绪
	d->_status = XNFrameObjectStatus::Ready;
	LOG_INFO("XNEventManager is prepared!");
	// 发送准备完成信号
	emit PrepareForExecute();
}

void XNEventManager::SetRTThreadPoolConfig(int maxThreads, int minPriority, int maxPriority)
{
	Q_D(XNEventManager);
	d->rtManager.stop();

	XNFramework *framework = qobject_cast<XNFramework *>(parent());
	if (framework == nullptr) {
		LOG_WARNING("XNFramework is nullptr!");
		return;
	}

	quint32 cpuAffinity = framework->GetCpuAffinity();

	// 找到最后一个可用的CPU
	int lastCpu = -1;
	for (int i = 0; i < 32; i++) { // 假设最多32个CPU
		if (cpuAffinity & (1u << i)) {
			lastCpu = i;
		}
	}

	if (lastCpu < 0) {
		LOG_WARNING("No available CPU found in affinity mask, using default CPU 1");
		lastCpu = 1;
	} else {
		LOG_INFO(QString("RT thread bound to CPU %1").arg(lastCpu));
	}

	d->rtManager.start(maxThreads, maxPriority, lastCpu);
}