Better names, more info, not using events
This commit is contained in:
parent
bb103947a2
commit
3c632743a1
|
@ -4,7 +4,7 @@ using Nancy;
|
||||||
using NzbDrone.Api.Extensions;
|
using NzbDrone.Api.Extensions;
|
||||||
using NzbDrone.Common.Composition;
|
using NzbDrone.Common.Composition;
|
||||||
using NzbDrone.Common.Messaging;
|
using NzbDrone.Common.Messaging;
|
||||||
using NzbDrone.Common.Messaging.Manager;
|
using NzbDrone.Common.Messaging.Tracking;
|
||||||
|
|
||||||
namespace NzbDrone.Api.Commands
|
namespace NzbDrone.Api.Commands
|
||||||
{
|
{
|
||||||
|
@ -12,13 +12,13 @@ namespace NzbDrone.Api.Commands
|
||||||
{
|
{
|
||||||
private readonly IMessageAggregator _messageAggregator;
|
private readonly IMessageAggregator _messageAggregator;
|
||||||
private readonly IContainer _container;
|
private readonly IContainer _container;
|
||||||
private readonly IManageCommands _commandManager;
|
private readonly ITrackCommands _trackCommands;
|
||||||
|
|
||||||
public CommandModule(IMessageAggregator messageAggregator, IContainer container, IManageCommands commandManager)
|
public CommandModule(IMessageAggregator messageAggregator, IContainer container, ITrackCommands trackCommands)
|
||||||
{
|
{
|
||||||
_messageAggregator = messageAggregator;
|
_messageAggregator = messageAggregator;
|
||||||
_container = container;
|
_container = container;
|
||||||
_commandManager = commandManager;
|
_trackCommands = trackCommands;
|
||||||
|
|
||||||
Post["/"] = x => RunCommand(ReadResourceFromRequest());
|
Post["/"] = x => RunCommand(ReadResourceFromRequest());
|
||||||
Get["/"] = x => GetAllCommands();
|
Get["/"] = x => GetAllCommands();
|
||||||
|
@ -39,7 +39,7 @@ namespace NzbDrone.Api.Commands
|
||||||
|
|
||||||
private Response GetAllCommands()
|
private Response GetAllCommands()
|
||||||
{
|
{
|
||||||
return _commandManager.Items.AsResponse();
|
return _trackCommands.AllTracked.AsResponse();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
|
@ -1,59 +0,0 @@
|
||||||
using System;
|
|
||||||
using System.Collections.Generic;
|
|
||||||
using System.Linq;
|
|
||||||
using NzbDrone.Common.Cache;
|
|
||||||
using NzbDrone.Common.Messaging.Events;
|
|
||||||
|
|
||||||
namespace NzbDrone.Common.Messaging.Manager
|
|
||||||
{
|
|
||||||
public interface IManageCommands
|
|
||||||
{
|
|
||||||
ICollection<CommandManagerItem> Items { get; }
|
|
||||||
Boolean ExistingItem(ICommand command);
|
|
||||||
}
|
|
||||||
|
|
||||||
public class CommandManager : IManageCommands,
|
|
||||||
IHandle<CommandStartedEvent>,
|
|
||||||
IHandle<CommandCompletedEvent>,
|
|
||||||
IHandle<CommandFailedEvent>
|
|
||||||
{
|
|
||||||
private readonly ICached<CommandManagerItem> _cache;
|
|
||||||
|
|
||||||
public CommandManager(ICacheManger cacheManger)
|
|
||||||
{
|
|
||||||
_cache = cacheManger.GetCache<CommandManagerItem>(GetType());
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Handle(CommandStartedEvent message)
|
|
||||||
{
|
|
||||||
_cache.Set(message.Command.CommandId, new CommandManagerItem(message.Command, CommandState.Running));
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Handle(CommandCompletedEvent message)
|
|
||||||
{
|
|
||||||
_cache.Set(message.Command.CommandId, new CommandManagerItem(message.Command, CommandState.Completed));
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Handle(CommandFailedEvent message)
|
|
||||||
{
|
|
||||||
_cache.Set(message.Command.CommandId, new CommandManagerItem(message.Command, CommandState.Failed));
|
|
||||||
}
|
|
||||||
|
|
||||||
public ICollection<CommandManagerItem> Items
|
|
||||||
{
|
|
||||||
get
|
|
||||||
{
|
|
||||||
return _cache.Values;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public bool ExistingItem(ICommand command)
|
|
||||||
{
|
|
||||||
var running = Items.Where(i => i.Type == command.GetType().FullName && i.State == CommandState.Running);
|
|
||||||
|
|
||||||
var result = running.Select(r => r.Command).Contains(command, new CommandEqualityComparer());
|
|
||||||
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -5,7 +5,7 @@ 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.Events;
|
||||||
using NzbDrone.Common.Messaging.Manager;
|
using NzbDrone.Common.Messaging.Tracking;
|
||||||
using NzbDrone.Common.Serializer;
|
using NzbDrone.Common.Serializer;
|
||||||
using NzbDrone.Common.TPL;
|
using NzbDrone.Common.TPL;
|
||||||
|
|
||||||
|
@ -15,14 +15,14 @@ namespace NzbDrone.Common.Messaging
|
||||||
{
|
{
|
||||||
private readonly Logger _logger;
|
private readonly Logger _logger;
|
||||||
private readonly IServiceFactory _serviceFactory;
|
private readonly IServiceFactory _serviceFactory;
|
||||||
private readonly IManageCommands _commandManager;
|
private readonly ITrackCommands _trackCommands;
|
||||||
private readonly TaskFactory _taskFactory;
|
private readonly TaskFactory _taskFactory;
|
||||||
|
|
||||||
public MessageAggregator(Logger logger, IServiceFactory serviceFactory, IManageCommands commandManager)
|
public MessageAggregator(Logger logger, IServiceFactory serviceFactory, ITrackCommands trackCommands)
|
||||||
{
|
{
|
||||||
_logger = logger;
|
_logger = logger;
|
||||||
_serviceFactory = serviceFactory;
|
_serviceFactory = serviceFactory;
|
||||||
_commandManager = commandManager;
|
_trackCommands = trackCommands;
|
||||||
var scheduler = new LimitedConcurrencyLevelTaskScheduler(2);
|
var scheduler = new LimitedConcurrencyLevelTaskScheduler(2);
|
||||||
_taskFactory = new TaskFactory(scheduler);
|
_taskFactory = new TaskFactory(scheduler);
|
||||||
}
|
}
|
||||||
|
@ -87,10 +87,13 @@ namespace NzbDrone.Common.Messaging
|
||||||
_logger.Debug("{0} -> {1}", command.GetType().Name, handler.GetType().Name);
|
_logger.Debug("{0} -> {1}", command.GetType().Name, handler.GetType().Name);
|
||||||
|
|
||||||
var sw = Stopwatch.StartNew();
|
var sw = Stopwatch.StartNew();
|
||||||
|
TrackedCommand queuedCommand = null;
|
||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
if (_commandManager.ExistingItem(command))
|
queuedCommand = _trackCommands.TrackIfNew(command);
|
||||||
|
|
||||||
|
if (queuedCommand == null)
|
||||||
{
|
{
|
||||||
_logger.Info("Command is already in progress: {0}", command.GetType().Name);
|
_logger.Info("Command is already in progress: {0}", command.GetType().Name);
|
||||||
return;
|
return;
|
||||||
|
@ -99,10 +102,17 @@ namespace NzbDrone.Common.Messaging
|
||||||
PublishEvent(new CommandStartedEvent(command));
|
PublishEvent(new CommandStartedEvent(command));
|
||||||
handler.Execute(command);
|
handler.Execute(command);
|
||||||
sw.Stop();
|
sw.Stop();
|
||||||
|
|
||||||
|
_trackCommands.Completed(queuedCommand, sw.Elapsed);
|
||||||
PublishEvent(new CommandCompletedEvent(command));
|
PublishEvent(new CommandCompletedEvent(command));
|
||||||
}
|
}
|
||||||
catch (Exception e)
|
catch (Exception e)
|
||||||
{
|
{
|
||||||
|
if (queuedCommand != null)
|
||||||
|
{
|
||||||
|
_trackCommands.Failed(queuedCommand, e);
|
||||||
|
}
|
||||||
|
|
||||||
PublishEvent(new CommandFailedEvent(command, e));
|
PublishEvent(new CommandFailedEvent(command, e));
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,78 @@
|
||||||
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Linq;
|
||||||
|
using NzbDrone.Common.Cache;
|
||||||
|
|
||||||
|
namespace NzbDrone.Common.Messaging.Tracking
|
||||||
|
{
|
||||||
|
public interface ITrackCommands
|
||||||
|
{
|
||||||
|
TrackedCommand TrackIfNew(ICommand command);
|
||||||
|
TrackedCommand Completed(TrackedCommand trackedCommand, TimeSpan runtime);
|
||||||
|
TrackedCommand Failed(TrackedCommand trackedCommand, Exception e);
|
||||||
|
ICollection<TrackedCommand> AllTracked { get; }
|
||||||
|
Boolean ExistingCommand(ICommand command);
|
||||||
|
}
|
||||||
|
|
||||||
|
public class TrackCommands : ITrackCommands
|
||||||
|
{
|
||||||
|
private readonly ICached<TrackedCommand> _cache;
|
||||||
|
|
||||||
|
public TrackCommands(ICacheManger cacheManger)
|
||||||
|
{
|
||||||
|
_cache = cacheManger.GetCache<TrackedCommand>(GetType());
|
||||||
|
}
|
||||||
|
|
||||||
|
public TrackedCommand TrackIfNew(ICommand command)
|
||||||
|
{
|
||||||
|
if (ExistingCommand(command))
|
||||||
|
{
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
var trackedCommand = new TrackedCommand(command, CommandState.Running);
|
||||||
|
_cache.Set(command.CommandId, trackedCommand);
|
||||||
|
|
||||||
|
return trackedCommand;
|
||||||
|
}
|
||||||
|
|
||||||
|
public TrackedCommand Completed(TrackedCommand trackedCommand, TimeSpan runtime)
|
||||||
|
{
|
||||||
|
trackedCommand.StateChangeTime = DateTime.UtcNow;
|
||||||
|
trackedCommand.State = CommandState.Completed;
|
||||||
|
trackedCommand.Runtime = runtime;
|
||||||
|
|
||||||
|
_cache.Set(trackedCommand.Command.CommandId, trackedCommand);
|
||||||
|
|
||||||
|
return trackedCommand;
|
||||||
|
}
|
||||||
|
|
||||||
|
public TrackedCommand Failed(TrackedCommand trackedCommand, Exception e)
|
||||||
|
{
|
||||||
|
trackedCommand.StateChangeTime = DateTime.UtcNow;
|
||||||
|
trackedCommand.State = CommandState.Failed;
|
||||||
|
trackedCommand.Exception = e;
|
||||||
|
|
||||||
|
_cache.Set(trackedCommand.Command.CommandId, trackedCommand);
|
||||||
|
|
||||||
|
return trackedCommand;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ICollection<TrackedCommand> AllTracked
|
||||||
|
{
|
||||||
|
get
|
||||||
|
{
|
||||||
|
return _cache.Values;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public bool ExistingCommand(ICommand command)
|
||||||
|
{
|
||||||
|
var running = AllTracked.Where(i => i.Type == command.GetType().FullName && i.State == CommandState.Running);
|
||||||
|
|
||||||
|
var result = running.Select(r => r.Command).Contains(command, new CommandEqualityComparer());
|
||||||
|
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -1,18 +1,22 @@
|
||||||
using System;
|
using System;
|
||||||
|
|
||||||
namespace NzbDrone.Common.Messaging.Manager
|
namespace NzbDrone.Common.Messaging.Tracking
|
||||||
{
|
{
|
||||||
public class CommandManagerItem
|
public class TrackedCommand
|
||||||
{
|
{
|
||||||
public String Type { get; private set; }
|
public String Type { get; private set; }
|
||||||
public ICommand Command { get; private set; }
|
public ICommand Command { get; private set; }
|
||||||
public CommandState State { get; set; }
|
public CommandState State { get; set; }
|
||||||
|
public DateTime StateChangeTime { get; set; }
|
||||||
|
public TimeSpan Runtime { get; set; }
|
||||||
|
public Exception Exception { get; set; }
|
||||||
|
|
||||||
public CommandManagerItem(ICommand command, CommandState state)
|
public TrackedCommand(ICommand command, CommandState state)
|
||||||
{
|
{
|
||||||
Type = command.GetType().FullName;
|
Type = command.GetType().FullName;
|
||||||
Command = command;
|
Command = command;
|
||||||
State = state;
|
State = state;
|
||||||
|
StateChangeTime = DateTime.UtcNow;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -92,8 +92,8 @@
|
||||||
<Compile Include="IEnumerableExtensions.cs" />
|
<Compile Include="IEnumerableExtensions.cs" />
|
||||||
<Compile Include="Instrumentation\GlobalExceptionHandlers.cs" />
|
<Compile Include="Instrumentation\GlobalExceptionHandlers.cs" />
|
||||||
<Compile Include="Instrumentation\ExceptronTarget.cs" />
|
<Compile Include="Instrumentation\ExceptronTarget.cs" />
|
||||||
<Compile Include="Messaging\Manager\CommandManager.cs" />
|
<Compile Include="Messaging\Tracking\CommandTrackingService.cs" />
|
||||||
<Compile Include="Messaging\Manager\CommandManagerItem.cs" />
|
<Compile Include="Messaging\Tracking\TrackedCommand.cs" />
|
||||||
<Compile Include="Messaging\Events\CommandStartedEvent.cs" />
|
<Compile Include="Messaging\Events\CommandStartedEvent.cs" />
|
||||||
<Compile Include="Messaging\CommandEqualityComparer.cs" />
|
<Compile Include="Messaging\CommandEqualityComparer.cs" />
|
||||||
<Compile Include="PathEqualityComparer.cs" />
|
<Compile Include="PathEqualityComparer.cs" />
|
||||||
|
|
Loading…
Reference in New Issue