Use a more reliable approach to managing Redis connection

Author: ethanhann

src/InEngine.Core/Queue/Broker.cs

@@ -15,36 +15,35 @@ public class Broker
         public string SecondaryWaitingQueueName { get { return QueueBaseName + ":SecondaryWaiting"; } }
         public string SecondaryProcessingQueueName { get { return QueueBaseName + ":SecondaryProcessing"; } }
 
+        private static Lazy<ConnectionMultiplexer> lazyConnection = new Lazy<ConnectionMultiplexer>(() => { 
+            var queueSettings = InEngineSettings.Make().Queue;
+            var redisConfig = ConfigurationOptions.Parse($"{queueSettings.RedisHost}:{queueSettings.RedisPort}");
+            redisConfig.Password = string.IsNullOrWhiteSpace(queueSettings.RedisPassword) ? 
+                null : 
+                queueSettings.RedisPassword;
+            redisConfig.AbortOnConnectFail = false;
+            return ConnectionMultiplexer.Connect(redisConfig); 
+        });
+
+        public static ConnectionMultiplexer Connection { get { return lazyConnection.Value; } } 
+            
         public ConnectionMultiplexer _connectionMultiplexer;
         public IDatabase Redis
         {
             get
             {
-                if (_connectionMultiplexer == null || !_connectionMultiplexer.IsConnected)
-                {
-                    var redisConfig = ConfigurationOptions.Parse($"{RedisHost}:{RedisPort}");
-                    redisConfig.Password = string.IsNullOrWhiteSpace(RedisPassword) ? null : RedisPassword;
-                    redisConfig.AbortOnConnectFail = false;
-                    _connectionMultiplexer = ConnectionMultiplexer.Connect(redisConfig);
-                }
-                return _connectionMultiplexer.GetDatabase(RedisDb);
+                return Connection.GetDatabase(RedisDb);
             }
         }
-        public string RedisHost { get; set; }
-        public int RedisDb { get; set; }
-        public int RedisPort { get; set; }
-        public string RedisPassword { get; set; }
+        public static string RedisHost { get; set; }
+        public static int RedisDb { get; set; }
+        public static int RedisPort { get; set; }
+        public static string RedisPassword { get; set; }
 
         public static Broker Make()
         {
-            var queueSettings = InEngineSettings.Make().Queue;
-            return new Broker()
-            {
-                QueueBaseName = queueSettings.QueueName,
-                RedisHost = queueSettings.RedisHost,
-                RedisPort = queueSettings.RedisPort,
-                RedisDb = queueSettings.RedisDb,
-                RedisPassword = queueSettings.RedisPassword,
+            return new Broker() {
+                QueueBaseName = InEngineSettings.Make().Queue.QueueName
             };
         }
 

src/InEngine.Core/Queue/Commands/Consume.cs

@@ -1,5 +1,6 @@
 using System;
 using CommandLine;
+using Quartz;
 
 namespace InEngine.Core.Queue.Commands
 {
@@ -19,5 +20,10 @@ public override void Run()
             while (shouldConsume)
                 shouldConsume = broker.Consume(UseSecondaryQueue) && ShouldConsumeAll;
         }
+
+        public override void Failed(Exception exception)
+        {
+            Write.Error(exception.Message);
+        }
     }
 }

src/InEngine.Core/Queue/Commands/Publish.cs

@@ -1,4 +1,5 @@
-using System.Linq;
+using System;
+using System.Linq;
 using CommandLine;
 using InEngine.Core.Exceptions;
 
@@ -31,5 +32,10 @@ public override void Run()
 
             Broker.Make().Publish(command, UseSecondaryQueue);
         }
+
+        public override void Failed(Exception exception)
+        {
+            Write.Error(exception.Message);
+        }
     }
 }

src/InEngine.Core/Queue/Jobs.cs

@@ -25,7 +25,8 @@ private void ScheduleQueueConsumerJobs(IScheduler scheduler, int consumers, bool
                 var consume = new Consume() {
                     ScheduleId = $"{(useSecondaryQueue ? "secondary" : "primary")}:{index.ToString()}"
                 };
-                var job = consume.MakeJobBuilder().RequestRecovery(true).Build();
+
+                var job = consume.MakeJobBuilder().Build();
                 job.JobDataMap.Add("useSecondaryQueue", useSecondaryQueue);
 
                 var trigger = consume

src/InEngine/NLog.config

@@ -11,6 +11,6 @@
   </targets>
 
   <rules>
-<!--    <logger name="*" minlevel="Error" writeTo="c" />-->
+<!--    <logger name="*" minlevel="Debug" writeTo="c" />-->
   </rules>
 </nlog>