MessageBus: ReceiveAsync

phongnguyend · May 28, 2023 · 2c6cf56 · 2c6cf56
1 parent e4a3588
commit 2c6cf56
Show file tree

Hide file tree

Showing 11 changed files with 67 additions and 116 deletions.
diff --git a/src/ModularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageReceiver.cs b/src/ModularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IMessageReceiver.cs
@@ -1,8 +1,10 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace ClassifiedAds.Domain.Infrastructure.MessageBrokers;
 
 public interface IMessageReceiver<T>
 {
-    void Receive(Action<T, MetaData> action);
+    Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken);
 }
diff --git a/...nolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureEventGrid/AzureEventGridSender.cs b/...nolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureEventGrid/AzureEventGridSender.cs
@@ -24,7 +24,6 @@ public AzureEventGridSender(string domainEndpoint, string domainKey, string topi
 
     public async Task SendAsync(T message, MetaData metaData, CancellationToken cancellationToken = default)
     {
-
         EventGridPublisherClient client = new EventGridPublisherClient(new Uri(_domainEndpoint), new AzureKeyCredential(_domainKey));
 
         var data = new BinaryData(JsonSerializer.Serialize(new Message<T>

diff --git a/...nolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureEventHub/AzureEventHubReceiver.cs b/...nolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureEventHub/AzureEventHubReceiver.cs
@@ -6,6 +6,7 @@
 using System;
 using System.Text;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ClassifiedAds.Infrastructure.MessageBrokers.AzureEventHub;
@@ -29,29 +30,22 @@ public void Dispose()
     {
     }
 
-    public void Receive(Action<T, MetaData> action)
-    {
-        ReceiveAsync(action).GetAwaiter().GetResult();
-    }
-
-    public async Task ReceiveAsync(Action<T, MetaData> action)
+    public async Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
         var storageClient = new BlobContainerClient(_storageConnectionString, _storageContainerName);
 
-        Task ProcessEventHandler(ProcessEventArgs eventArgs)
+        async Task ProcessEventHandler(ProcessEventArgs eventArgs)
         {
             try
             {
                 var messageAsString = Encoding.UTF8.GetString(eventArgs.Data.EventBody);
                 var message = JsonSerializer.Deserialize<Message<T>>(messageAsString);
-                action(message.Data, message.MetaData);
+                await action(message.Data, message.MetaData);
             }
             catch (Exception ex)
             {
                 Console.WriteLine(ex);
             }
-
-            return Task.CompletedTask;
         }
 
         Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
@@ -76,6 +70,6 @@ Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
 
         processor.ProcessEventAsync += ProcessEventHandler;
         processor.ProcessErrorAsync += ProcessErrorHandler;
-        await processor.StartProcessingAsync();
+        await processor.StartProcessingAsync(cancellationToken);
     }
 }
diff --git a/...ularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureQueue/AzureQueueReceiver.cs b/...ularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/AzureQueue/AzureQueueReceiver.cs
@@ -2,6 +2,7 @@
 using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
 using System;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ClassifiedAds.Infrastructure.MessageBrokers.AzureQueue;
@@ -19,57 +20,47 @@ public AzureQueueReceiver(string connectionString, string queueName, QueueMessag
         _messageEncoding = messageEncoding;
     }
 
-    public void Receive(Action<T, MetaData> action)
+    public async Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
-        Task.Factory.StartNew(() => ReceiveAsync(action));
-    }
-
-    private Task ReceiveAsync(Action<T, MetaData> action)
-    {
-        return ReceiveStringAsync(retrievedMessage =>
+        await ReceiveStringAsync(async retrievedMessage =>
         {
             var message = JsonSerializer.Deserialize<Message<T>>(retrievedMessage);
-            action(message.Data, message.MetaData);
-        });
-    }
-
-    public void ReceiveString(Action<string> action)
-    {
-        Task.Factory.StartNew(() => ReceiveStringAsync(action));
+            await action(message.Data, message.MetaData);
+        }, cancellationToken);
     }
 
