RabbitMQ: Configure Dead Letter Queue

kritulrathod · Aug 5, 2024 · 2376a7e · 2376a7e
1 parent 6465b65
commit 2376a7e
Show file tree

Hide file tree

Showing 9 changed files with 114 additions and 6 deletions.
diff --git a/.github/workflows/docker-build-microservices.yml b/.github/workflows/docker-build-microservices.yml
@@ -16,10 +16,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: docker-compose build
+      - name: docker compose build
         run: |
           cd src/Microservices
-          docker-compose build
+          docker compose build
       - name: Log into Registry
         run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login docker.pkg.github.com -u ${{ github.actor }} --password-stdin
       - name: Tag and Push Images

diff --git a/.github/workflows/docker-build-modularmonolith.yml b/.github/workflows/docker-build-modularmonolith.yml
@@ -16,10 +16,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: docker-compose build
+      - name: docker compose build
         run: |
           cd src/ModularMonolith
-          docker-compose build
+          docker compose build
       - name: Log into Registry
         run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login docker.pkg.github.com -u ${{ github.actor }} --password-stdin
       - name: Tag and Push Images

diff --git a/.github/workflows/docker-build-monolith.yml b/.github/workflows/docker-build-monolith.yml
@@ -16,10 +16,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: docker-compose build
+      - name: docker compose build
         run: |
           cd src/Monolith
-          docker-compose build
+          docker compose build
       - name: Log into Registry
         run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login docker.pkg.github.com -u ${{ github.actor }} --password-stdin
       - name: Tag and Push Images

diff --git a/...oservices/Common/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs b/...oservices/Common/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs
@@ -65,6 +65,25 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
                 arguments["x-single-active-consumer"] = true;
             }
 
+            if (_options.DeadLetter != null)
+            {
+                if (!string.IsNullOrEmpty(_options.DeadLetter.ExchangeName))
+                {
+                    arguments["x-dead-letter-exchange"] = _options.DeadLetter.ExchangeName;
+                }
+
+                if (!string.IsNullOrEmpty(_options.DeadLetter.RoutingKey))
+                {
+                    arguments["x-dead-letter-routing-key"] = _options.DeadLetter.RoutingKey;
+                }
+
+                if (_options.DeadLetter.AutomaticCreateEnabled && !string.IsNullOrEmpty(_options.DeadLetter.QueueName))
+                {
+                    _channel.QueueDeclare(_options.DeadLetter.QueueName, true, false, false, null);
+                    _channel.QueueBind(_options.DeadLetter.QueueName, _options.DeadLetter.ExchangeName, _options.DeadLetter.RoutingKey, null);
+                }
+            }
+
             arguments = arguments.Count == 0 ? null : arguments;
 
             _channel.QueueDeclare(_options.QueueName, true, false, false, arguments);
@@ -108,6 +127,8 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
             catch (Exception ex)
             {
                 // TODO: log here
+                await Task.Delay(1000);
+                _channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: _options.RequeueOnFailure);
             }
         };
 

diff --git a/...es/Common/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs b/...es/Common/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs
@@ -23,4 +23,19 @@ public class RabbitMQReceiverOptions
     public bool MessageEncryptionEnabled { get; set; }
 
     public string MessageEncryptionKey { get; set; }
+
+    public bool RequeueOnFailure { get; set; }
+
+    public DeadLetterOptions DeadLetter { get; set; }
+}
+
+public class DeadLetterOptions
+{
+    public string ExchangeName { get; set; }
+
+    public string RoutingKey { get; set; }
+
+    public string QueueName { get; set; }
+
+    public bool AutomaticCreateEnabled { get; set; }
 }
diff --git a/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs b/src/ModularMonolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs
@@ -65,6 +65,25 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
                 arguments["x-single-active-consumer"] = true;
             }
 
