Merge pull request #7 from antoniaelek/feature/dlx

Added dlx parameter.
datalust · Jan 31, 2025 · 6619f9f · 6619f9f
2 parents 7f6b505 + ca6425b
commit 6619f9f
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 10 deletions.
diff --git a/src/Seq.Input.RabbitMQ/RabbitMQInput.cs b/src/Seq.Input.RabbitMQ/RabbitMQInput.cs
@@ -80,6 +80,12 @@ public class RabbitMQInput : SeqApp, IPublishJson, IDisposable
             HelpText = "Whether or not messages should be auto-acknowledged. The default is true.")]
         public bool IsReceiveAutoAck { get; set; } = true;
 
+        [SeqAppSetting(
+            DisplayName = "Dead Letter Exchange",
+            IsOptional = true,
+            HelpText = "The name of the dead letter exchange associated with this queue. If specified, the exchange will be used when declaring the queue, otherwise no dead lettering will be configured.")]
+        public string Dlx { get; set; }
+
         public void Start(TextWriter inputWriter)
         {
             var sync = new object();
@@ -111,7 +117,8 @@ void Receive(ReadOnlyMemory<byte> body)
                 IsQueueDurable,
                 IsQueueAutoDelete,
                 IsQueueExclusive,
-                IsReceiveAutoAck);
+                IsReceiveAutoAck,
+                Dlx);
         }
 
         public void Stop()

diff --git a/src/Seq.Input.RabbitMQ/RabbitMQListener.cs b/src/Seq.Input.RabbitMQ/RabbitMQListener.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Net.Security;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
@@ -10,19 +11,19 @@ class RabbitMQListener : IDisposable
         readonly IConnection _connection;
         readonly IModel _channel;
 
-        public RabbitMQListener(
-            Action<ReadOnlyMemory<byte>> receive,
+        public RabbitMQListener(Action<ReadOnlyMemory<byte>> receive,
             string rabbitMQHost,
             string rabbitMQVHost,
-            int rabbitMQPort, 
-            string rabbitMQUser, 
+            int rabbitMQPort,
+            string rabbitMQUser,
             string rabbitMQPassword,
-            string rabbitMQQueue, 
+            string rabbitMQQueue,
             bool isSsl,
-            bool isQueueDurable, 
-            bool isQueueAutoDelete, 
+            bool isQueueDurable,
+            bool isQueueAutoDelete,
             bool isQueueExclusive,
-            bool isReceiveAutoAck)
+            bool isReceiveAutoAck, 
+            string dlx)
         {
             var factory = new ConnectionFactory
             {
@@ -40,12 +41,16 @@ public RabbitMQListener(
             _connection = factory.CreateConnection();
             _channel = _connection.CreateModel();
 
+            var arguments = string.IsNullOrWhiteSpace(dlx) 
+                ? null 
+                : new Dictionary<string, object> { {"x-dead-letter-exchange", dlx} };
+
             _channel.QueueDeclare(
                 rabbitMQQueue, 
                 durable: isQueueDurable, 
                 exclusive: isQueueExclusive,
                 autoDelete: isQueueAutoDelete, 
-                arguments: null);
+                arguments: arguments);
 
             var consumer = new EventingBasicConsumer(_channel);
             consumer.Received += (model, ea) => receive(ea.Body);