Radarr/NzbDrone.Common/Messaging/MessageAggregator.cs

114 lines
3.8 KiB
C#
Raw Normal View History

using System;
2013-05-13 02:52:55 +00:00
using System.Linq;
using System.Threading.Tasks;
2013-02-23 20:09:44 +00:00
using NLog;
2013-05-13 02:52:55 +00:00
using NzbDrone.Common.EnsureThat;
using NzbDrone.Common.Serializer;
2013-07-12 06:10:34 +00:00
using NzbDrone.Common.TPL;
namespace NzbDrone.Common.Messaging
{
public class MessageAggregator : IMessageAggregator
{
2013-02-23 20:09:44 +00:00
private readonly Logger _logger;
private readonly IServiceFactory _serviceFactory;
private readonly TaskFactory _taskFactory;
public MessageAggregator(Logger logger, IServiceFactory serviceFactory)
{
2013-02-23 20:09:44 +00:00
_logger = logger;
_serviceFactory = serviceFactory;
var scheduler = new LimitedConcurrencyLevelTaskScheduler(2);
_taskFactory = new TaskFactory(scheduler);
}
2013-05-13 02:52:55 +00:00
public void PublishEvent<TEvent>(TEvent @event) where TEvent : class ,IEvent
{
2013-05-13 02:52:55 +00:00
Ensure.That(() => @event).IsNotNull();
2013-05-11 20:06:57 +00:00
var eventName = GetEventName(@event.GetType());
_logger.Trace("Publishing {0}", eventName);
//call synchronous handlers first.
foreach (var handler in _serviceFactory.BuildAll<IHandle<TEvent>>())
{
try
{
2013-05-11 20:06:57 +00:00
_logger.Debug("{0} -> {1}", eventName, handler.GetType().Name);
handler.Handle(@event);
2013-05-11 20:06:57 +00:00
_logger.Debug("{0} <- {1}", eventName, handler.GetType().Name);
}
catch (Exception e)
{
2013-05-11 20:06:57 +00:00
_logger.ErrorException(string.Format("{0} failed while processing [{1}]", handler.GetType().Name, eventName), e);
}
}
foreach (var handler in _serviceFactory.BuildAll<IHandleAsync<TEvent>>())
{
var handlerLocal = handler;
_taskFactory.StartNew(() =>
{
2013-05-11 20:06:57 +00:00
_logger.Debug("{0} ~> {1}", eventName, handlerLocal.GetType().Name);
handlerLocal.HandleAsync(@event);
2013-05-11 20:06:57 +00:00
_logger.Debug("{0} <~ {1}", eventName, handlerLocal.GetType().Name);
2013-07-12 06:10:34 +00:00
}, TaskCreationOptions.PreferFairness)
.LogExceptions();
}
}
2013-05-11 20:06:57 +00:00
private static string GetEventName(Type eventType)
{
if (!eventType.IsGenericType)
{
return eventType.Name;
}
return string.Format("{0}<{1}>", eventType.Name.Remove(eventType.Name.IndexOf('`')), eventType.GetGenericArguments()[0].Name);
}
2013-05-13 02:52:55 +00:00
public void PublishCommand<TCommand>(TCommand command) where TCommand : class, ICommand
{
2013-05-13 02:52:55 +00:00
Ensure.That(() => command).IsNotNull();
var handlerContract = typeof(IExecute<>).MakeGenericType(command.GetType());
_logger.Trace("Publishing {0}", command.GetType().Name);
2013-05-21 02:49:08 +00:00
var handler = (IExecute<TCommand>)_serviceFactory.Build(handlerContract);
_logger.Debug("{0} -> {1}", command.GetType().Name, handler.GetType().Name);
2013-05-11 20:06:57 +00:00
try
{
2013-05-21 02:49:08 +00:00
handler.Execute(command);
PublishEvent(new CommandCompletedEvent(command));
}
2013-05-21 02:49:08 +00:00
catch (Exception e)
{
PublishEvent(new CommandFailedEvent(command, e));
throw;
}
finally
{
PublishEvent(new CommandExecutedEvent(command));
}
_logger.Debug("{0} <- {1}", command.GetType().Name, handler.GetType().Name);
}
2013-05-13 02:52:55 +00:00
public void PublishCommand(string commandTypeName)
{
var commandType = _serviceFactory.GetImplementations(typeof(ICommand))
.Single(c => c.FullName.Equals(commandTypeName, StringComparison.InvariantCultureIgnoreCase));
2013-05-21 02:49:08 +00:00
dynamic command = Json.Deserialize("{}", commandType);
PublishCommand(command);
2013-05-13 02:52:55 +00:00
}
}
}