Annotations

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

Author: Gsantomaggio

RabbitMQ.AMQP.Client/IConsumer.cs

@@ -3,6 +3,7 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace RabbitMQ.AMQP.Client
@@ -27,6 +28,9 @@ public interface IContext
     {
         Task AcceptAsync();
         Task DiscardAsync();
+        Task DiscardAsync(Dictionary<string, object> annotations);
+
         Task RequeueAsync();
+        Task RequeueAsync(Dictionary<string, object> annotations);
     }
 }

RabbitMQ.AMQP.Client/Impl/DeliveryContext.cs

@@ -2,8 +2,10 @@
 // 2.0, and the Mozilla Public License, version 2.0.
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using Amqp;
+using Amqp.Types;
 
 namespace RabbitMQ.AMQP.Client.Impl
 {
@@ -54,6 +56,29 @@ public Task DiscardAsync()
             return rejectTask;
         }
 
+        public Task DiscardAsync(Dictionary<string, object> annotations)
+        {
+
+            if (_link.IsClosed)
+            {
+                throw new ConsumerException("Link is closed");
+            }
+
+            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)
@@ -70,5 +95,29 @@ public Task RequeueAsync()
 
             return requeueTask;
         }
+
+        public Task RequeueAsync(Dictionary<string, object> annotations)
+        {
+
+            if (_link.IsClosed)
+            {
+                throw new ConsumerException("Link is closed");
+            }
+
+            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/PublicAPI.Unshipped.txt

@@ -125,7 +125,9 @@ RabbitMQ.AMQP.Client.IConsumerBuilder.Stream() -> RabbitMQ.AMQP.Client.IConsumer
 RabbitMQ.AMQP.Client.IContext
 RabbitMQ.AMQP.Client.IContext.AcceptAsync() -> System.Threading.Tasks.Task!
 RabbitMQ.AMQP.Client.IContext.DiscardAsync() -> System.Threading.Tasks.Task!
+RabbitMQ.AMQP.Client.IContext.DiscardAsync(System.Collections.Generic.Dictionary<string!, object!>! annotations) -> System.Threading.Tasks.Task!
 RabbitMQ.AMQP.Client.IContext.RequeueAsync() -> System.Threading.Tasks.Task!
+RabbitMQ.AMQP.Client.IContext.RequeueAsync(System.Collections.Generic.Dictionary<string!, object!>! annotations) -> System.Threading.Tasks.Task!
 RabbitMQ.AMQP.Client.IEntityInfo
 RabbitMQ.AMQP.Client.IEntityInfoSpecification<T>
 RabbitMQ.AMQP.Client.IEntityInfoSpecification<T>.DeclareAsync() -> System.Threading.Tasks.Task<T>!

RabbitMQ.AMQP.Client/Utils.cs

@@ -3,6 +3,8 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Security.Cryptography;
 using System.Text;
 using System.Web;
@@ -177,8 +179,18 @@ internal static bool CompareMap(Map map1, Map map2)
                     return false;
                 }
             }
+
             return true;
         }
+
+        internal static void CheckMessageAnnotations(Dictionary<string, object> annotations)
+        {
+            foreach (var kvp in annotations.Where(kvp => !kvp.Key.StartsWith("x-")))
+            {
+                throw new ArgumentException(
+                    $"Message annotation keys must start with 'x-': {kvp.Key}");
+            }
+        }
     }
 
     // TODO why can't we use normal HTTP encoding?