|
|
@ -4,6 +4,8 @@ using System.Linq;
|
|
|
|
using System.Threading.Tasks;
|
|
|
|
using System.Threading.Tasks;
|
|
|
|
using NLog;
|
|
|
|
using NLog;
|
|
|
|
using NzbDrone.Common.EnsureThat;
|
|
|
|
using NzbDrone.Common.EnsureThat;
|
|
|
|
|
|
|
|
using NzbDrone.Common.Messaging.Events;
|
|
|
|
|
|
|
|
using NzbDrone.Common.Messaging.Tracking;
|
|
|
|
using NzbDrone.Common.Serializer;
|
|
|
|
using NzbDrone.Common.Serializer;
|
|
|
|
using NzbDrone.Common.TPL;
|
|
|
|
using NzbDrone.Common.TPL;
|
|
|
|
|
|
|
|
|
|
|
@ -13,12 +15,14 @@ namespace NzbDrone.Common.Messaging
|
|
|
|
{
|
|
|
|
{
|
|
|
|
private readonly Logger _logger;
|
|
|
|
private readonly Logger _logger;
|
|
|
|
private readonly IServiceFactory _serviceFactory;
|
|
|
|
private readonly IServiceFactory _serviceFactory;
|
|
|
|
|
|
|
|
private readonly ITrackCommands _trackCommands;
|
|
|
|
private readonly TaskFactory _taskFactory;
|
|
|
|
private readonly TaskFactory _taskFactory;
|
|
|
|
|
|
|
|
|
|
|
|
public MessageAggregator(Logger logger, IServiceFactory serviceFactory)
|
|
|
|
public MessageAggregator(Logger logger, IServiceFactory serviceFactory, ITrackCommands trackCommands)
|
|
|
|
{
|
|
|
|
{
|
|
|
|
_logger = logger;
|
|
|
|
_logger = logger;
|
|
|
|
_serviceFactory = serviceFactory;
|
|
|
|
_serviceFactory = serviceFactory;
|
|
|
|
|
|
|
|
_trackCommands = trackCommands;
|
|
|
|
var scheduler = new LimitedConcurrencyLevelTaskScheduler(2);
|
|
|
|
var scheduler = new LimitedConcurrencyLevelTaskScheduler(2);
|
|
|
|
_taskFactory = new TaskFactory(scheduler);
|
|
|
|
_taskFactory = new TaskFactory(scheduler);
|
|
|
|
}
|
|
|
|
}
|
|
|
@ -60,7 +64,6 @@ namespace NzbDrone.Common.Messaging
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
private static string GetEventName(Type eventType)
|
|
|
|
private static string GetEventName(Type eventType)
|
|
|
|
{
|
|
|
|
{
|
|
|
|
if (!eventType.IsGenericType)
|
|
|
|
if (!eventType.IsGenericType)
|
|
|
@ -71,15 +74,69 @@ namespace NzbDrone.Common.Messaging
|
|
|
|
return string.Format("{0}<{1}>", eventType.Name.Remove(eventType.Name.IndexOf('`')), eventType.GetGenericArguments()[0].Name);
|
|
|
|
return string.Format("{0}<{1}>", eventType.Name.Remove(eventType.Name.IndexOf('`')), eventType.GetGenericArguments()[0].Name);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public void PublishCommand<TCommand>(TCommand command) where TCommand : class, ICommand
|
|
|
|
public void PublishCommand<TCommand>(TCommand command) where TCommand : class, ICommand
|
|
|
|
{
|
|
|
|
{
|
|
|
|
Ensure.That(() => command).IsNotNull();
|
|
|
|
Ensure.That(() => command).IsNotNull();
|
|
|
|
|
|
|
|
|
|
|
|
var handlerContract = typeof(IExecute<>).MakeGenericType(command.GetType());
|
|
|
|
_logger.Trace("Publishing {0}", command.GetType().Name);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
var trackedCommand = _trackCommands.TrackIfNew(command);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if (trackedCommand == null)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
_logger.Info("Command is already in progress: {0}", command.GetType().Name);
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
ExecuteCommand<TCommand>(trackedCommand);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public void PublishCommand(string commandTypeName)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
dynamic command = GetCommand(commandTypeName);
|
|
|
|
|
|
|
|
PublishCommand(command);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public TrackedCommand PublishCommandAsync<TCommand>(TCommand command) where TCommand : class, ICommand
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
Ensure.That(() => command).IsNotNull();
|
|
|
|
|
|
|
|
|
|
|
|
_logger.Trace("Publishing {0}", command.GetType().Name);
|
|
|
|
_logger.Trace("Publishing {0}", command.GetType().Name);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
var existingCommand = _trackCommands.TrackNewOrGet(command);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if (existingCommand.Existing)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
_logger.Info("Command is already in progress: {0}", command.GetType().Name);
|
|
|
|
|
|
|
|
return existingCommand.TrackedCommand;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_taskFactory.StartNew(() => ExecuteCommand<TCommand>(existingCommand.TrackedCommand)
|
|
|
|
|
|
|
|
, TaskCreationOptions.PreferFairness)
|
|
|
|
|
|
|
|
.LogExceptions();
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return existingCommand.TrackedCommand;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public TrackedCommand PublishCommandAsync(string commandTypeName)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
dynamic command = GetCommand(commandTypeName);
|
|
|
|
|
|
|
|
return PublishCommandAsync(command);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
private dynamic GetCommand(string commandTypeName)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
var commandType = _serviceFactory.GetImplementations(typeof(ICommand))
|
|
|
|
|
|
|
|
.Single(c => c.FullName.Equals(commandTypeName, StringComparison.InvariantCultureIgnoreCase));
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return Json.Deserialize("{}", commandType);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
private void ExecuteCommand<TCommand>(TrackedCommand trackedCommand) where TCommand : class, ICommand
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
var command = (TCommand)trackedCommand.Command;
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
var handlerContract = typeof(IExecute<>).MakeGenericType(command.GetType());
|
|
|
|
var handler = (IExecute<TCommand>)_serviceFactory.Build(handlerContract);
|
|
|
|
var handler = (IExecute<TCommand>)_serviceFactory.Build(handlerContract);
|
|
|
|
|
|
|
|
|
|
|
|
_logger.Debug("{0} -> {1}", command.GetType().Name, handler.GetType().Name);
|
|
|
|
_logger.Debug("{0} -> {1}", command.GetType().Name, handler.GetType().Name);
|
|
|
@ -88,30 +145,27 @@ namespace NzbDrone.Common.Messaging
|
|
|
|
|
|
|
|
|
|
|
|
try
|
|
|
|
try
|
|
|
|
{
|
|
|
|
{
|
|
|
|
|
|
|
|
MappedDiagnosticsContext.Set("CommandId", trackedCommand.Command.CommandId);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
PublishEvent(new CommandStartedEvent(trackedCommand));
|
|
|
|
handler.Execute(command);
|
|
|
|
handler.Execute(command);
|
|
|
|
sw.Stop();
|
|
|
|
sw.Stop();
|
|
|
|
PublishEvent(new CommandCompletedEvent(command));
|
|
|
|
|
|
|
|
|
|
|
|
_trackCommands.Completed(trackedCommand, sw.Elapsed);
|
|
|
|
|
|
|
|
PublishEvent(new CommandCompletedEvent(trackedCommand));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
catch (Exception e)
|
|
|
|
catch (Exception e)
|
|
|
|
{
|
|
|
|
{
|
|
|
|
PublishEvent(new CommandFailedEvent(command, e));
|
|
|
|
_trackCommands.Failed(trackedCommand, e);
|
|
|
|
|
|
|
|
PublishEvent(new CommandFailedEvent(trackedCommand, e));
|
|
|
|
throw;
|
|
|
|
throw;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
finally
|
|
|
|
finally
|
|
|
|
{
|
|
|
|
{
|
|
|
|
PublishEvent(new CommandExecutedEvent(command));
|
|
|
|
PublishEvent(new CommandExecutedEvent(trackedCommand));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
_logger.Debug("{0} <- {1} [{2}]", command.GetType().Name, handler.GetType().Name, sw.Elapsed.ToString(""));
|
|
|
|
_logger.Debug("{0} <- {1} [{2}]", command.GetType().Name, handler.GetType().Name, sw.Elapsed.ToString(""));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
public void PublishCommand(string commandTypeName)
|
|
|
|
|
|
|
|
{
|
|
|
|
|
|
|
|
var commandType = _serviceFactory.GetImplementations(typeof(ICommand))
|
|
|
|
|
|
|
|
.Single(c => c.FullName.Equals(commandTypeName, StringComparison.InvariantCultureIgnoreCase));
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
dynamic command = Json.Deserialize("{}", commandType);
|
|
|
|
|
|
|
|
PublishCommand(command);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|