using Pulsar.Common.Messages.Other; using Pulsar.Common.Networking; using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; namespace Pulsar.Common.Messages { /// /// Handles registrations of s and processing of s. /// public static class MessageHandler { /// /// List of registered s. /// private static readonly List Processors = new List(); /// /// Used in lock statements to synchronize access to between threads. /// private static readonly object SyncLock = new object(); /// /// Registers a to the available . /// /// The to register. public static void Register(IMessageProcessor proc) { lock (SyncLock) { if (Processors.Contains(proc)) return; Processors.Add(proc); } } /// /// Unregisters a from the available . /// /// public static void Unregister(IMessageProcessor proc) { lock (SyncLock) { Processors.Remove(proc); } } /// /// Forwards the received to the appropriate s to execute it. /// /// The sender of the message. /// The received message. /// When true, executes each handler on the thread pool to avoid blocking the caller. public static void Process(ISender sender, IMessage msg, bool dispatchAsync = true) { if (msg == null) { return; } IEnumerable availableProcessors; lock (SyncLock) { // select appropriate message processors availableProcessors = Processors.Where(x => x.CanExecute(msg) && x.CanExecuteFrom(sender)).ToList(); // ToList() is required to retrieve a thread-safe enumerator representing a moment-in-time snapshot of the message processors } foreach (var executor in availableProcessors) { if (dispatchAsync) { QueueProcessorExecution(executor, sender, msg); } else { ExecuteProcessorSafely(executor, sender, msg); } } } private static void QueueProcessorExecution(IMessageProcessor processor, ISender sender, IMessage message) { var context = new MessageDispatchContext { Processor = processor, Sender = sender, Message = message }; ThreadPool.UnsafeQueueUserWorkItem(DispatchCallback, context); } private static void ExecuteProcessorSafely(IMessageProcessor processor, ISender sender, IMessage message) { try { processor.Execute(sender, message); } catch (Exception ex) { Debug.WriteLine($"[MessageHandler] Processor '{processor?.GetType().Name}' threw an exception: {ex}"); } } private static readonly WaitCallback DispatchCallback = state => { if (state is MessageDispatchContext context) { context.Invoke(); } }; private sealed class MessageDispatchContext { public IMessageProcessor Processor; public ISender Sender; public IMessage Message; public void Invoke() { try { Processor?.Execute(Sender, Message); } catch (Exception ex) { Debug.WriteLine($"[MessageHandler] Processor '{Processor?.GetType().Name}' threw an exception: {ex}"); } finally { Processor = null; Sender = null; Message = null; } } } } }