Merge branch 'master' into develop

Author: Ethan Hann

IntegrationEngine.ConsoleHost/Program.cs

@@ -22,20 +22,21 @@ public static void Main(string[] args)
             {
                 Start(args);
                 Console.WriteLine("Press any key to stop...");
-                Console.ReadKey(true);
+                Console.ReadLine();
                 Stop();
             }
         }
 
-
         private static void Start(string[] args)
         {
             EngineHosts = new EngineHost(typeof(Program).Assembly);
             EngineHosts.Initialize();
         }
 
         private static void Stop()
-        {}
+        {
+            EngineHosts.Dispose();
+        }
 
         public class Service : ServiceBase
         {

IntegrationEngine/EngineHost.cs

@@ -5,7 +5,7 @@
 
 namespace IntegrationEngine
 {
-    public class EngineHost
+    public class EngineHost : IDisposable
     {
         EngineHostConfiguration _engineConfiguration;
         public IList<Assembly> AssembliesWithJobs { get; set; }
@@ -15,10 +15,10 @@ public EngineHost(params Assembly[] assembliesWithJobs)
             AssembliesWithJobs = assembliesWithJobs.ToList();
         }
 
-        ~EngineHost()
+        public void Dispose()
         {
-            if (_engineConfiguration == null)
-                _engineConfiguration.Shutdown();
+            if (_engineConfiguration != null)
+                _engineConfiguration.Dispose();
         }
 
         public void Initialize()

IntegrationEngine/EngineHostConfiguration.cs

@@ -20,7 +20,7 @@
 
 namespace IntegrationEngine
 {
-    public class EngineHostConfiguration
+    public class EngineHostConfiguration : IDisposable
     {
         public IUnityContainer Container { get; set; }
         public EngineConfiguration Configuration { get; set; }
@@ -113,6 +113,7 @@ public void SetupMessageQueueListener()
                 IntegrationEngineContext = Container.Resolve<IntegrationEngineContext>(),
                 ElasticClient = Container.Resolve<IElasticClient>(),
             };
+            Container.RegisterInstance<IMessageQueueListener>(rabbitMqListener);
             rabbitMqListener.Listen();
         }
 
@@ -176,10 +177,12 @@ public void SetupElasticClient()
             });
         }
 
-        public void Shutdown()
+        public void Dispose()
         {
             var engineScheduler = Container.Resolve<IEngineScheduler>();
             engineScheduler.Shutdown();
+            var messageQueueListener = Container.Resolve<IMessageQueueListener>();
+            messageQueueListener.Dispose();
         }
     }
 }

IntegrationEngine/MessageQueue/IMessageQueueListener.cs

@@ -6,7 +6,7 @@
 
 namespace IntegrationEngine.MessageQueue
 {
-    public interface IMessageQueueListener
+    public interface IMessageQueueListener : IDisposable
     {
         void Listen();
     }

IntegrationEngine/MessageQueue/MsmqListener.cs

@@ -54,6 +54,10 @@ public void Listen()
             }
         }
 
+        public void Dispose()
+        {
+            throw new NotImplementedException();
+        }
         T AutoWireJob<T>(T job, Type type)
         {
             if (type.GetInterface(typeof(IMailJob).Name) != null)

IntegrationEngine/MessageQueue/RabbitMQListener.cs

@@ -9,24 +9,26 @@
 using RabbitMQ.Client.Events;
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Text;
 using System.Threading;
 
 namespace IntegrationEngine.MessageQueue
 {
-    public class RabbitMQListener : IMessageQueueListener, IDisposable
+    public class RabbitMQListener : IMessageQueueListener
     {
         Thread listenerThread;
-        bool shouldTerminate;
+        volatile bool shouldTerminate;
+        QueueingBasicConsumer consumer;
         public IList<Type> IntegrationJobTypes { get; set; }
         public MessageQueueConfiguration MessageQueueConfiguration { get; set; }
         public MessageQueueConnection MessageQueueConnection { get; set; }
         public ILog Log { get; set; }
         public IMailClient MailClient { get; set; }
         public IntegrationEngineContext IntegrationEngineContext { get; set; }
         public IElasticClient ElasticClient { get; set; }
-        
+
         public RabbitMQListener()
         {
             shouldTerminate = false;
@@ -35,6 +37,8 @@ public RabbitMQListener()
         public void Dispose()
         {
             shouldTerminate = true;
+            if (consumer != null)
+                consumer.Queue.Close();
             listenerThread.Join();
         }
 
@@ -43,35 +47,44 @@ void _listen()
             var connection = MessageQueueConnection.GetConnection();
             using (var channel = connection.CreateModel())
             {
-                var consumer = new QueueingBasicConsumer(channel);
+                consumer = new QueueingBasicConsumer(channel);
                 channel.BasicConsume(MessageQueueConfiguration.QueueName, true, consumer);
-
                 Log.Info(x => x("Waiting for messages..."));
+
                 while (true)
                 {
-                    if (shouldTerminate)
-                        return;
-                    var eventArgs = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
-                    var body = eventArgs.Body;
-                    var message = JsonConvert.DeserializeObject<DispatchMessage>(Encoding.UTF8.GetString(body));
-                    Log.Debug(x => x("Message queue listener received {0}", message));
-                    if (IntegrationJobTypes != null && !IntegrationJobTypes.Any())
-                        continue;
-                    var type = IntegrationJobTypes.FirstOrDefault(t => t.FullName.Equals(message.JobTypeName));
-                    var integrationJob = Activator.CreateInstance(type) as IIntegrationJob;
-                    integrationJob = AutoWireJob(integrationJob, type);
+                    var message = new DispatchMessage();
                     try
                     {
+                        if (shouldTerminate)
+                        {
+                            connection.Close();
+                            Log.Info("Message queue listener has stopped listening for messages.");
+                            return;
+                        }
+                        var eventArgs = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
+                        var body = eventArgs.Body;
+                        message = JsonConvert.DeserializeObject<DispatchMessage>(Encoding.UTF8.GetString(body));
+                        Log.Debug(x => x("Message queue listener received {0}", message));
+                        if (IntegrationJobTypes != null && !IntegrationJobTypes.Any())
+                            continue;
+                        var type = IntegrationJobTypes.FirstOrDefault(t => t.FullName.Equals(message.JobTypeName));
+                        var integrationJob = Activator.CreateInstance(type) as IIntegrationJob;
+                        integrationJob = AutoWireJob(integrationJob, type);
                         if (integrationJob != null)
                         {
                             if (integrationJob.GetType() is IParameterizedJob)
                                 (integrationJob as IParameterizedJob).Parameters = message.Parameters;
                             integrationJob.Run();
                         }
                     }
+                    catch (EndOfStreamException exception)
+                    {
+                        Log.Debug(x => x("The message queue ({0}) has closed.", MessageQueueConfiguration.QueueName), exception);
+                    }
                     catch (Exception exception)
                     {
-                        Log.Error(x => x("Integration job did not run successfully ({0})}", message), exception);
+                        Log.Error(x => x("Integration job did not run successfully ({0})}", message.JobTypeName), exception);
                     }
                 }
             }