add tests

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.AMQP.Client/Impl/AmqpMessage.cs

@@ -130,7 +130,7 @@ public IMessage Annotation(string key, object value)
         public object Annotation(string key)
         {
             ThrowIfAnnotationsNotSet();
-            return NativeMessage.MessageAnnotations[key];
+            return NativeMessage.MessageAnnotations[new Symbol(key)];
         }
     }
 }

RabbitMQ.AMQP.Client/Impl/DeliveryContext.cs

@@ -58,27 +58,29 @@ public Task DiscardAsync()
 
         public Task DiscardAsync(Dictionary<string, object> annotations)
         {
-
             if (_link.IsClosed)
             {
                 throw new ConsumerException("Link is closed");
             }
 
+            Utils.ValidateMessageAnnotations(annotations);
+
             Task rejectTask = Task.Run(() =>
             {
                 Fields messageAnnotations = new();
                 foreach (var kvp in annotations)
                 {
                     messageAnnotations.Add(new Symbol(kvp.Key), kvp.Value);
                 }
+
                 _link.Modify(_message, true, true, messageAnnotations);
                 _unsettledMessageCounter.Decrement();
                 _message.Dispose();
             });
 
             return rejectTask;
-
         }
+
         public Task RequeueAsync()
         {
             if (_link.IsClosed)
@@ -98,26 +100,25 @@ public Task RequeueAsync()
 
         public Task RequeueAsync(Dictionary<string, object> annotations)
         {
-
             if (_link.IsClosed)
             {
                 throw new ConsumerException("Link is closed");
             }
-
+            Utils.ValidateMessageAnnotations(annotations);
             Task requeueTask = Task.Run(() =>
             {
                 Fields messageAnnotations = new();
                 foreach (var kvp in annotations)
                 {
                     messageAnnotations.Add(new Symbol(kvp.Key), kvp.Value);
                 }
+
                 _link.Modify(_message, false, false, messageAnnotations);
                 _unsettledMessageCounter.Decrement();
                 _message.Dispose();
             });
 
             return requeueTask;
-
         }
     }
 }

RabbitMQ.AMQP.Client/Utils.cs

@@ -183,7 +183,7 @@ internal static bool CompareMap(Map map1, Map map2)
             return true;
         }
 
-        internal static void CheckMessageAnnotations(Dictionary<string, object> annotations)
+        internal static void ValidateMessageAnnotations(Dictionary<string, object> annotations)
         {
             foreach (var kvp in annotations.Where(kvp => !kvp.Key.StartsWith("x-")))
             {

Tests/Consumer/ConsumerOutcomeTests.cs

@@ -0,0 +1,86 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using RabbitMQ.AMQP.Client;
+using RabbitMQ.AMQP.Client.Impl;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Tests.Consumer;
+
+public class ConsumerOutcomeTests(ITestOutputHelper testOutputHelper) : IntegrationTest(testOutputHelper)
+{
+    [Fact]
+    public void ValidateAnnotations()
+    {
+        const string wrongAnnotationKey = "missing-the-start-x-annotation-key";
+        const string annotationValue = "annotation-value";
+        // This should throw an exception because the annotation key does not start with "x-"
+        Assert.Throws<ArgumentException>(() =>
+            Utils.ValidateMessageAnnotations(new Dictionary<string, object>
+            {
+                { wrongAnnotationKey, annotationValue }
+            }));
+        
+        const string correctAnnotationKey = "x-otp-annotation-key";
+        // This should not throw an exception because the annotation key starts with "x-"
+        Utils.ValidateMessageAnnotations(new Dictionary<string, object>
+        {
+            { correctAnnotationKey, annotationValue }
+        });
+    }
+
+    [Fact]
+    public async Task DiscardedMessageWithAnnotationsShouldBeDeadLeadLetteredAndContainAnnotationsWhenConfigured()
+    {
+        string dlqQueueName = $"dlq_{_queueName}";
+        await DeclareDeadLetterTopology(_queueName, dlqQueueName);
+
+
+        Assert.NotNull(_connection);
+        Assert.NotNull(_management);
+
+        const string annotationKey = "x-opt-annotation-key";
+        const string annotationValue = "annotation-value";
+        TaskCompletionSource<bool> tcs =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        IPublisher publisher = await _connection.PublisherBuilder().Queue(_queueName).BuildAsync();
+        IConsumer consumer = await _connection.ConsumerBuilder().MessageHandler(
+            async (context, _) =>
+            {
+                await context.DiscardAsync(new Dictionary<string, object> { { annotationKey, annotationValue } });
+                tcs.SetResult(true);
+            }
+        ).Queue(_queueName).BuildAndStartAsync();
+
+        IMessage message = new AmqpMessage($"message");
+        PublishResult pr = await publisher.PublishAsync(message);
+        Assert.Equal(OutcomeState.Accepted, pr.Outcome.State);
+        await tcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
+        await consumer.CloseAsync();
+        TaskCompletionSource<IMessage> tcsDl =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        IConsumer dlConsumer = await _connection.ConsumerBuilder().MessageHandler(async (context, message1) =>
+        {
+            await context.AcceptAsync();
+            tcsDl.SetResult(message1);
+        }).Queue(dlqQueueName).BuildAndStartAsync();
+
+        IMessage mResult = await tcsDl.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        Assert.NotNull(mResult);
+        Assert.Equal(mResult.Annotation(annotationKey), annotationValue);
+        await dlConsumer.CloseAsync();
+    }
+
+
+    private async Task DeclareDeadLetterTopology(string queueName, string dlxQueueName)
+    {
+        string dlx = $"{queueName}.dlx";
+        Assert.NotNull(_management);
+        await _management.Queue().Name(queueName).Type(QueueType.QUORUM).DeadLetterExchange(dlx).DeclareAsync();
+        await _management.Exchange(dlx).Type(ExchangeType.FANOUT).AutoDelete(true).DeclareAsync();
+        await _management.Queue(dlxQueueName).Exclusive(true).DeclareAsync();
+        await _management.Binding().SourceExchange(dlx).DestinationQueue(dlxQueueName).BindAsync();
+    }
+}