Fix async queue processing issues

Author: ethanhann

src/InEngine.Core/Commands/AlwaysFail.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
 
 namespace InEngine.Core.Commands;
@@ -8,7 +9,7 @@ namespace InEngine.Core.Commands;
 /// </summary>
 public class AlwaysFail : AbstractCommand
 {
-    public override void Run()
+    public override async Task RunAsync()
     {
         throw new CommandFailedException("This command always fails.");
     }

src/InEngine.Core/Commands/Sleep.cs

@@ -8,7 +8,7 @@ namespace InEngine.Core.Commands;
 public class Sleep : AbstractCommand
 {
     [Option("duration", HelpText = "The number of seconds to sleep.")]
-    public int DurationInSeconds { get; set; }
+    public int DurationInSeconds { get; set; } = 3;
 
     public override async Task RunAsync()
     {

src/InEngine.Core/InEngine.Core.csproj

@@ -27,7 +27,7 @@
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="2.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="2.0.0" />
     <PackageReference Include="Serilog" Version="2.11.0" />
-    <PackageReference Include="StackExchange.Redis" Version="1.2.4" />
+    <PackageReference Include="StackExchange.Redis" Version="2.6.48" />
     <PackageReference Include="Goblinfactory.Konsole" Version="6.2.2" />
     <PackageReference Include="RestSharp" Version="106.12.0" />
     <PackageReference Include="MailKit" Version="1.22.0" />

src/InEngine.Core/Queuing/Clients/FileClient.cs

@@ -64,8 +64,7 @@ public async Task Consume(CancellationToken cancellationToken)
             {
                 try
                 {
-                    if (await Consume() == null)
-                        Thread.Sleep(5000);
+                    await Consume();
                 }
                 catch (Exception exception)
                 {
@@ -108,7 +107,7 @@ public async Task<ICommandEnvelope> Consume()
 
         ConsumeLock.ReleaseMutex();
 
-        var commandEnvelope = File.ReadAllText(inProgressFilePath).DeserializeFromJson<CommandEnvelope>();
+        var commandEnvelope = (await File.ReadAllTextAsync(inProgressFilePath)).DeserializeFromJson<CommandEnvelope>();
         var command = commandEnvelope.GetCommandInstanceAndIncrementRetry(() =>
         {
             File.Move(inProgressFilePath, Path.Combine(FailedQueuePath, fileInfo.Name));

src/InEngine.Core/Queuing/Clients/RedisClient.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
 using InEngine.Core.IO;
@@ -29,17 +30,16 @@ public class RedisClient : IQueueClient
 
     public string FailedQueueName => QueueBaseName + $":{QueueName}:{QueueNames.Failed}";
 
-    public static Lazy<ConnectionMultiplexer> lazyConnection = new Lazy<ConnectionMultiplexer>(() =>
+    public static readonly Lazy<ConnectionMultiplexer> LazyConnection = new(() =>
     {
         var redisConfig = ConfigurationOptions.Parse($"{ClientSettings.Host}:{ClientSettings.Port}");
         redisConfig.Password = string.IsNullOrWhiteSpace(ClientSettings.Password) ? null : ClientSettings.Password;
         redisConfig.AbortOnConnectFail = false;
         return ConnectionMultiplexer.Connect(redisConfig);
     });
 
-    public static ConnectionMultiplexer Connection => lazyConnection.Value;
+    public static ConnectionMultiplexer Connection => LazyConnection.Value;
 
-    public ConnectionMultiplexer _connectionMultiplexer;
     private bool isDisposed;
 
     public IDatabase Redis => Connection.GetDatabase(ClientSettings.Database);
@@ -87,12 +87,11 @@ public async Task Consume(CancellationToken cancellationToken)
         try
         {
             InitChannel();
-            Connection.GetSubscriber().Subscribe(RedisChannel,
-                delegate
-                {
-                    Task.Factory.StartNew(Consume, cancellationToken, TaskCreationOptions.LongRunning,
-                        TaskScheduler.Default);
-                });
+            var channelMessageQueue = await Connection.GetSubscriber().SubscribeAsync(RedisChannel);
+            channelMessageQueue.OnMessage(async _ =>
+            {
+                await Consume();
+            });
         }
         catch (OperationCanceledException exception)
         {
@@ -108,8 +107,7 @@ public async Task<ICommandEnvelope> Consume()
     {
         var rawRedisMessageValue = Redis.ListRightPopLeftPush(PendingQueueName, InProgressQueueName);
         var serializedMessage = rawRedisMessageValue.ToString();
-        if (serializedMessage == null)
-            return null;
+
         var commandEnvelope = serializedMessage.DeserializeFromJson<CommandEnvelope>();
         if (commandEnvelope == null)
             throw new CommandFailedException("Could not deserialize the command.");
@@ -189,11 +187,10 @@ public void Dispose()
 
     private void Dispose(bool disposing)
     {
-        if (isDisposed) 
+        if (isDisposed)
             return;
-        if (!disposing) 
+        if (!disposing)
             return;
-        _connectionMultiplexer?.Dispose();
         isDisposed = true;
     }
 }

src/InEngine.Core/Queuing/Dequeue.cs

@@ -25,29 +25,26 @@ public Dequeue()
 
     public async Task StartAsync()
     {
-        var allTasks = new List<Task>();
         Log.LogDebug("Start dequeue tasks for primary queue...");
-        allTasks.AddRange(MakeTasks(true, QueueSettings.PrimaryQueueConsumers));
+        await AddConsumers(false, QueueSettings.PrimaryQueueConsumers);
         Log.LogDebug("Start dequeue tasks for secondary queue...");
-        allTasks.AddRange(MakeTasks(false, QueueSettings.SecondaryQueueConsumers));
-        await Task.WhenAll(allTasks);
+        await AddConsumers(true, QueueSettings.SecondaryQueueConsumers);
 
         // Recover from restart, if necessary.
         QueueAdapter.Make(false, QueueSettings, MailSettings).Recover();
         QueueAdapter.Make(true, QueueSettings, MailSettings).Recover();
     }
 
-    IList<Task> MakeTasks(bool useSecondaryQueue = false, int numberOfTasks = 0)
+    private async Task AddConsumers(bool useSecondaryQueue = false, int numberOfTasks = 0)
     {
-        return Enumerable.Range(0, numberOfTasks).Select((i) => {
+        for (var i = 0; i < numberOfTasks; i++)
+        {
             Log.LogDebug("Registering Dequeuer {I}", i);
-            return Task.Factory.StartNew(() => {
-                var queue = QueueAdapter.Make(useSecondaryQueue, QueueSettings, MailSettings);
-                queue.Id = i;
-                queueAdapters.Add(queue);
-                queue.Consume(CancellationTokenSource.Token);
-            }, TaskCreationOptions.LongRunning);
-        }).ToList();
+            var queue = QueueAdapter.Make(useSecondaryQueue, QueueSettings, MailSettings);
+            queue.Id = i;
+            queueAdapters.Add(queue);
+            await queue.Consume(CancellationTokenSource.Token);
+        }
     }
 
     public void Dispose()

src/InEngine.Core/Queuing/Message/CommandEnvelope.cs

@@ -24,7 +24,7 @@ public AbstractCommand GetCommandInstanceAndIncrementRetry(Action actionOnFail =
         }
         catch (Exception exception)
         {
-            actionOnFail.Invoke();
+            actionOnFail?.Invoke();
             throw new CommandNotExtractableFromEnvelopeException(CommandClassName, exception);
         }
     }