event dispatcher

apps/LoopThroughWithOpenGLCompositing/runtime/events/RuntimeEventDispatcher.h

@@ -0,0 +1,117 @@
+#pragma once
+
+#include "RuntimeEventQueue.h"
+
+#include <atomic>
+#include <functional>
+#include <map>
+#include <mutex>
+#include <vector>
+
+struct RuntimeEventDispatchResult
+{
+	std::size_t dispatchedEvents = 0;
+	std::size_t handlerInvocations = 0;
+	std::size_t handlerFailures = 0;
+};
+
+class RuntimeEventDispatcher
+{
+public:
+	using Handler = std::function<void(const RuntimeEvent&)>;
+
+	explicit RuntimeEventDispatcher(std::size_t queueCapacity = 1024) :
+		mQueue(queueCapacity)
+	{
+	}
+
+	bool Publish(RuntimeEvent event)
+	{
+		if (!event.PayloadMatchesType())
+			return false;
+
+		if (event.sequence == 0)
+			event.sequence = mNextSequence.fetch_add(1);
+
+		return mQueue.Push(std::move(event));
+	}
+
+	template <typename Payload>
+	bool PublishPayload(Payload payload, std::string source = {})
+	{
+		return Publish(MakeRuntimeEvent(std::move(payload), std::move(source)));
+	}
+
+	void Subscribe(RuntimeEventType type, Handler handler)
+	{
+		std::lock_guard<std::mutex> lock(mHandlerMutex);
+		mHandlers[type].push_back(std::move(handler));
+	}
+
+	void SubscribeAll(Handler handler)
+	{
+		std::lock_guard<std::mutex> lock(mHandlerMutex);
+		mAllHandlers.push_back(std::move(handler));
+	}
+
+	RuntimeEventDispatchResult DispatchPending(std::size_t maxEvents = 0)
+	{
+		RuntimeEventDispatchResult result;
+		std::vector<RuntimeEvent> events = mQueue.Drain(maxEvents);
+		result.dispatchedEvents = events.size();
+
+		for (const RuntimeEvent& event : events)
+		{
+			std::vector<Handler> handlers = HandlersFor(event.type);
+			result.handlerInvocations += handlers.size();
+
+			for (const Handler& handler : handlers)
+			{
+				try
+				{
+					handler(event);
+				}
+				catch (...)
+				{
+					++result.handlerFailures;
+				}
+			}
+		}
+
+		return result;
+	}
+
+	bool TryPop(RuntimeEvent& event)
+	{
+		return mQueue.TryPop(event);
+	}
+
+	RuntimeEventQueueMetrics GetQueueMetrics(std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now()) const
+	{
+		return mQueue.GetMetrics(now);
+	}
+
+	std::size_t QueueDepth() const
+	{
+		return mQueue.Depth();
+	}
+
+private:
+	std::vector<Handler> HandlersFor(RuntimeEventType type) const
+	{
+		std::lock_guard<std::mutex> lock(mHandlerMutex);
+		std::vector<Handler> handlers = mAllHandlers;
+
+		const auto found = mHandlers.find(type);
+		if (found != mHandlers.end())
+			handlers.insert(handlers.end(), found->second.begin(), found->second.end());
+
+		return handlers;
+	}
+
+	RuntimeEventQueue mQueue;
+	std::atomic<uint64_t> mNextSequence{ 1 };
+	mutable std::mutex mHandlerMutex;
+	std::map<RuntimeEventType, std::vector<Handler>> mHandlers;
+	std::vector<Handler> mAllHandlers;
+};