EventQueue.mpp

export module CppUtils.Execution.EventQueue;

import std;
import CppUtils.String.Hash;
import CppUtils.Execution.EventDispatcher;
import CppUtils.Execution.ScopeGuard;
import CppUtils.Thread.ThreadLoop;
import CppUtils.Thread.UniqueLocker;
import CppUtils.Chrono.Concept;

export namespace CppUtils::Execution
{
	class EventQueue
	{
	public:
		inline EventQueue():
			m_worker{
				[this] { workerThread(); },
				[this] { m_condition.notify_all(); }}
		{
			m_worker.start();
		}

		inline ~EventQueue()
		{
			waitUntilFinished();
		}

		EventQueue(const EventQueue&) = delete;
		EventQueue& operator=(const EventQueue&) = delete;
		EventQueue(EventQueue&&) = delete;
		EventQueue& operator=(EventQueue&&) = delete;

		template<String::Hasher eventName = Type::Hash{}, class... Args>
		inline auto emit(Args&&... args) -> void
		{
			emit(static_cast<Type::Hash>(eventName), std::forward<Args>(args)...);
		}

		template<class... Args>
		inline auto emit(Type::Hash eventName, Args&&... args) -> void
		{
			auto task = [this, eventName, ... args = std::forward<Args>(args)]() mutable {
				m_dispatcher.emit(eventName, std::move(args)...);
			};
			enqueue(std::move(task));
		}

		template<String::Hasher eventName = Type::Hash{}>
		inline auto subscribe(auto&& function) -> void
		{
			m_dispatcher.subscribe<eventName>(std::forward<decltype(function)>(function));
		}

		template<Chrono::Duration Duration = std::chrono::milliseconds>
		inline auto waitUntilFinished(Duration timeout = Duration::zero()) -> bool
		{
			auto accessor = m_queue.access();
			auto predicate = [this, &accessor] { return std::empty(accessor.value()) and not m_isTaskRunning; };
			if (timeout == Duration::zero())
			{
				m_condition.wait(accessor.getLockGuard(), predicate);
				return true;
			}
			return m_condition.wait_for(accessor.getLockGuard(), timeout, predicate);
		}

	private:
		inline auto enqueue(std::function<void()> task) -> void
		{
			{
				auto accessor = m_queue.access();
				accessor.value().push(std::move(task));
			}
			m_condition.notify_one();
		}

		inline auto workerThread() -> void
		{
			auto task = std::function<void()>{};
			{
				auto accessor = m_queue.access();
				m_condition.wait(accessor.getLockGuard(), [this, &accessor] {
					return not std::empty(accessor.value()) or m_worker.isStopRequested();
				});

				if (std::empty(accessor.value()))
					return;

				m_isTaskRunning = true;
				task = std::move(accessor.value().front());
				accessor.value().pop();
			}
			if (task)
			{
				auto _ = ScopeGuard{[this] { m_isTaskRunning = false; }};
				task();
			}
			m_condition.notify_all();
		}

		EventDispatcher m_dispatcher;
		Thread::UniqueLocker<std::queue<std::function<void()>>> m_queue;
		std::condition_variable m_condition;
		Thread::ThreadLoop m_worker;
		std::atomic<bool> m_isTaskRunning = false;
	};
}