Change consumer options to uniform all the AMQP 1.0 clients interfaces (#144)

This pull request refactors the consumer builder API to use a unified enum-based approach for configuring consumer features, replacing the previous boolean-based methods. The change consolidates DirectReplyTo(bool) and PreSettled(bool) methods into a single Feature(ConsumerFeature) method that accepts an enum value.

Changes:

* Introduced ConsumerSettleStrategy enum with three values: ExplicitSettle, DirectReplyTo, and PreSettled
* Replaced IConsumerBuilder.DirectReplyTo(bool) and IConsumerBuilder.PreSettled(bool) with IConsumerBuilder.SettleStrategy(ConsumerSettleStrategy)
* Updated all consumer creation code across tests, examples, and internal implementations to use the new API
Reviewed changes
---------

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

RabbitMQ.AMQP.Client/IConsumerBuilder.cs

@@ -15,32 +15,43 @@ public enum StreamOffsetSpecification
         Next
     }
 
+    /// <summary>
+    ///   ConsumerSettleStrategy defines the Consumer Settle available for Classic and Quorum queue consumers.
+    /// </summary>
+    public enum ConsumerSettleStrategy
+    {
+        // ExplicitSettle: means that the consumer will be created with the default settings.
+        // message settle mode will be the default one (explicit settle)
+        // via Context. See IContext for more details on message settlement.
+        // ExplicitSettle is the default behavior if no feature is specified.
+        ExplicitSettle,
+
+        // DirectReplyTo: Enables Direct Reply-To consumer behavior.
+        // Feature in RabbitMQ, allowing for simplified request-reply messaging patterns.
+        // Docs: https://www.rabbitmq.com/docs/direct-reply-to#usage-amqp
+        // with DirectReplyTo presettled is by default. 
+        DirectReplyTo,
+
+        // Presettled: Deliveries are pre-settled.
+        // When enabled, messages are automatically settled when received,
+        // meaning they cannot be redelivered if processing fails.
+        PreSettled,
+    }
+
     // TODO IAddressBuilder<IConsumerBuilder>?
     public interface IConsumerBuilder
     {
         IConsumerBuilder Queue(IQueueSpecification queueSpecification);
         IConsumerBuilder Queue(string? queueName);
 
-        /// <summary>
-        /// If direct reply-to is enabled, the client will use the direct reply-to feature of AMQP 1.0.
-        /// The server must also support direct reply-to.
-        /// This feature allows the server to send the reply directly to the client without going through a reply queue.
-        /// This can improve performance and reduce latency.
-        /// Default is false.
-        /// https://www.rabbitmq.com/docs/direct-reply-to
-        /// </summary>
-        IConsumerBuilder DirectReplyTo(bool directReplyTo);
-
         IConsumerBuilder MessageHandler(MessageHandler handler);
 
         IConsumerBuilder InitialCredits(int initialCredits);
-
+        
         /// <summary>
-        /// If pre-settled is enabled, the receiver will use ReceiverSettleMode.Second,
-        /// meaning messages are pre-settled and the receiver does not need to explicitly settle them.
-        /// Default is false.
+        /// SettleStrategy: ConsumerSettleStrategy for more details on the Consumer Settle that can be enabled for the consumer.
         /// </summary>
-        IConsumerBuilder PreSettled(bool preSettled);
+        IConsumerBuilder SettleStrategy(ConsumerSettleStrategy settleStrategy);
 
         /// <summary>
         /// SubscriptionListener interface callback to add behavior before a subscription is created.

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -75,15 +75,15 @@ public override async Task OpenAsync()
 
                 Attach attach;
 
-                if (_configuration.DirectReplyTo)
+                if (_configuration.SettleStrategy == ConsumerSettleStrategy.DirectReplyTo)
                 {
                     attach = Utils.CreateDirectReplyToAttach(_id, _configuration.Filters);
                 }
                 else
                 {
                     string address = AddressBuilderHelper.AddressBuilder().Queue(_configuration.Queue).Address();
                     attach = Utils.CreateAttach(address, DeliveryMode.AtLeastOnce, _id,
-                        _configuration.Filters, _configuration.PreSettled);
+                        _configuration.Filters, _configuration.SettleStrategy == ConsumerSettleStrategy.PreSettled);
                 }
 
                 void OnAttached(ILink argLink, Attach argAttach)
