Switch to StackExchange.Redis

Author: ethanhann

docs-src/configuration.md

@@ -7,6 +7,8 @@ Configuration is accomplished by modifying the appsettings.json file that comes
 {
   "InEngine": {
     "Queue": {
+      "PrimaryQueueConsumers":  16,
+      "SecondaryQueueConsumers": -1,
       "QueueName": "InEngine:Queue",
       "RedisHost": "localhost",
       "RedisPort": 6379,
@@ -15,6 +17,7 @@ Configuration is accomplished by modifying the appsettings.json file that comes
     }
   }
 }
+
 ```
 
 The -c, --configuration argument can also be used to specify an alternate configuration file.

src/InEngine.Core/InEngine.Core.csproj

@@ -58,9 +58,6 @@
     <Reference Include="CommandLine">
       <HintPath>..\packages\CommandLineParser.1.9.71\lib\net45\CommandLine.dll</HintPath>
     </Reference>
-    <Reference Include="RedisBoost">
-      <HintPath>..\packages\RedisBoost.1.6.4\lib\net40\RedisBoost.dll</HintPath>
-    </Reference>
     <Reference Include="System.Configuration" />
     <Reference Include="Common.Logging.Core">
       <HintPath>..\packages\Common.Logging.Core.3.4.1\lib\net40\Common.Logging.Core.dll</HintPath>
@@ -102,6 +99,9 @@
       <HintPath>..\packages\System.IO.Compression.4.3.0\lib\net46\System.IO.Compression.dll</HintPath>
     </Reference>
     <Reference Include="mscorlib" />
+    <Reference Include="StackExchange.Redis">
+      <HintPath>..\packages\StackExchange.Redis.1.2.4\lib\net46\StackExchange.Redis.dll</HintPath>
+    </Reference>
   </ItemGroup>
   <ItemGroup>
     <Compile Include="Properties\AssemblyInfo.cs" />
@@ -135,6 +135,7 @@
     <Compile Include="Commands\Options.cs" />
     <Compile Include="IO\Write.cs" />
     <Compile Include="IO\IWrite.cs" />
+    <Compile Include="JsonExtensions.cs" />
   </ItemGroup>
   <ItemGroup>
     <None Include="packages.config" />

src/InEngine.Core/JsonExtensions.cs

@@ -0,0 +1,17 @@
+using Newtonsoft.Json;
+
+namespace InEngine.Core
+{
+    public static class JsonExtensions
+    {
+        public static string SerializeToJson<T>(this T message) where T : class
+        {
+            return JsonConvert.SerializeObject(message);
+        }
+
+        public static T DeserializeFromJson<T>(this string payload)
+        {
+            return JsonConvert.DeserializeObject<T>(payload);
+        }
+    }
+}

src/InEngine.Core/Queue/Broker.cs

@@ -3,7 +3,7 @@
 using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
 using Newtonsoft.Json;
-using RedisBoost;
+using StackExchange.Redis;
 
 namespace InEngine.Core.Queue
 {
@@ -15,16 +15,16 @@ public class Broker
         public string SecondaryWaitingQueueName { get { return QueueBaseName + ":SecondaryWaiting"; } }
         public string SecondaryProcessingQueueName { get { return QueueBaseName + ":SecondaryProcessing"; } }
 
-        public IRedisClient _redis;
-        public IRedisClient Redis
+        public ConnectionMultiplexer _redis;
+        public ConnectionMultiplexer Redis
         {
             get
             {
                 if (_redis == null)
                 {
-                    var connectionTask = RedisClient.ConnectAsync(RedisHost, RedisPort, RedisDb);
-                    connectionTask.Wait();
-                    _redis = connectionTask.Result;
+                    var redisConfig = ConfigurationOptions.Parse($"{RedisHost}:{RedisPort}");
+                    redisConfig.Password = string.IsNullOrWhiteSpace(RedisPassword) ? null : RedisPassword;
+                    _redis = ConnectionMultiplexer.Connect(redisConfig);
                 }
                 return _redis;
             }
@@ -49,22 +49,26 @@ public static Broker Make()
 
         public void Publish(ICommand command, bool useSecondaryQueue = false)
         {
-            Redis.LPushAsync(useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName,
-                new Message()
-                {
+            Redis.GetDatabase(RedisDb).ListLeftPush(
+                useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName,
+                new Message() {
                     CommandClassName = command.GetType().FullName,
                     CommandAssemblyName = command.GetType().Assembly.GetName().Name + ".dll",
                     SerializedCommand = JsonConvert.SerializeObject(command)
-                });
+                }.SerializeToJson()
+            );
         }
 
         public bool Consume(bool useSecondaryQueue = false)
         {
             var waitingQueueName = useSecondaryQueue ? SecondaryWaitingQueueName : PrimaryWaitingQueueName;
             var processingQueueName = useSecondaryQueue ? SecondaryProcessingQueueName : PrimaryProcessingQueueName;
-            var stageMessageTask = Redis.RPopLPushAsync(waitingQueueName, processingQueueName);
-            stageMessageTask.Wait();
-            var message = stageMessageTask.Result.As<Message>();
+
+            var stageMessageTask = Redis.GetDatabase(RedisDb).ListRightPopLeftPush(waitingQueueName, processingQueueName);
+            var serializedMessage = stageMessageTask.ToString();
+            if (serializedMessage == null)
+                return false;
+            var message = serializedMessage.DeserializeFromJson<Message>();
             if (message == null)
                 return false;
 
@@ -76,7 +80,7 @@ public bool Consume(bool useSecondaryQueue = false)
             try
             {
                 commandInstance.Run();
-                Redis.LRemAsync(processingQueueName, 1, message).Wait();
+                Redis.GetDatabase(RedisDb).ListRemove(processingQueueName, serializedMessage, 1);
             }
             catch (Exception exception)
             {
@@ -88,52 +92,45 @@ public bool Consume(bool useSecondaryQueue = false)
         #region Primary Queue Management Methods
         public long GetPrimaryWaitingQueueLength()
         {
-            return WaitAndReturnResult(Redis.LLenAsync(PrimaryWaitingQueueName));
+            return Redis.GetDatabase(RedisDb).ListLength(PrimaryWaitingQueueName);
         }
 
         public long GetPrimaryProcessingQueueLength()
         {
-            return WaitAndReturnResult(Redis.LLenAsync(PrimaryProcessingQueueName));
+            return Redis.GetDatabase(RedisDb).ListLength(PrimaryProcessingQueueName);
         }
 
-        public long ClearPrimaryWaitingQueue()
+        public bool ClearPrimaryWaitingQueue()
         {
-            return WaitAndReturnResult(Redis.DelAsync(SecondaryWaitingQueueName));
+            return Redis.GetDatabase(RedisDb).KeyDelete(PrimaryWaitingQueueName);
         }
 
-        public long ClearPrimaryProcessingQueue()
+        public bool ClearPrimaryProcessingQueue()
         {
-            return WaitAndReturnResult(Redis.DelAsync(SecondaryProcessingQueueName));
+            return Redis.GetDatabase(RedisDb).KeyDelete(PrimaryProcessingQueueName);
         }
         #endregion
 
         #region Secondary Queue Management Methods
         public long GetSecondaryWaitingQueueLength()
         {
-            return WaitAndReturnResult(Redis.LLenAsync(PrimaryWaitingQueueName));
+            return Redis.GetDatabase(RedisDb).ListLength(SecondaryWaitingQueueName);
         }
 
         public long GetSecondaryProcessingQueueLength()
         {
-            return WaitAndReturnResult(Redis.LLenAsync(PrimaryProcessingQueueName));
+            return Redis.GetDatabase(RedisDb).ListLength(SecondaryProcessingQueueName);
         }
 
-        public long ClearSecondaryWaitingQueue()
+        public bool ClearSecondaryWaitingQueue()
         {
-            return WaitAndReturnResult(Redis.DelAsync(SecondaryWaitingQueueName));
+            return Redis.GetDatabase(RedisDb).KeyDelete(SecondaryWaitingQueueName);
         }
 
-        public long ClearSecondaryProcessingQueue()
+        public bool ClearSecondaryProcessingQueue()
         {
-            return WaitAndReturnResult(Redis.DelAsync(SecondaryProcessingQueueName));
+            return Redis.GetDatabase(RedisDb).KeyDelete(SecondaryProcessingQueueName);
         }
         #endregion
-
-
-        public long WaitAndReturnResult(Task<long> task)
-        {
-            task.Wait();
-            return task.Result;
-        }
     }
 }

src/InEngine.Core/packages.config

@@ -18,7 +18,7 @@
   <package id="Newtonsoft.Json" version="10.0.3" targetFramework="net462" />
   <package id="NLog" version="4.4.12" targetFramework="net462" />
   <package id="Quartz" version="2.6.1" targetFramework="net462" />
-  <package id="RedisBoost" version="1.6.4" targetFramework="net462" />
+  <package id="StackExchange.Redis" version="1.2.4" targetFramework="net462" />
   <package id="System.IO.Compression" version="4.3.0" targetFramework="net462" />
   <package id="System.Runtime.CompilerServices.Unsafe" version="4.4.0" targetFramework="net462" />
 </packages>

src/InEngine/appsettings.json

@@ -1,8 +1,8 @@
 {
   "InEngine": {
     "Queue": {
-      "PrimaryQueueConsumers":  16,
-      "SecondaryQueueConsumers": -1,
+      "PrimaryQueueConsumers":  4,
+      "SecondaryQueueConsumers": 4,
       "QueueName": "InEngine:Queue",
       "RedisHost": "localhost",
       "RedisPort": 6379,