Thread/AsyncEventDispatcher

Author: MorganCaron

modules/Thread/AsyncEventDispatcher.mpp

@@ -0,0 +1,43 @@
+export module CppUtils.Thread.AsyncEventDispatcher;
+
+import std;
+import CppUtils.Execution.EventDispatcher;
+import CppUtils.String.Hash;
+import CppUtils.Thread.ThreadPool;
+
+export namespace CppUtils::Thread
+{
+	class AsyncEventDispatcher final
+	{
+	public:
+		explicit AsyncEventDispatcher(
+			std::size_t numberThreads = std::thread::hardware_concurrency(),
+			std::function<void(std::exception_ptr)> onError = nullptr,
+			std::function<void()> finally = nullptr):
+			m_threadPool{numberThreads, std::move(onError), std::move(finally)}
+		{}
+
+		template<String::Hasher eventName = String::Hash{}>
+		inline auto subscribe(auto&& function) -> void
+		{
+			m_eventDispatcher.subscribe<eventName>(std::forward<decltype(function)>(function));
+		}
+
+		template<String::Hasher eventName = String::Hash{}, class Event = std::nullptr_t>
+		inline auto emit(Event&& event = nullptr) -> void
+		{
+			m_threadPool.call([this, event = std::forward<Event>(event)] {
+				m_eventDispatcher.emit<eventName>(event);
+			});
+		}
+
+		inline auto waitUntilFinished() -> void
+		{
+			m_threadPool.waitUntilFinished();
+		}
+
+	private:
+		ThreadPool m_threadPool;
+		Execution::EventDispatcher m_eventDispatcher;
+	};
+}

modules/Thread/Thread.mpp

@@ -7,4 +7,6 @@ export import CppUtils.Thread.ThreadLoop;
 export import CppUtils.Thread.ThreadPool;
 export import CppUtils.Thread.TryAsync;
 export import CppUtils.Thread.UniqueLocker;
+export import CppUtils.Thread.AsyncEventDispatcher;
 export import CppUtils.Thread.ScheduledEventDispatcher;
+export import CppUtils.Thread.Scheduler;

tests/Thread/AsyncEventDispatcher.mpp

@@ -0,0 +1,39 @@
+export module CppUtils.UnitTests.Thread.AsyncEventDispatcher;
+
+import std;
+import CppUtils;
+
+export namespace CppUtils::UnitTest::Thread::AsyncEventDispatcher
+{
+	inline auto _ = TestSuite{"Thread/AsyncEventDispatcher", {"Thread/ThreadPool", "Execution/EventDispatcher"}, [](auto& suite) {
+		suite.addTest("Emit and Wait", [&] {
+			auto dispatcher = CppUtils::Thread::AsyncEventDispatcher{};
+			std::atomic_bool received = false;
+
+			dispatcher.subscribe<"Event">([&](bool value) {
+				suite.expectEqual(value, true);
+				received = true;
+			});
+			dispatcher.emit<"Event">(true);
+
+			dispatcher.waitUntilFinished();
+			suite.expect(received);
+		});
+
+		suite.addTest("Error Handling", [&] {
+			std::atomic_bool errorCaught = false;
+			auto onError = [&](std::exception_ptr) {
+				errorCaught = true;
+			};
+			auto dispatcher = CppUtils::Thread::AsyncEventDispatcher{1, onError};
+
+			dispatcher.subscribe<"Event">([&](std::nullptr_t) {
+				throw std::runtime_error{"Test Error"};
+			});
+			dispatcher.emit<"Event">(nullptr);
+
+			dispatcher.waitUntilFinished();
+			suite.expect(errorCaught);
+		});
+	}};
+}

tests/UnitTests.mpp

@@ -34,8 +34,9 @@ export import CppUtils.UnitTests.String.Utility;
 export import CppUtils.UnitTests.System.Error;
 export import CppUtils.UnitTests.Terminal.Canvas;
 export import CppUtils.UnitTests.Terminal;
-export import CppUtils.UnitTests.Thread.Scheduler;
+export import CppUtils.UnitTests.Thread.AsyncEventDispatcher;
 export import CppUtils.UnitTests.Thread.ScheduledEventDispatcher;
+export import CppUtils.UnitTests.Thread.Scheduler;
 export import CppUtils.UnitTests.Thread.SharedLocker;
 export import CppUtils.UnitTests.Thread.ThreadLoop;
 export import CppUtils.UnitTests.Thread.ThreadPool;