Use async/await in JobProcessor instead of Thread

ethanhann · ethanhann · commit fb748c68e1e0 · 2015-02-22T22:02:08.000-05:00
diff --git a/IntegrationEngine.Tests/EngineHostCompositionRootTest.cs b/IntegrationEngine.Tests/EngineHostCompositionRootTest.cs
@@ -56,9 +56,9 @@ public void ShouldSetupMessageQueueListener()
             Subject.Container.RegisterType<IRabbitMQConfiguration, RabbitMQConfiguration>(configName,
                 new InjectionConstructor(new ResolvedParameter<IEngineConfiguration>(), configName));
 
-            Subject.SetupThreadedListenerManager();
+            Subject.SetupMessageQueueListenerManager();
 
-            Subject.Container.Resolve<IThreadedListenerManager>();
+            Assert.That(Subject.MessageQueueListenerManager, Is.Not.Null);
         }
 
         [Test]
diff --git a/IntegrationEngine.Tests/IntegrationEngine.Tests.csproj b/IntegrationEngine.Tests/IntegrationEngine.Tests.csproj
@@ -135,8 +135,8 @@
     <Compile Include="Api\TriggerControllerBaseTest.cs" />
     <Compile Include="Api\TriggerStub.cs" />
     <Compile Include="EngineHostCompositionRootTest.cs" />
-    <Compile Include="JobProcessor\ThreadedListenerManagerTest.cs" />
     <Compile Include="Api\WebApiApplicationTest.cs" />
+    <Compile Include="JobProcessor\MessageQueueListenerManagerTest.cs" />
   </ItemGroup>
   <ItemGroup>
     <None Include="..\configuration\IntegrationEngine.json">
diff --git a/IntegrationEngine.Tests/JobProcessor/MessageQueueListenerManagerTest.cs b/IntegrationEngine.Tests/JobProcessor/MessageQueueListenerManagerTest.cs
@@ -0,0 +1,56 @@
+﻿using BeekmanLabs.UnitTesting;
+using IntegrationEngine.JobProcessor;
+using NUnit.Framework;
+using Moq;
+using System;
+using System.Threading;
+
+namespace IntegrationEngine.Tests
+{
+    public class MessageQueueListenerManagerTest : TestBase<MessageQueueListenerManager>
+    {
+        public Mock<MessageQueueListenerFactory> MockMessageQueueListenerFactory { get; set; }
+
+        [SetUp]
+        public void Setup()
+        {
+            MockMessageQueueListenerFactory = new Mock<MessageQueueListenerFactory>();
+            MockMessageQueueListenerFactory.Setup(x => x.CreateRabbitMQListener())
+                .Returns<IMessageQueueListener>(null);
+            Subject.MessageQueueListenerFactory = MockMessageQueueListenerFactory.Object;
+        }
+
+        [Test]
+        public void ShouldStartListener()
+        {
+            Subject.ListenerTaskCount = 1;
+
+            Subject.StartListener();
+
+            MockMessageQueueListenerFactory.Verify(x => x.CreateRabbitMQListener(), Times.Once);
+        }
+
+        [Test]
+        public void ShouldStartMultipleListeners()
+        {
+            var listenerTaskCount = 4;
+            Subject.ListenerTaskCount = listenerTaskCount;
+
+            Subject.StartListener();
+
+            MockMessageQueueListenerFactory.Verify(x => x.CreateRabbitMQListener(), 
+                Times.Exactly(listenerTaskCount));
+        }
+
+        [Test]
+        public void ShouldSetCancellationTokenOnDispose()
+        {
+            Subject.CancellationTokenSource = new CancellationTokenSource();
+
+            Subject.Dispose();
+
+            Assert.That(Subject.CancellationTokenSource.IsCancellationRequested, Is.True);
+        }
+    }
+}
+
diff --git a/IntegrationEngine.Tests/JobProcessor/ThreadedListenerManagerTest.cs b/IntegrationEngine.Tests/JobProcessor/ThreadedListenerManagerTest.cs
diff --git a/IntegrationEngine/EngineHost.cs b/IntegrationEngine/EngineHost.cs
@@ -30,7 +30,7 @@ public void Initialize(bool isThreadedListenerEnabled = true,
             engineHostCompositionRoot = new EngineHostCompositionRoot(AssembliesWithJobs) {
                 IsWebApiEnabled = isWebApiEnabled,
                 IsSchedulerEnabled = isSchedulerEnabled,
-                IsThreadedListenerEnabled = isThreadedListenerEnabled,
+                IsMessageQueueListenerManagerEnabled = isThreadedListenerEnabled,
             };
             engineHostCompositionRoot.Configure();
         }
