split MessageAggregator in EventAggregator and CommandExecutor

NzbDrone.Core/Messaging/Commands/Command.cs

@@ -2,7 +2,7 @@ using System;
 using FluentMigrator.Runner;
 using NzbDrone.Common.Messaging;
 using NzbDrone.Core.Datastore;
-using NzbDrone.Core.Messaging.Tracking;
+using NzbDrone.Core.Messaging.Commands.Tracking;
 
 namespace NzbDrone.Core.Messaging.Commands
 {

NzbDrone.Core/Messaging/Commands/CommandExecutor.cs

@@ -4,80 +4,33 @@ using System.Threading.Tasks;
 using NLog;
 using NzbDrone.Common;
 using NzbDrone.Common.EnsureThat;
-using NzbDrone.Common.Messaging;
 using NzbDrone.Common.Serializer;
 using NzbDrone.Common.TPL;
-using NzbDrone.Core.Messaging.Commands;
+using NzbDrone.Core.Messaging.Commands.Tracking;
 using NzbDrone.Core.Messaging.Events;
-using NzbDrone.Core.Messaging.Tracking;
 using NzbDrone.Core.ProgressMessaging;
 
-namespace NzbDrone.Core.Messaging
+namespace NzbDrone.Core.Messaging.Commands
 {
-    public class MessageAggregator : IMessageAggregator
+    public class CommandExecutor : ICommandExecutor
     {
         private readonly Logger _logger;
         private readonly IServiceFactory _serviceFactory;
         private readonly ITrackCommands _trackCommands;
+        private readonly IEventAggregator _eventAggregator;
         private readonly TaskFactory _taskFactory;
 
-        public MessageAggregator(Logger logger, IServiceFactory serviceFactory, ITrackCommands trackCommands)
+        public CommandExecutor(Logger logger, IServiceFactory serviceFactory, ITrackCommands trackCommands, IEventAggregator eventAggregator)
         {
             var scheduler = new LimitedConcurrencyLevelTaskScheduler(3);
 
             _logger = logger;
             _serviceFactory = serviceFactory;
             _trackCommands = trackCommands;
+            _eventAggregator = eventAggregator;
             _taskFactory = new TaskFactory(scheduler);
         }
 
-        public void PublishEvent<TEvent>(TEvent @event) where TEvent : class ,IEvent
-        {
-            Ensure.That(() => @event).IsNotNull();
-
-            var eventName = GetEventName(@event.GetType());
-
-            _logger.Trace("Publishing {0}", eventName);
-
-            //call synchronous handlers first.
-            foreach (var handler in _serviceFactory.BuildAll<IHandle<TEvent>>())
-            {
-                try
-                {
-                    _logger.Trace("{0} -> {1}", eventName, handler.GetType().Name);
-                    handler.Handle(@event);
-                    _logger.Trace("{0} <- {1}", eventName, handler.GetType().Name);
-                }
-                catch (Exception e)
-                {
-                    _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(() =>
-                {
-                    _logger.Trace("{0} ~> {1}", eventName, handlerLocal.GetType().Name);
-                    handlerLocal.HandleAsync(@event);
-                    _logger.Trace("{0} <~ {1}", eventName, handlerLocal.GetType().Name);
-                }, TaskCreationOptions.PreferFairness)
-                .LogExceptions();
-            }
-        }
-
-        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);
-        }
-
         public void PublishCommand<TCommand>(TCommand command) where TCommand : Command
         {
             Ensure.That(() => command).IsNotNull();
@@ -148,7 +101,7 @@ namespace NzbDrone.Core.Messaging
             try
             {
                 _trackCommands.Start(command);
-                PublishEvent(new CommandUpdatedEvent(command));
+                _eventAggregator.PublishEvent(new CommandUpdatedEvent(command));
 
                 if (!MappedDiagnosticsContext.Contains("CommandId") && command.SendUpdatesToClient)
                 {
@@ -157,13 +110,10 @@ namespace NzbDrone.Core.Messaging
 
                 handler.Execute((TCommand)command);
                 _trackCommands.Completed(command);
-                PublishEvent(new CommandUpdatedEvent(command));
-
             }
             catch (Exception e)
             {
                 _trackCommands.Failed(command, e);
-                PublishEvent(new CommandUpdatedEvent(command));
                 throw;
             }
             finally
@@ -172,11 +122,10 @@ namespace NzbDrone.Core.Messaging
                 {
                     MappedDiagnosticsContext.Remove("CommandId");
                 }
+                _eventAggregator.PublishEvent(new CommandUpdatedEvent(command));
+                _eventAggregator.PublishEvent(new CommandExecutedEvent(command));
             }
 
-            PublishEvent(new CommandExecutedEvent(command));
-            PublishEvent(new CommandUpdatedEvent(command));
-
             _logger.Trace("{0} <- {1} [{2}]", command.GetType().Name, handler.GetType().Name, command.Runtime.ToString(""));
         }
     }

NzbDrone.Core/Messaging/Commands/ICommandExecutor.cs

@@ -0,0 +1,10 @@
+namespace NzbDrone.Core.Messaging.Commands
+{
+    public interface ICommandExecutor
+    {
+        void PublishCommand<TCommand>(TCommand command) where TCommand : Command;
+        void PublishCommand(string commandTypeName);
+        Command PublishCommandAsync<TCommand>(TCommand command) where TCommand : Command;
+        Command PublishCommandAsync(string commandTypeName);
+    }
+}

NzbDrone.Core/Messaging/Commands/IExecute.cs

@@ -1,6 +1,4 @@
-using NzbDrone.Core.Messaging.Commands;
-
-namespace NzbDrone.Core.Messaging
+namespace NzbDrone.Core.Messaging.Commands
 {
     public interface IExecute<TCommand> : IProcessMessage<TCommand> where TCommand : Command
     {

NzbDrone.Core/Messaging/Commands/Tracking/CommandStatus.cs

@@ -1,4 +1,4 @@
-namespace NzbDrone.Core.Messaging.Tracking
+namespace NzbDrone.Core.Messaging.Commands.Tracking
 {
     public enum CommandStatus
     {

NzbDrone.Core/Messaging/Commands/Tracking/CommandTrackingService.cs

@@ -2,9 +2,8 @@ using System;
 using System.Collections.Generic;
 using System.Linq;
 using NzbDrone.Common.Cache;
-using NzbDrone.Core.Messaging.Commands;
 
-namespace NzbDrone.Core.Messaging.Tracking
+namespace NzbDrone.Core.Messaging.Commands.Tracking
 {
     public interface ITrackCommands
     {

NzbDrone.Core/Messaging/Commands/Tracking/ExistingCommand.cs

@@ -1,7 +1,6 @@
 using System;
-using NzbDrone.Core.Messaging.Commands;
 
-namespace NzbDrone.Core.Messaging.Tracking
+namespace NzbDrone.Core.Messaging.Commands.Tracking
 {
     public class ExistingCommand
     {

NzbDrone.Core/Messaging/Commands/Tracking/TrackedCommandCleanupCommand.cs

@@ -0,0 +1,7 @@
+namespace NzbDrone.Core.Messaging.Commands.Tracking
+{
+    public class TrackedCommandCleanupCommand : Command
+    {
+ 
+    }
+}

NzbDrone.Core/Messaging/Events/EventAggregator.cs

@@ -0,0 +1,73 @@
+using System;
+using System.Threading.Tasks;
+using NLog;
+using NzbDrone.Common;
+using NzbDrone.Common.EnsureThat;
+using NzbDrone.Common.Messaging;
+using NzbDrone.Common.TPL;
+
+namespace NzbDrone.Core.Messaging.Events
+{
+    public class EventAggregator : IEventAggregator
+    {
+        private readonly Logger _logger;
+        private readonly IServiceFactory _serviceFactory;
+        private readonly TaskFactory _taskFactory;
+
+        public EventAggregator(Logger logger, IServiceFactory serviceFactory)
+        {
+            var scheduler = new LimitedConcurrencyLevelTaskScheduler(3);
+
+            _logger = logger;
+            _serviceFactory = serviceFactory;
+            _taskFactory = new TaskFactory(scheduler);
+        }
+
+        public void PublishEvent<TEvent>(TEvent @event) where TEvent : class ,IEvent
+        {
+            Ensure.That(() => @event).IsNotNull();
+
+            var eventName = GetEventName(@event.GetType());
+
+            _logger.Trace("Publishing {0}", eventName);
+
+            //call synchronous handlers first.
+            foreach (var handler in _serviceFactory.BuildAll<IHandle<TEvent>>())
+            {
+                try
+                {
+                    _logger.Trace("{0} -> {1}", eventName, handler.GetType().Name);
+                    handler.Handle(@event);
+                    _logger.Trace("{0} <- {1}", eventName, handler.GetType().Name);
+                }
+                catch (Exception e)
+                {
+                    _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(() =>
+                {
+                    _logger.Trace("{0} ~> {1}", eventName, handlerLocal.GetType().Name);
+                    handlerLocal.HandleAsync(@event);
+                    _logger.Trace("{0} <~ {1}", eventName, handlerLocal.GetType().Name);
+                }, TaskCreationOptions.PreferFairness)
+                .LogExceptions();
+            }
+        }
+
+        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);
+        }
+    }
+}

NzbDrone.Core/Messaging/Events/IEventAggregator.cs

@@ -0,0 +1,9 @@
+using NzbDrone.Common.Messaging;
+
+namespace NzbDrone.Core.Messaging.Events
+{
+    public interface IEventAggregator
+    {
+        void PublishEvent<TEvent>(TEvent @event) where TEvent : class,  IEvent;
+    }
+}

NzbDrone.Core/Messaging/Events/IHandle.cs

@@ -1,6 +1,6 @@
 using NzbDrone.Common.Messaging;
 
-namespace NzbDrone.Core.Messaging
+namespace NzbDrone.Core.Messaging.Events
 {
     public interface IHandle<TEvent> : IProcessMessage<TEvent> where TEvent : IEvent
     {

NzbDrone.Core/Messaging/IMessageAggregator.cs

@@ -1,17 +0,0 @@
-using NzbDrone.Common.Messaging;
-using NzbDrone.Core.Messaging.Commands;
-
-namespace NzbDrone.Core.Messaging
-{
-    /// <summary>
-    ///   Enables loosely-coupled publication of events.
-    /// </summary>
-    public interface IMessageAggregator
-    {
-        void PublishEvent<TEvent>(TEvent @event) where TEvent : class,  IEvent;
-        void PublishCommand<TCommand>(TCommand command) where TCommand : Command;
-        void PublishCommand(string commandTypeName);
-        Command PublishCommandAsync<TCommand>(TCommand command) where TCommand : Command;
-        Command PublishCommandAsync(string commandTypeName);
-    }
-}

NzbDrone.Core/Messaging/MessageExtensions.cs

@@ -1,18 +0,0 @@
-using System;
-using NzbDrone.Core.Messaging.Commands;
-
-namespace NzbDrone.Core.Messaging
-{
-    public static class MessageExtensions
-    {
-        public static string GetExecutorName(this Type commandType)
-        {
-            if (!typeof(Command).IsAssignableFrom(commandType))
-            {
-                throw new ArgumentException("commandType must implement ICommand");
-            }
-
-            return string.Format("I{0}Executor", commandType.Name);
-        }
-    }
-}

NzbDrone.Core/Messaging/Tracking/TrackedCommandCleanupCommand.cs

@@ -1,9 +0,0 @@
-using NzbDrone.Core.Messaging.Commands;
-
-namespace NzbDrone.Core.Messaging.Tracking
-{
-    public class TrackedCommandCleanupCommand : Command
-    {
- 
-    }
-}