-    private async Task ReceiveStringAsync(Action<string> action)
+    private async Task ReceiveStringAsync(Func<string, Task> action, CancellationToken cancellationToken)
     {
         var queueClient = new QueueClient(_connectionString, _queueName, new QueueClientOptions
         {
             MessageEncoding = _messageEncoding,
         });
 
-        await queueClient.CreateIfNotExistsAsync();
+        await queueClient.CreateIfNotExistsAsync(cancellationToken: cancellationToken);
 
-        while (true)
+        while (!cancellationToken.IsCancellationRequested)
         {
             try
             {
-                var retrievedMessages = (await queueClient.ReceiveMessagesAsync()).Value;
+                var retrievedMessages = (await queueClient.ReceiveMessagesAsync(cancellationToken)).Value;
 
                 if (retrievedMessages.Length > 0)
                 {
                     foreach (var retrievedMessage in retrievedMessages)
                     {
-                        action(retrievedMessage.Body.ToString());
-                        await queueClient.DeleteMessageAsync(retrievedMessage.MessageId, retrievedMessage.PopReceipt);
+                        await action(retrievedMessage.Body.ToString());
+                        await queueClient.DeleteMessageAsync(retrievedMessage.MessageId, retrievedMessage.PopReceipt, cancellationToken);
                     }
                 }
                 else
                 {
-                    await Task.Delay(1000);
+                    await Task.Delay(1000, cancellationToken);
                 }
             }
             catch (Exception ex)
             {
                 Console.WriteLine(ex);
-                await Task.Delay(1000);
+                await Task.Delay(1000, cancellationToken);
             }
         }
     }

diff --git a/...th/ClassifiedAds.Infrastructure/MessageBrokers/AzureServiceBus/AzureServiceBusReceiver.cs b/...th/ClassifiedAds.Infrastructure/MessageBrokers/AzureServiceBus/AzureServiceBusReceiver.cs
@@ -3,6 +3,7 @@
 using System;
 using System.Text;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ClassifiedAds.Infrastructure.MessageBrokers.AzureServiceBus;
@@ -18,42 +19,32 @@ public AzureServiceBusReceiver(string connectionString, string queueName)
         _queueName = queueName;
     }
 
-    public void Receive(Action<T, MetaData> action)
+    public async Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
-        Task.Factory.StartNew(() => ReceiveAsync(action));
-    }
-
-    private Task ReceiveAsync(Action<T, MetaData> action)
-    {
-        return ReceiveStringAsync(retrievedMessage =>
+        await ReceiveStringAsync(async retrievedMessage =>
         {
             var message = JsonSerializer.Deserialize<Message<T>>(retrievedMessage);
-            action(message.Data, message.MetaData);
-        });
-    }
-
-    public void ReceiveString(Action<string> action)
-    {
-        Task.Factory.StartNew(() => ReceiveStringAsync(action));
+            await action(message.Data, message.MetaData);
+        }, cancellationToken);
     }
 
-    private async Task ReceiveStringAsync(Action<string> action)
+    private async Task ReceiveStringAsync(Func<string, Task> action, CancellationToken cancellationToken)
     {
         await using var client = new ServiceBusClient(_connectionString);
         ServiceBusReceiver receiver = client.CreateReceiver(_queueName);
 
-        while (true)
+        while (!cancellationToken.IsCancellationRequested)
         {
-            var retrievedMessage = await receiver.ReceiveMessageAsync();
+            var retrievedMessage = await receiver.ReceiveMessageAsync(cancellationToken: cancellationToken);
 
             if (retrievedMessage != null)
             {
-                action(Encoding.UTF8.GetString(retrievedMessage.Body));
-                await receiver.CompleteMessageAsync(retrievedMessage);
+                await action(Encoding.UTF8.GetString(retrievedMessage.Body));
+                await receiver.CompleteMessageAsync(retrievedMessage, cancellationToken);
             }
             else
             {
-                await Task.Delay(1000);
+                await Task.Delay(1000, cancellationToken);
             }
         }
     }

diff --git a/...dAds.Infrastructure/MessageBrokers/AzureServiceBus/AzureServiceBusSubscriptionReceiver.cs b/...dAds.Infrastructure/MessageBrokers/AzureServiceBus/AzureServiceBusSubscriptionReceiver.cs
@@ -3,6 +3,7 @@
 using System;
 using System.Text;
 using System.Text.Json;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ClassifiedAds.Infrastructure.MessageBrokers.AzureServiceBus;
@@ -20,42 +21,32 @@ public AzureServiceBusSubscriptionReceiver(string connectionString, string topic
         _subscriptionName = subscriptionName;
     }
 