+            if (_options.DeadLetter != null)
+            {
+                if (!string.IsNullOrEmpty(_options.DeadLetter.ExchangeName))
+                {
+                    arguments["x-dead-letter-exchange"] = _options.DeadLetter.ExchangeName;
+                }
+
+                if (!string.IsNullOrEmpty(_options.DeadLetter.RoutingKey))
+                {
+                    arguments["x-dead-letter-routing-key"] = _options.DeadLetter.RoutingKey;
+                }
+
+                if (_options.DeadLetter.AutomaticCreateEnabled && !string.IsNullOrEmpty(_options.DeadLetter.QueueName))
+                {
+                    _channel.QueueDeclare(_options.DeadLetter.QueueName, true, false, false, null);
+                    _channel.QueueBind(_options.DeadLetter.QueueName, _options.DeadLetter.ExchangeName, _options.DeadLetter.RoutingKey, null);
+                }
+            }
+
             arguments = arguments.Count == 0 ? null : arguments;
 
             _channel.QueueDeclare(_options.QueueName, true, false, false, arguments);
@@ -108,6 +127,8 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
             catch (Exception ex)
             {
                 // TODO: log here
+                await Task.Delay(1000);
+                _channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: _options.RequeueOnFailure);
             }
         };
 

diff --git a/...rMonolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs b/...rMonolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs
@@ -23,4 +23,19 @@ public class RabbitMQReceiverOptions
     public bool MessageEncryptionEnabled { get; set; }
 
     public string MessageEncryptionKey { get; set; }
+
+    public bool RequeueOnFailure { get; set; }
+
+    public DeadLetterOptions DeadLetter { get; set; }
+}
+
+public class DeadLetterOptions
+{
+    public string ExchangeName { get; set; }
+
+    public string RoutingKey { get; set; }
+
+    public string QueueName { get; set; }
+
+    public bool AutomaticCreateEnabled { get; set; }
 }
diff --git a/src/Monolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs b/src/Monolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiver.cs
@@ -65,6 +65,25 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
                 arguments["x-single-active-consumer"] = true;
             }
 
+            if (_options.DeadLetter != null)
+            {
+                if (!string.IsNullOrEmpty(_options.DeadLetter.ExchangeName))
+                {
+                    arguments["x-dead-letter-exchange"] = _options.DeadLetter.ExchangeName;
+                }
+
+                if (!string.IsNullOrEmpty(_options.DeadLetter.RoutingKey))
+                {
+                    arguments["x-dead-letter-routing-key"] = _options.DeadLetter.RoutingKey;
+                }
+
+                if (_options.DeadLetter.AutomaticCreateEnabled && !string.IsNullOrEmpty(_options.DeadLetter.QueueName))
+                {
+                    _channel.QueueDeclare(_options.DeadLetter.QueueName, true, false, false, null);
+                    _channel.QueueBind(_options.DeadLetter.QueueName, _options.DeadLetter.ExchangeName, _options.DeadLetter.RoutingKey, null);
+                }
+            }
+
             arguments = arguments.Count == 0 ? null : arguments;
 
             _channel.QueueDeclare(_options.QueueName, true, false, false, arguments);
@@ -108,6 +127,8 @@ public Task ReceiveAsync(Func<T, MetaData, Task> action, CancellationToken cance
             catch (Exception ex)
             {
                 // TODO: log here
+                await Task.Delay(1000);
+                _channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: _options.RequeueOnFailure);
             }
         };
 

diff --git a/src/Monolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs b/src/Monolith/ClassifiedAds.Infrastructure/MessageBrokers/RabbitMQ/RabbitMQReceiverOptions.cs
@@ -23,4 +23,19 @@ public class RabbitMQReceiverOptions
     public bool MessageEncryptionEnabled { get; set; }
 
     public string MessageEncryptionKey { get; set; }
+
+    public bool RequeueOnFailure { get; set; }
+
+    public DeadLetterOptions DeadLetter { get; set; }
+}
+
+public class DeadLetterOptions
+{
+    public string ExchangeName { get; set; }
+
+    public string RoutingKey { get; set; }
+
+    public string QueueName { get; set; }
+
+    public bool AutomaticCreateEnabled { get; set; }
 }