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;
}
}
}
}
}