-    public void Receive(Action<T, MetaData> action)
+    public async Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
-        Task.Factory.StartNew(() => ReceiveAsync(action));
-    }
-
-    private Task ReceiveAsync(Action<T, MetaData> action)
-    {
-        return ReceiveStringAsync(retrievedMessage =>
+        await ReceiveStringAsync(async retrievedMessage =>
         {
             var message = JsonSerializer.Deserialize<Message<T>>(retrievedMessage);
-            action(message.Data, message.MetaData);
-        });
-    }
-
-    public void ReceiveString(Action<string> action)
-    {
-        Task.Factory.StartNew(() => ReceiveStringAsync(action));
+            await action(message.Data, message.MetaData);
+        }, cancellationToken);
     }
 
-    private async Task ReceiveStringAsync(Action<string> action)
+    private async Task ReceiveStringAsync(Func<string, Task> action, CancellationToken cancellationToken)
     {
         await using var client = new ServiceBusClient(_connectionString);
         ServiceBusReceiver receiver = client.CreateReceiver(_topicName, _subscriptionName);
 
-        while (true)
+        while (!cancellationToken.IsCancellationRequested)
         {
-            var retrievedMessage = await receiver.ReceiveMessageAsync();
+            var retrievedMessage = await receiver.ReceiveMessageAsync(cancellationToken: cancellationToken);
 
             if (retrievedMessage != null)
             {
-                action(Encoding.UTF8.GetString(retrievedMessage.Body));
-                await receiver.CompleteMessageAsync(retrievedMessage);
+                await action(Encoding.UTF8.GetString(retrievedMessage.Body));
+                await receiver.CompleteMessageAsync(retrievedMessage, cancellationToken);
             }
             else
             {
-                await Task.Delay(1000);
+                await Task.Delay(1000, cancellationToken);
             }
         }
     }

diff --git a/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/Fake/FakeReceiver.cs b/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/Fake/FakeReceiver.cs
@@ -1,12 +1,15 @@
 using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace ClassifiedAds.Infrastructure.MessageBrokers.Fake;
 
 public class FakeReceiver<T> : IMessageReceiver<T>
 {
-    public void Receive(Action<T, MetaData> action)
+    public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
         // do nothing
+        return Task.CompletedTask;
     }
 }
diff --git a/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/Kafka/KafkaReceiver.cs b/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/Kafka/KafkaReceiver.cs
@@ -29,28 +29,9 @@ public void Dispose()
         _consumer.Dispose();
     }
 
-    public void Receive(Action<T, MetaData> action)
+    public async Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cancellationToken)
     {
-        CancellationTokenSource cts = new CancellationTokenSource();
-        var cancellationToken = cts.Token;
-
-        Task.Factory.StartNew(() =>
-        {
-            try
-            {
-                StartReceiving(action, cancellationToken);
-            }
-            catch (OperationCanceledException)
-            {
-                Console.WriteLine("Closing consumer.");
-                _consumer.Close();
-            }
-        });
-    }
-
-    private void StartReceiving(Action<T, MetaData> action, CancellationToken cancellationToken)
-    {
-        while (true)
+        while (!cancellationToken.IsCancellationRequested)
         {
             try
             {
@@ -62,7 +43,7 @@ private void StartReceiving(Action<T, MetaData> action, CancellationToken cancel
                 }
 
                 var message = JsonSerializer.Deserialize<Message<T>>(consumeResult.Message.Value);
-                action(message.Data, message.MetaData);
+                await action(message.Data, message.MetaData);
             }
             catch (ConsumeException e)
             {

diff --git a/...onolith/ClassifiedAds.Infrastructure/MessageBrokers/MessageBrokersCollectionExtensions.cs b/...onolith/ClassifiedAds.Infrastructure/MessageBrokers/MessageBrokersCollectionExtensions.cs
@@ -180,7 +180,6 @@ public static IServiceCollection AddMessageBusSender<T>(this IServiceCollection
                     name: $"Message Broker (Azure Queue) {typeof(T).Name}",
                     failureStatus: HealthStatus.Degraded);
             }
-
         }
         else if (options.UsedAzureServiceBus())
         {
@@ -194,7 +193,6 @@ public static IServiceCollection AddMessageBusSender<T>(this IServiceCollection
                     name: $"Message Broker (Azure Service Bus) {typeof(T).Name}",
                     failureStatus: HealthStatus.Degraded);
             }
-
         }
         else if (options.UsedAzureEventGrid())
         {