@@ -176,14 +176,14 @@ private async Task ProcessMessages()
                         continue;
                     }
 
-                    IContext context = _configuration.PreSettled switch
+                    IContext context = _configuration.SettleStrategy switch
                     {
-                        true => new PreSettledDeliveryContext(),
-                        false => new DeliveryContext(_receiverLink, nativeMessage, _unsettledMessageCounter,
+                        ConsumerSettleStrategy.PreSettled => new PreSettledDeliveryContext(),
+                        _ => new DeliveryContext(_receiverLink, nativeMessage, _unsettledMessageCounter,
                             _metricsReporter)
                     };
 
-                    if (!_configuration.PreSettled)
+                    if (_configuration.SettleStrategy != ConsumerSettleStrategy.PreSettled)
                     {
                         _unsettledMessageCounter.Increment();
                     }

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -16,7 +16,6 @@ namespace RabbitMQ.AMQP.Client.Impl
     /// </summary>
     internal sealed class ConsumerConfiguration
     {
-
         public string? Queue { get; set; } = null;
         public int InitialCredits { get; set; } = 100; // TODO use constant, check with Java lib
 
@@ -25,22 +24,7 @@ internal sealed class ConsumerConfiguration
         // TODO is a MessageHandler *really* optional???
         public MessageHandler? Handler { get; set; }
 
-        /// <summary>
-        /// If direct reply-to is enabled, the client will use the direct reply-to feature of AMQP 1.0.
-        /// The server must also support direct reply-to.
-        /// This feature allows the server to send the reply directly to the client without going through a reply queue.
-        /// This can improve performance and reduce latency.
-        /// Default is false.
-        /// https://www.rabbitmq.com/docs/direct-reply-to
-        /// </summary>
-        public bool DirectReplyTo { get; set; }
-
-        /// <summary>
-        /// If pre-settled is enabled, the receiver will use ReceiverSettleMode.Second,
-        /// meaning messages are pre-settled and the receiver does not need to explicitly settle them.
-        /// Default is false.
-        /// </summary>
-        public bool PreSettled { get; set; }
+        public ConsumerSettleStrategy SettleStrategy { get; set; } = ConsumerSettleStrategy.ExplicitSettle;
 
         // TODO re-name to ListenerContextAction? Callback?
         public Action<IConsumerBuilder.ListenerContext>? ListenerContext = null;
@@ -79,21 +63,15 @@ public IConsumerBuilder MessageHandler(MessageHandler handler)
             return this;
         }
 
-        public IConsumerBuilder DirectReplyTo(bool directReplyTo)
-        {
-            _configuration.DirectReplyTo = directReplyTo;
-            return this;
-        }
-
         public IConsumerBuilder InitialCredits(int initialCredits)
         {
             _configuration.InitialCredits = initialCredits;
             return this;
         }
 
-        public IConsumerBuilder PreSettled(bool preSettled)
+        public IConsumerBuilder SettleStrategy(ConsumerSettleStrategy settleStrategy)
         {
-            _configuration.PreSettled = preSettled;
+            _configuration.SettleStrategy = settleStrategy;
             return this;
         }
 
@@ -340,7 +318,6 @@ public IConsumerBuilder.IStreamFilterOptions PropertySymbol(string key, string v
 
         private StreamFilterOptions PropertyFilter(string propertyKey, object propertyValue)
         {
-
             DescribedValue propertiesFilterValue = Filter(Consts.AmqpPropertiesFilter);
             Map propertiesFilter = (Map)propertiesFilterValue.Value;
             // Note: you MUST use a symbol as the key
@@ -350,7 +327,6 @@ private StreamFilterOptions PropertyFilter(string propertyKey, object propertyVa
 
         private StreamFilterOptions ApplicationPropertyFilter(string propertyKey, object propertyValue)
         {
-
             DescribedValue applicationPropertiesFilterValue = Filter(Consts.AmqpApplicationPropertiesFilter);
             Map applicationPropertiesFilter = (Map)applicationPropertiesFilterValue.Value;
             // Note: do NOT put a symbol as the key

RabbitMQ.AMQP.Client/Impl/AmqpRequester.cs

@@ -177,11 +177,14 @@ public override async Task OpenAsync()
             // The user is always free to create custom Requester and Responder 
             bool canApplyDirectReplyTo = isDirectReplyToSupported &&
                                          string.IsNullOrEmpty(_configuration.ReplyToQueue);
+            ConsumerSettleStrategy consumerSettleStrategy = canApplyDirectReplyTo
+                ? ConsumerSettleStrategy.DirectReplyTo
+                : ConsumerSettleStrategy.ExplicitSettle;
 
             _publisher = await _configuration.Connection.PublisherBuilder().BuildAsync().ConfigureAwait(false);
             _consumer = await _configuration.Connection.ConsumerBuilder()
                 .Queue(queueReplyTo)
-                .DirectReplyTo(canApplyDirectReplyTo)
+                .SettleStrategy(consumerSettleStrategy)
                 .MessageHandler((context, message) =>
                 {
                     // TODO MessageHandler funcs should catch all exceptions

RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt

@@ -104,6 +104,10 @@ RabbitMQ.AMQP.Client.ConnectionSettingsBuilder.VirtualHost(string! virtualHost)
 RabbitMQ.AMQP.Client.Consts
 RabbitMQ.AMQP.Client.ConsumerException
 RabbitMQ.AMQP.Client.ConsumerException.ConsumerException(string! message) -> void
+RabbitMQ.AMQP.Client.ConsumerSettleStrategy
+RabbitMQ.AMQP.Client.ConsumerSettleStrategy.DirectReplyTo = 1 -> RabbitMQ.AMQP.Client.ConsumerSettleStrategy
+RabbitMQ.AMQP.Client.ConsumerSettleStrategy.ExplicitSettle = 0 -> RabbitMQ.AMQP.Client.ConsumerSettleStrategy
+RabbitMQ.AMQP.Client.ConsumerSettleStrategy.PreSettled = 2 -> RabbitMQ.AMQP.Client.ConsumerSettleStrategy
 RabbitMQ.AMQP.Client.Error
 RabbitMQ.AMQP.Client.Error.Description.get -> string?
 RabbitMQ.AMQP.Client.Error.Error(string? errorCode, string? description) -> void
@@ -176,7 +180,6 @@ RabbitMQ.AMQP.Client.IConsumer.Unpause() -> void
 RabbitMQ.AMQP.Client.IConsumer.UnsettledMessageCount.get -> long
 RabbitMQ.AMQP.Client.IConsumerBuilder
 RabbitMQ.AMQP.Client.IConsumerBuilder.BuildAndStartAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IConsumer!>!
-RabbitMQ.AMQP.Client.IConsumerBuilder.DirectReplyTo(bool directReplyTo) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IConsumerBuilder.InitialCredits(int initialCredits) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamFilterOptions
 RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamFilterOptions.AbsoluteExpiryTime(System.DateTime absoluteExpiryTime) -> RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamFilterOptions!
@@ -209,9 +212,9 @@ RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext
 RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext.ListenerContext(RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamOptions! streamOptions) -> void
 RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext.StreamOptions.get -> RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamOptions!
 RabbitMQ.AMQP.Client.IConsumerBuilder.MessageHandler(RabbitMQ.AMQP.Client.MessageHandler! handler) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
-RabbitMQ.AMQP.Client.IConsumerBuilder.PreSettled(bool preSettled) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IConsumerBuilder.Queue(RabbitMQ.AMQP.Client.IQueueSpecification! queueSpecification) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IConsumerBuilder.Queue(string? queueName) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
+RabbitMQ.AMQP.Client.IConsumerBuilder.SettleStrategy(RabbitMQ.AMQP.Client.ConsumerSettleStrategy settleStrategy) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IConsumerBuilder.Stream() -> RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamOptions!
 RabbitMQ.AMQP.Client.IConsumerBuilder.SubscriptionListener(System.Action<RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext!>! listenerContext) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.IContext
@@ -373,12 +376,11 @@ RabbitMQ.AMQP.Client.Impl.AmqpConsumer.UnsettledMessageCount.get -> long
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.AmqpConsumerBuilder(RabbitMQ.AMQP.Client.Impl.AmqpConnection! connection, RabbitMQ.AMQP.Client.IMetricsReporter? metricsReporter) -> void
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.BuildAndStartAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IConsumer!>!
-RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.DirectReplyTo(bool directReplyTo) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.InitialCredits(int initialCredits) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.MessageHandler(RabbitMQ.AMQP.Client.MessageHandler! handler) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
-RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.PreSettled(bool preSettled) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.Queue(RabbitMQ.AMQP.Client.IQueueSpecification! queueSpec) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.Queue(string? queueName) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
+RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.SettleStrategy(RabbitMQ.AMQP.Client.ConsumerSettleStrategy settleStrategy) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.Stream() -> RabbitMQ.AMQP.Client.IConsumerBuilder.IStreamOptions!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder.SubscriptionListener(System.Action<RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext!>! context) -> RabbitMQ.AMQP.Client.IConsumerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpEnvironment

Tests/Consumer/PreSettledConsumerTests.cs

@@ -27,7 +27,7 @@ public async Task PreSettledConsumerCanReceiveMessages()
         TaskCompletionSource<IMessage> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 // With PreSettled, messages are already settled, so we don't need to call Accept()
@@ -64,7 +64,7 @@ public async Task PreSettledConsumerAcceptThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -104,7 +104,7 @@ public async Task PreSettledConsumerDiscardThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -144,7 +144,7 @@ public async Task PreSettledConsumerDiscardWithAnnotationsThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -185,7 +185,7 @@ public async Task PreSettledConsumerRequeueThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -225,7 +225,7 @@ public async Task PreSettledConsumerRequeueWithAnnotationsThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -266,7 +266,7 @@ public async Task PreSettledConsumerBatchThrowsException()
         TaskCompletionSource<InvalidOperationException> tcs = new();
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 try
@@ -308,7 +308,7 @@ public async Task PreSettledConsumerUnsettledMessageCountIsZero()
         int receivedCount = 0;
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 receivedCount++;
@@ -347,7 +347,7 @@ public async Task NonPreSettledConsumerUnsettledMessageCountIncrements()
         int receivedCount = 0;
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(false) // Explicitly set to false
+            .SettleStrategy(ConsumerSettleStrategy.ExplicitSettle) // Explicitly avoid pre-settled feature
             .MessageHandler((context, message) =>
             {
                 receivedCount++;
@@ -387,7 +387,7 @@ public async Task PreSettledConsumerCanReceiveMultipleMessages()
         int receivedCount = 0;
         IConsumer consumer = await _connection.ConsumerBuilder()
             .Queue(queueSpec)
-            .PreSettled(true)
+            .SettleStrategy(ConsumerSettleStrategy.PreSettled)
             .MessageHandler((context, message) =>
             {
                 receivedCount++;

Tests/DirectReply/DirectReplyTests.cs

@@ -23,7 +23,7 @@ public async Task ValidateDirectReplyQName()
                 "DirectReply is not supported by the connection.");
 
             IConsumer consumer = await _connection.ConsumerBuilder()
-                .DirectReplyTo(true)
+                .SettleStrategy(ConsumerSettleStrategy.DirectReplyTo)
                 .MessageHandler((IContext _, IMessage _) => Task.CompletedTask)
                 .BuildAndStartAsync();
 
@@ -42,7 +42,7 @@ public async Task UseDirectReplyToReceiveAMessage()
                 "DirectReply is not supported by the connection.");
 
             IConsumer consumer = await _connection.ConsumerBuilder()
-                .DirectReplyTo(true)
+                .SettleStrategy(ConsumerSettleStrategy.DirectReplyTo)
                 .MessageHandler((IContext _, IMessage msg) =>
                 {
                     tcs.SetResult(msg);