diff --git a/IntegrationEngine/EngineHostCompositionRoot.cs b/IntegrationEngine/EngineHostCompositionRoot.cs
@@ -33,9 +33,10 @@ public class EngineHostCompositionRoot : IDisposable
         public IList<Type> IntegrationJobTypes { get; set; }
         public ILog Log { get; set; }
         public IWebApiApplication WebApiApplication { get; set; }
+        public IMessageQueueListenerManager MessageQueueListenerManager { get; set; }
         public bool IsWebApiEnabled { get; set; }
         public bool IsSchedulerEnabled { get; set; }
-        public bool IsThreadedListenerEnabled { get; set; }
+        public bool IsMessageQueueListenerManagerEnabled { get; set; }
 
         public EngineHostCompositionRoot()
         {}
@@ -63,8 +64,8 @@ public void Configure()
             RegisterIntegrationJobs();
             SetupRScriptRunner();
             SetupElasticsearchRepository();
-            if (IsThreadedListenerEnabled)
-                SetupThreadedListenerManager();
+            if (IsMessageQueueListenerManagerEnabled)
+                SetupMessageQueueListenerManager();
             if (IsSchedulerEnabled)
                 SetupEngineScheduler();
             if (IsWebApiEnabled)
@@ -173,21 +174,14 @@ public void RegisterIntegrationJobs()
             });
         }
 
-        public void SetupThreadedListenerManager()
+        public async void SetupMessageQueueListenerManager()
         {
             var config = Container.Resolve<IRabbitMQConfiguration>("DefaultRabbitMQ");
-            var rabbitMqListener = new RabbitMQListener() {
-                IntegrationJobTypes = IntegrationJobTypes,
-                MessageQueueConnection = new MessageQueueConnection(config),
-                RabbitMQConfiguration = config,
-                UnityContainer = Container,
-            };
-
-            var threadedListenerManager = new ThreadedListenerManager() {
-                MessageQueueListener = rabbitMqListener,
+            var messageQueueListenerFactory = new MessageQueueListenerFactory(Container, IntegrationJobTypes, config);
+            MessageQueueListenerManager = new MessageQueueListenerManager() {
+                MessageQueueListenerFactory = messageQueueListenerFactory,
             };
-            Container.RegisterInstance<IThreadedListenerManager>(threadedListenerManager);
-            threadedListenerManager.StartListener();
+            await MessageQueueListenerManager.StartListener();
         }
             
         public void SetupEngineScheduler()
@@ -241,6 +235,8 @@ public void Dispose()
         {
             if (WebApiApplication != null)
                 WebApiApplication.Dispose();
+            if (MessageQueueListenerManager != null)
+                MessageQueueListenerManager.Dispose();
         }
     }
 }
diff --git a/IntegrationEngine/IntegrationEngine.csproj b/IntegrationEngine/IntegrationEngine.csproj
@@ -176,12 +176,13 @@
     <Compile Include="Api\TriggerControllerBase.cs" />
     <Compile Include="Api\Controllers\LogEventController.cs" />
     <Compile Include="EngineHostCompositionRoot.cs" />
-    <Compile Include="JobProcessor\IThreadedListenerManager.cs" />
     <Compile Include="JobProcessor\IMessageQueueListener.cs" />
     <Compile Include="JobProcessor\MsmqListener.cs" />
     <Compile Include="JobProcessor\RabbitMQListener.cs" />
     <Compile Include="EngineConfiguration.cs" />
-    <Compile Include="JobProcessor\ThreadedListenerManager.cs" />
+    <Compile Include="JobProcessor\IMessageQueueListenerManager.cs" />
+    <Compile Include="JobProcessor\MessageQueueListenerManager.cs" />
+    <Compile Include="JobProcessor\MessageQueueListenerFactory.cs" />
   </ItemGroup>
   <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
   <ItemGroup>
diff --git a/IntegrationEngine/JobProcessor/IMessageQueueListener.cs b/IntegrationEngine/JobProcessor/IMessageQueueListener.cs
@@ -5,6 +5,6 @@ namespace IntegrationEngine.JobProcessor
 {
     public interface IMessageQueueListener : IDisposable
     {
-        void Listen(CancellationToken cancellationToken);
+        void Listen(CancellationToken cancellationToken, int listenerId);
     }
 }
