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 final
	{
	public:
		using SubscriptionIdentifier = EventDispatcher::SubscriptionIdentifier;

		inline EventQueue():
			m_worker{
				[this] { workerThread(); },
				[this] { m_workCondition.notify_all(); }}
		{
			m_worker.start();
		}

		inline ~EventQueue()
		{
			waitUntilFinished();
			m_worker.stop();
		}

		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... eventNames>
		inline auto subscribe(auto&& function)
		{
			return m_dispatcher.template subscribe<eventNames...>(std::forward<decltype(function)>(function));
		}

		template<String::Hasher... eventNames>
		inline auto unsubscribe() -> bool
		{
			return m_dispatcher.template unsubscribe<eventNames...>();
		}

		inline auto unsubscribe(auto&& subscriptionIdentifierOrEventHash)
		{
			return m_dispatcher.unsubscribe(std::forward<decltype(subscriptionIdentifierOrEventHash)>(subscriptionIdentifierOrEventHash));
		}

		inline auto clearSubscribers() -> void
		{
			m_dispatcher.clearSubscribers();
		}

		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::ranges::empty(accessor.value()) and not m_isTaskRunning; };
			if (timeout == Duration::zero())
			{
				m_finishedCondition.wait(accessor.getLockGuard(), predicate);
				return true;
			}
			return m_finishedCondition.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_workCondition.notify_one();
		}

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

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

				m_isTaskRunning = true;
				task = std::move(accessor.value().front());
				accessor.value().pop();
			}

			auto _ = ScopeGuard{[this] {
				{
					auto accessor = m_queue.access();
					m_isTaskRunning = false;
				}
				m_finishedCondition.notify_all();
			}};

			if (task)
				task();
		}

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