Make queue clients consume methods async

ethanhann · ethanhann · commit ea7a0dec21cd · 2022-08-29T07:36:46.000-07:00
diff --git a/src/InEngine.Core/Queuing/Clients/FileClient.cs b/src/InEngine.Core/Queuing/Clients/FileClient.cs
@@ -3,6 +3,7 @@
 using System.IO;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
 using InEngine.Core.IO;
 using InEngine.Core.Queuing.Message;
@@ -55,15 +56,15 @@ private void PublishToQueue(CommandEnvelope commandEnvelope, string queuePath)
         );
     }
 
-    public void Consume(CancellationToken cancellationToken)
+    public async Task Consume(CancellationToken cancellationToken)
     {
         try
         {
             while (true)
             {
                 try
                 {
-                    if (Consume() == null)
+                    if (await Consume() == null)
                         Thread.Sleep(5000);
                 }
                 catch (Exception exception)
@@ -84,7 +85,7 @@ public void Consume(CancellationToken cancellationToken)
         }
     }
 
-    public ICommandEnvelope Consume()
+    public async Task<ICommandEnvelope> Consume()
     {
         FileInfo fileInfo;
         var inProgressFilePath = string.Empty;
@@ -116,7 +117,7 @@ public ICommandEnvelope Consume()
         try
         {
             command.WriteSummaryToConsole();
-            command.RunWithLifeCycleAsync().RunSynchronously();
+            await command.RunWithLifeCycleAsync();
         }
         catch (Exception exception)
         {
diff --git a/src/InEngine.Core/Queuing/Clients/RabbitMQClient.cs b/src/InEngine.Core/Queuing/Clients/RabbitMQClient.cs
@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 using InEngine.Core.Exceptions;
 using InEngine.Core.IO;
 using InEngine.Core.Queuing.Message;
@@ -94,13 +95,18 @@ public void Recover()
     {
     }
 
-    public void Consume(CancellationToken cancellationToken)
+    public async Task Consume(CancellationToken cancellationToken)
     {
         InitChannel();
         var consumer = new EventingBasicConsumer(Channel);
-        consumer.Received += (model, result) =>
+        consumer.Received += async (model, result) =>
         {
             var eventingConsumer = (EventingBasicConsumer)model;
+            if (eventingConsumer == null)
+            {
+                Log.LogWarning("EventingBasicConsumer is null while attempting to consume messages");
+                return;
+            }
 
             var serializedMessage = Encoding.UTF8.GetString(result.Body);
             var commandEnvelope = serializedMessage.DeserializeFromJson<CommandEnvelope>();
@@ -115,7 +121,7 @@ public void Consume(CancellationToken cancellationToken)
             try
             {
                 command.WriteSummaryToConsole();
-                command.RunWithLifeCycleAsync().RunSynchronously();
+                await command.RunWithLifeCycleAsync();
             }
             catch (Exception exception)
             {
@@ -135,7 +141,7 @@ public void Consume(CancellationToken cancellationToken)
         Channel.BasicConsume(queue: PendingQueueName, autoAck: false, consumer: consumer);
     }
 
-    public ICommandEnvelope Consume()
+    public async Task<ICommandEnvelope> Consume()
     {
         InitChannel();
         var result = Channel.BasicGet(PendingQueueName, false);
@@ -155,7 +161,7 @@ public ICommandEnvelope Consume()
         try
         {
             command.WriteSummaryToConsole();
-            command.RunWithLifeCycleAsync().RunSynchronously();
+            await command.RunWithLifeCycleAsync();
         }
         catch (Exception exception)
         {
diff --git a/src/InEngine.Core/Queuing/Clients/RedisClient.cs b/src/InEngine.Core/Queuing/Clients/RedisClient.cs
@@ -82,7 +82,7 @@ public void Recover()
             PublishToChannel();
     }
 
-    public void Consume(CancellationToken cancellationToken)
+    public async Task Consume(CancellationToken cancellationToken)
     {
         try
         {
@@ -104,7 +104,7 @@ public void Consume(CancellationToken cancellationToken)
         }
     }
 
-    public ICommandEnvelope Consume()
+    public async Task<ICommandEnvelope> Consume()
     {
         var rawRedisMessageValue = Redis.ListRightPopLeftPush(PendingQueueName, InProgressQueueName);
         var serializedMessage = rawRedisMessageValue.ToString();
@@ -122,7 +122,7 @@ public ICommandEnvelope Consume()
         try
         {
             command.WriteSummaryToConsole();
-            command.RunWithLifeCycleAsync().RunSynchronously();
+            await command.RunWithLifeCycleAsync();
         }
         catch (Exception exception)
         {
diff --git a/src/InEngine.Core/Queuing/Clients/SyncClient.cs b/src/InEngine.Core/Queuing/Clients/SyncClient.cs
@@ -1,6 +1,7 @@
 ﻿using System;
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using InEngine.Core.IO;
 using InEngine.Core.Queuing.Message;
 
@@ -43,12 +44,12 @@ public void Recover()
     {
     }
 
-    public void Consume(CancellationToken cancellationToken)
+    public async Task Consume(CancellationToken cancellationToken)
     {
         throw new NotImplementedException();
     }
 
-    public ICommandEnvelope Consume()
+    public async Task<ICommandEnvelope> Consume()
     {
         throw new NotImplementedException();
     }
diff --git a/src/InEngine.Core/Queuing/IQueueClient.cs b/src/InEngine.Core/Queuing/IQueueClient.cs
@@ -1,5 +1,6 @@
 ﻿using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using InEngine.Core.Queuing.Message;
 using InEngine.Core.IO;
 using Microsoft.Extensions.Logging;
@@ -16,8 +17,8 @@ public interface IQueueClient : IHasMailSettings, IDisposable
     string QueueName { get; set; }
     bool UseCompression { get; set; }
     void Publish(AbstractCommand command);
-    void Consume(CancellationToken cancellationToken);
-    ICommandEnvelope Consume();
+    Task Consume(CancellationToken cancellationToken);
+    Task<ICommandEnvelope> Consume();
     void Recover();
     Dictionary<string, long> GetQueueLengths();
     bool ClearPendingQueue();
diff --git a/src/InEngine.Core/Queuing/QueueAdapter.cs b/src/InEngine.Core/Queuing/QueueAdapter.cs
@@ -1,6 +1,7 @@
 ﻿using System;
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using InEngine.Core.IO;
 using InEngine.Core.Queuing.Clients;
 using InEngine.Core.Queuing.Message;
@@ -89,8 +90,8 @@ public static QueueAdapter Make(bool useSecondaryQueue, QueueSettings queueSetti
     }
 
     public void Publish(AbstractCommand command) => QueueClient.Publish(command);
-    public void Consume(CancellationToken cancellationToken) => QueueClient.Consume(cancellationToken);
-    public ICommandEnvelope Consume() => QueueClient.Consume();
+    public async Task Consume(CancellationToken cancellationToken) => await QueueClient.Consume(cancellationToken);
+    public async Task<ICommandEnvelope> Consume() => await QueueClient.Consume();
     public void Recover() => QueueClient.Recover();
     public bool ClearPendingQueue() => QueueClient.ClearPendingQueue();
     public bool ClearInProgressQueue() => QueueClient.ClearInProgressQueue();

Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@`
`3`	`3`	`using System.IO;`
`4`	`4`	`using System.Linq;`
`5`	`5`	`using System.Threading;`
	`6`	`+using System.Threading.Tasks;`
`6`	`7`	`using InEngine.Core.Exceptions;`
`7`	`8`	`using InEngine.Core.IO;`
`8`	`9`	`using InEngine.Core.Queuing.Message;`
`@@ -55,15 +56,15 @@ private void PublishToQueue(CommandEnvelope commandEnvelope, string queuePath)`
`55`	`56`	`);`
`56`	`57`	`}`
`57`	`58`
`58`		`- public void Consume(CancellationToken cancellationToken)`
	`59`	`+ public async Task Consume(CancellationToken cancellationToken)`
`59`	`60`	`{`
`60`	`61`	`try`
`61`	`62`	`{`
`62`	`63`	`while (true)`
`63`	`64`	`{`
`64`	`65`	`try`
`65`	`66`	`{`
`66`		`- if (Consume() == null)`
	`67`	`+ if (await Consume() == null)`
`67`	`68`	`Thread.Sleep(5000);`
`68`	`69`	`}`
`69`	`70`	`catch (Exception exception)`
`@@ -84,7 +85,7 @@ public void Consume(CancellationToken cancellationToken)`
`84`	`85`	`}`
`85`	`86`	`}`
`86`	`87`
`87`		`- public ICommandEnvelope Consume()`
	`88`	`+ public async Task<ICommandEnvelope> Consume()`
`88`	`89`	`{`
`89`	`90`	`FileInfo fileInfo;`
`90`	`91`	`var inProgressFilePath = string.Empty;`
`@@ -116,7 +117,7 @@ public ICommandEnvelope Consume()`
`116`	`117`	`try`
`117`	`118`	`{`
`118`	`119`	`command.WriteSummaryToConsole();`
`119`		`- command.RunWithLifeCycleAsync().RunSynchronously();`
	`120`	`+ await command.RunWithLifeCycleAsync();`
`120`	`121`	`}`
`121`	`122`	`catch (Exception exception)`
`122`	`123`	`{`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`	`using System.Collections.Generic;`
`3`	`3`	`using System.Text;`
`4`	`4`	`using System.Threading;`
	`5`	`+using System.Threading.Tasks;`
`5`	`6`	`using InEngine.Core.Exceptions;`
`6`	`7`	`using InEngine.Core.IO;`
`7`	`8`	`using InEngine.Core.Queuing.Message;`
`@@ -94,13 +95,18 @@ public void Recover()`
`94`	`95`	`{`
`95`	`96`	`}`
`96`	`97`
`97`		`- public void Consume(CancellationToken cancellationToken)`
	`98`	`+ public async Task Consume(CancellationToken cancellationToken)`
`98`	`99`	`{`
`99`	`100`	`InitChannel();`
`100`	`101`	`var consumer = new EventingBasicConsumer(Channel);`
`101`		`- consumer.Received += (model, result) =>`
	`102`	`+ consumer.Received += async (model, result) =>`
`102`	`103`	`{`
`103`	`104`	`var eventingConsumer = (EventingBasicConsumer)model;`
	`105`	`+ if (eventingConsumer == null)`
	`106`	`+ {`
	`107`	`+ Log.LogWarning("EventingBasicConsumer is null while attempting to consume messages");`
	`108`	`+ return;`
	`109`	`+ }`
`104`	`110`
`105`	`111`	`var serializedMessage = Encoding.UTF8.GetString(result.Body);`
`106`	`112`	`var commandEnvelope = serializedMessage.DeserializeFromJson<CommandEnvelope>();`
`@@ -115,7 +121,7 @@ public void Consume(CancellationToken cancellationToken)`
`115`	`121`	`try`
`116`	`122`	`{`
`117`	`123`	`command.WriteSummaryToConsole();`
`118`		`- command.RunWithLifeCycleAsync().RunSynchronously();`
	`124`	`+ await command.RunWithLifeCycleAsync();`
`119`	`125`	`}`
`120`	`126`	`catch (Exception exception)`
`121`	`127`	`{`
`@@ -135,7 +141,7 @@ public void Consume(CancellationToken cancellationToken)`
`135`	`141`	`Channel.BasicConsume(queue: PendingQueueName, autoAck: false, consumer: consumer);`
`136`	`142`	`}`
`137`	`143`
`138`		`- public ICommandEnvelope Consume()`
	`144`	`+ public async Task<ICommandEnvelope> Consume()`
`139`	`145`	`{`
`140`	`146`	`InitChannel();`
`141`	`147`	`var result = Channel.BasicGet(PendingQueueName, false);`
`@@ -155,7 +161,7 @@ public ICommandEnvelope Consume()`
`155`	`161`	`try`
`156`	`162`	`{`
`157`	`163`	`command.WriteSummaryToConsole();`
`158`		`- command.RunWithLifeCycleAsync().RunSynchronously();`
	`164`	`+ await command.RunWithLifeCycleAsync();`
`159`	`165`	`}`
`160`	`166`	`catch (Exception exception)`
`161`	`167`	`{`
Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ public void Recover()`
`82`	`82`	`PublishToChannel();`
`83`	`83`	`}`
`84`	`84`
`85`		`- public void Consume(CancellationToken cancellationToken)`
	`85`	`+ public async Task Consume(CancellationToken cancellationToken)`
`86`	`86`	`{`
`87`	`87`	`try`
`88`	`88`	`{`
`@@ -104,7 +104,7 @@ public void Consume(CancellationToken cancellationToken)`
`104`	`104`	`}`
`105`	`105`	`}`
`106`	`106`
`107`		`- public ICommandEnvelope Consume()`
	`107`	`+ public async Task<ICommandEnvelope> Consume()`
`108`	`108`	`{`
`109`	`109`	`var rawRedisMessageValue = Redis.ListRightPopLeftPush(PendingQueueName, InProgressQueueName);`
`110`	`110`	`var serializedMessage = rawRedisMessageValue.ToString();`
`@@ -122,7 +122,7 @@ public ICommandEnvelope Consume()`
`122`	`122`	`try`
`123`	`123`	`{`
`124`	`124`	`command.WriteSummaryToConsole();`
`125`		`- command.RunWithLifeCycleAsync().RunSynchronously();`
	`125`	`+ await command.RunWithLifeCycleAsync();`
`126`	`126`	`}`
`127`	`127`	`catch (Exception exception)`
`128`	`128`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Generic;`
`3`	`3`	`using System.Threading;`
	`4`	`+using System.Threading.Tasks;`
`4`	`5`	`using InEngine.Core.IO;`
`5`	`6`	`using InEngine.Core.Queuing.Message;`
`6`	`7`
`@@ -43,12 +44,12 @@ public void Recover()`
`43`	`44`	`{`
`44`	`45`	`}`
`45`	`46`
`46`		`- public void Consume(CancellationToken cancellationToken)`
	`47`	`+ public async Task Consume(CancellationToken cancellationToken)`
`47`	`48`	`{`
`48`	`49`	`throw new NotImplementedException();`
`49`	`50`	`}`
`50`	`51`
`51`		`- public ICommandEnvelope Consume()`
	`52`	`+ public async Task<ICommandEnvelope> Consume()`
`52`	`53`	`{`
`53`	`54`	`throw new NotImplementedException();`
`54`	`55`	`}`