diff --git a/IntegrationEngine/JobProcessor/IMessageQueueListenerManager.cs b/IntegrationEngine/JobProcessor/IMessageQueueListenerManager.cs
@@ -14,14 +14,15 @@
 using System.Reflection;
 using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace IntegrationEngine.JobProcessor
 {
-    public interface IThreadedListenerManager : IDisposable
+    public interface IMessageQueueListenerManager : IDisposable
     {
         CancellationTokenSource CancellationTokenSource { get; set; }
-        IMessageQueueListener MessageQueueListener { get; set; }
+        MessageQueueListenerFactory MessageQueueListenerFactory { get; set; }
         ILog Log { get; set; }
-        void StartListener();
+        Task StartListener();
     }
 }
diff --git a/IntegrationEngine/JobProcessor/MessageQueueListenerFactory.cs b/IntegrationEngine/JobProcessor/MessageQueueListenerFactory.cs
@@ -0,0 +1,37 @@
+﻿using IntegrationEngine.Core.Configuration;
+using IntegrationEngine.Core.MessageQueue;
+using Microsoft.Practices.Unity;
+using System;
+using System.Collections.Generic;
+
+namespace IntegrationEngine.JobProcessor
+{
+    public class MessageQueueListenerFactory
+    {
+        public IUnityContainer UnityContainer { get; set; }
+        public IList<Type> IntegrationJobTypes { get; set; }
+        public IRabbitMQConfiguration RabbitMQConfiguration { get; set; }
+
+        public MessageQueueListenerFactory()
+        {
+        }
+
+        public MessageQueueListenerFactory(IUnityContainer unityContainer, IList<Type> integrationJobTypes, IRabbitMQConfiguration rabbitMQConfiguration)
+            : this()
+        {
+            UnityContainer = unityContainer;
+            IntegrationJobTypes = integrationJobTypes;
+            RabbitMQConfiguration = rabbitMQConfiguration;
+        }
+
+        public virtual IMessageQueueListener CreateRabbitMQListener()
+        {
+            return new RabbitMQListener() {
+                IntegrationJobTypes = IntegrationJobTypes,
+                MessageQueueConnection = new MessageQueueConnection(RabbitMQConfiguration),
+                RabbitMQConfiguration = RabbitMQConfiguration,
+                UnityContainer = UnityContainer,
+            };
+        }
+    }
+}
diff --git a/IntegrationEngine/JobProcessor/MessageQueueListenerManager.cs b/IntegrationEngine/JobProcessor/MessageQueueListenerManager.cs
@@ -0,0 +1,50 @@
+﻿using Common.Logging;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace IntegrationEngine.JobProcessor
+{
+    public class MessageQueueListenerManager : IMessageQueueListenerManager
+    {
+        public MessageQueueListenerFactory MessageQueueListenerFactory { get; set; }
+        public ILog Log { get; set; }
+        public CancellationTokenSource CancellationTokenSource { get; set; }
+        public int ListenerTaskCount { get; set; }
+        IList<IMessageQueueListener> _listeners;
+
+        public MessageQueueListenerManager()
+        {
+            _listeners = new List<IMessageQueueListener>();
+            ListenerTaskCount = 2;
+            CancellationTokenSource = new CancellationTokenSource();
+            Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
+        }
+
+        public async Task StartListener()
+        {
+            var tasks = Enumerable.Range(0, ListenerTaskCount).Select((i) => {
+                return Task.Factory.StartNew(() => {
+                    var listener = MessageQueueListenerFactory.CreateRabbitMQListener();
+                    _listeners.Add(listener);
+                    listener.Listen(CancellationTokenSource.Token, i);
+                }, 
+                    TaskCreationOptions.LongRunning);
+            }).ToList();
+            await Task.WhenAll(tasks);
+        }
+
+        public void Dispose()
+        {
+            CancellationTokenSource.Cancel();
+            foreach (var listener in _listeners)
+                if (listener != null)
+                    listener.Dispose();
+        }
+    }
+}
+
diff --git a/IntegrationEngine/JobProcessor/MsmqListener.cs b/IntegrationEngine/JobProcessor/MsmqListener.cs
@@ -33,7 +33,7 @@ public string QueueName
             }
         }
 
-        public void Listen(CancellationToken cancellationToken)
+        public void Listen(CancellationToken cancellationToken, int listenerId)
         {
             cancellationToken.ThrowIfCancellationRequested();
             Message newMessage = MSMessageQueue.Receive();
diff --git a/IntegrationEngine/JobProcessor/RabbitMQListener.cs b/IntegrationEngine/JobProcessor/RabbitMQListener.cs
@@ -14,11 +14,13 @@
 using System.Text;
 using System.Threading;
 using Microsoft.Practices.Unity;
+using RabbitMQ.Client.Exceptions;
 
 namespace IntegrationEngine.JobProcessor
 {
     public class RabbitMQListener : IMessageQueueListener
     {
+        int _listenerId;
         public IUnityContainer UnityContainer { get; set; }
         public QueueingBasicConsumer Consumer { get; set; }
         public IList<Type> IntegrationJobTypes { get; set; }
@@ -40,14 +42,15 @@ public void Dispose()
                 Connection.Close();
         }
 
-        public void Listen(CancellationToken cancellationToken)
+        public void Listen(CancellationToken cancellationToken, int listenerId)
         {
-            Connection = MessageQueueConnection.GetConnection();
+            _listenerId = listenerId;
+            Connection = MessageQueueConnection.GetConnectionFactory().CreateConnection();
             using (var channel = Connection.CreateModel())
             {
                 Consumer = new QueueingBasicConsumer(channel);
                 channel.BasicConsume(RabbitMQConfiguration.QueueName, true, Consumer);
-                Log.Info(x => x("Waiting for messages..."));
+                Log.Info(x => x("(Id={0}) Waiting for messages...", _listenerId));
 
                 while (true)
                 {
@@ -80,7 +83,9 @@ public void Listen(CancellationToken cancellationToken)
                     }
                     catch (OperationCanceledException exception)
                     { 
-                        Log.Info(x => x("Message queue listener has gracefully shutdown.", RabbitMQConfiguration.QueueName), exception);
+                        Log.Info(x => x("Message queue listener (id: {0}), listening on queue \"{1}\", has gracefully shutdown.", 
+                            _listenerId,
+                            RabbitMQConfiguration.QueueName), exception);
                         return;
                     }
                     catch (IntegrationJobRunFailureException exception)
diff --git a/IntegrationEngine/JobProcessor/ThreadedListenerManager.cs b/IntegrationEngine/JobProcessor/ThreadedListenerManager.cs

Original file line number	Diff line number	Diff line change
`@@ -56,9 +56,9 @@ public void ShouldSetupMessageQueueListener()`
`56`	`56`	`Subject.Container.RegisterType<IRabbitMQConfiguration, RabbitMQConfiguration>(configName,`
`57`	`57`	`new InjectionConstructor(new ResolvedParameter<IEngineConfiguration>(), configName));`
`58`	`58`
`59`		`- Subject.SetupThreadedListenerManager();`
	`59`	`+ Subject.SetupMessageQueueListenerManager();`
`60`	`60`
`61`		`- Subject.Container.Resolve<IThreadedListenerManager>();`
	`61`	`+ Assert.That(Subject.MessageQueueListenerManager, Is.Not.Null);`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`[Test]`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ public void Initialize(bool isThreadedListenerEnabled = true,`
`30`	`30`	`engineHostCompositionRoot = new EngineHostCompositionRoot(AssembliesWithJobs) {`
`31`	`31`	`IsWebApiEnabled = isWebApiEnabled,`
`32`	`32`	`IsSchedulerEnabled = isSchedulerEnabled,`
`33`		`- IsThreadedListenerEnabled = isThreadedListenerEnabled,`
	`33`	`+ IsMessageQueueListenerManagerEnabled = isThreadedListenerEnabled,`
`34`	`34`	`};`
`35`	`35`	`engineHostCompositionRoot.Configure();`
`36`	`36`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,6 @@ namespace IntegrationEngine.JobProcessor`
`5`	`5`	`{`
`6`	`6`	`public interface IMessageQueueListener : IDisposable`
`7`	`7`	`{`
`8`		`- void Listen(CancellationToken cancellationToken);`
	`8`	`+ void Listen(CancellationToken cancellationToken, int listenerId);`
`9`	`9`	`}`
`10`	`10`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,14 +14,15 @@`
`14`	`14`	`using System.Reflection;`
`15`	`15`	`using System.Text;`
`16`	`16`	`using System.Threading;`
	`17`	`+using System.Threading.Tasks;`
`17`	`18`
`18`	`19`	`namespace IntegrationEngine.JobProcessor`
`19`	`20`	`{`
`20`		`- public interface IThreadedListenerManager : IDisposable`
	`21`	`+ public interface IMessageQueueListenerManager : IDisposable`
`21`	`22`	`{`
`22`	`23`	`CancellationTokenSource CancellationTokenSource { get; set; }`
`23`		`- IMessageQueueListener MessageQueueListener { get; set; }`
	`24`	`+ MessageQueueListenerFactory MessageQueueListenerFactory { get; set; }`
`24`	`25`	`ILog Log { get; set; }`
`25`		`- void StartListener();`
	`26`	`+ Task StartListener();`
`26`	`27`	`}`
`27`	`28`	`}`