reply to wip

fixes #133

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

Author: Gsantomaggio

RabbitMQ.AMQP.Client/FeatureFlags.cs

@@ -24,6 +24,13 @@ public class FeatureFlags
         /// </summary>
         public bool IsBrokerCompatible { get; internal set; } = false;
 
+        /// <summary>
+        /// Check if Direct Reply-To is supported.
+        /// Direct Reply-To is available in RabbitMQ 4.2 and later.
+        /// see: https://www.rabbitmq.com/docs/direct-reply-to
+        /// </summary>
+        public bool IsDirectReplyToSupported { get; internal set; } = false;
+
         public void Validate()
         {
             if (!IsBrokerCompatible)

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -39,7 +39,7 @@ public class AmqpConnection : AbstractLifeCycle, IConnection
         private readonly IMetricsReporter? _metricsReporter;
 
         private readonly Dictionary<string, object> _connectionProperties = new();
-        internal readonly FeatureFlags _featureFlags = new FeatureFlags();
+        internal readonly FeatureFlags _featureFlags = new();
 
         /// <summary>
         /// _publishersDict contains all the publishers created by the connection.
@@ -670,6 +670,8 @@ private void HandleProperties(Fields properties)
             // check if the broker supports filter expressions
             // this is a feature that was introduced in RabbitMQ 4.2.0
             _featureFlags.IsSqlFeatureEnabled = Utils.Is4_2_OrMore(brokerVersion);
+            
+            _featureFlags.IsDirectReplyToSupported = Utils.Is4_2_OrMore(brokerVersion);
 
             _featureFlags.IsFilterFeatureEnabled = Utils.SupportsFilterExpressions(brokerVersion);
         }

RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt

@@ -116,6 +116,7 @@ RabbitMQ.AMQP.Client.ExchangeType.TOPIC = 2 -> RabbitMQ.AMQP.Client.ExchangeType
 RabbitMQ.AMQP.Client.FeatureFlags
 RabbitMQ.AMQP.Client.FeatureFlags.FeatureFlags() -> void
 RabbitMQ.AMQP.Client.FeatureFlags.IsBrokerCompatible.get -> bool
+RabbitMQ.AMQP.Client.FeatureFlags.IsDirectReplyToSupported.get -> bool
 RabbitMQ.AMQP.Client.FeatureFlags.IsFilterFeatureEnabled.get -> bool
 RabbitMQ.AMQP.Client.FeatureFlags.IsSqlFeatureEnabled.get -> bool
 RabbitMQ.AMQP.Client.FeatureFlags.Validate() -> void

Tests/Rpc/ResponderTests.cs

@@ -39,15 +39,8 @@ public async Task MockResponderPingPong()
             Assert.NotNull(_connection);
             TaskCompletionSource<IMessage> tcs = CreateTaskCompletionSource<IMessage>();
 
-            Task<IMessage> RpcHandler(IResponder.IContext context, IMessage request)
-            {
-                IMessage reply = context.Message("pong");
-                tcs.SetResult(reply);
-                return Task.FromResult(reply);
-            }
-
             IResponder responder = await _connection.ResponderBuilder()
-                .Handler(RpcHandler)
+                .Handler(Handler)
                 .RequestQueue(_requestQueueName)
                 .BuildAsync();
 
@@ -59,6 +52,14 @@ Task<IMessage> RpcHandler(IResponder.IContext context, IMessage request)
             IMessage m = await WhenTcsCompletes(tcs);
             Assert.Equal("pong", m.BodyAsString());
             await responder.CloseAsync();
+            return;
+
+            Task<IMessage> Handler(IResponder.IContext context, IMessage request)
+            {
+                IMessage reply = context.Message("pong");
+                tcs.SetResult(reply);
+                return Task.FromResult(reply);
+            }
         }
 
         [Fact]
@@ -69,14 +70,8 @@ public async Task ResponderValidateStateChange()
             List<(State, State)> states = [];
             TaskCompletionSource<int> tcs = CreateTaskCompletionSource<int>();
 
-            static Task<IMessage> RpcHandler(IResponder.IContext context, IMessage request)
-            {
-                IMessage m = context.Message(request.Body());
-                return Task.FromResult(m);
-            }
-
             IResponder responder = await _connection.ResponderBuilder()
-                .Handler(RpcHandler)
+                .Handler(Handler)
                 .RequestQueue(_requestQueueName)
                 .BuildAsync();
 
@@ -97,6 +92,13 @@ static Task<IMessage> RpcHandler(IResponder.IContext context, IMessage request)
             Assert.Equal(State.Closing, states[0].Item2);
             Assert.Equal(State.Closing, states[1].Item1);
             Assert.Equal(State.Closed, states[1].Item2);
+            return;
+
+            static Task<IMessage> Handler(IResponder.IContext context, IMessage request)
+            {
+                IMessage m = context.Message(request.Body());
+                return Task.FromResult(m);
+            }
         }
 
         /// <summary>
@@ -120,13 +122,6 @@ public async Task SimulateRpcCommunicationWithAPublisherShouldSuccess()
 
             TaskCompletionSource<IMessage> tcs = CreateTaskCompletionSource<IMessage>();
 
-            Task MessageHandler(IContext context, IMessage message)
-            {
-                context.Accept();
-                tcs.SetResult(message);
-                return Task.CompletedTask;
-            }
-
             IConsumer consumer = await _connection.ConsumerBuilder()
                 .Queue(replyQueueSpec)
                 .MessageHandler(MessageHandler)
@@ -148,14 +143,22 @@ Task MessageHandler(IContext context, IMessage message)
             await responder.CloseAsync();
             await consumer.CloseAsync();
             await publisher.CloseAsync();
+            return;
+
+            Task MessageHandler(IContext context, IMessage msg)
+            {
+                context.Accept();
+                tcs.SetResult(msg);
+                return Task.CompletedTask;
+            }
         }
 
         /// <summary>
         /// In this test the client has to create a reply queue since is not provided by the user
         /// with the ReplyToQueue method
         /// </summary>
         [Fact]
-        public async Task ResponderClientPingPongWithDefault()
+        public async Task ResponderRequesterPingPongWithDefault()
         {
             Assert.NotNull(_connection);
 
@@ -179,10 +182,10 @@ public async Task ResponderClientPingPongWithDefault()
         }
 
         /// <summary>
-        /// In this test the client has to use the ReplyToQueue provided by the user
+        /// In this test the Requester has to use the ReplyToQueue provided by the user
         /// </summary>
         [Fact]
-        public async Task ResponderClientPingPongWithCustomReplyToQueueAndCorrelationIdSupplier()
+        public async Task ResponderRequesterPingPongWithCustomReplyToQueueAndCorrelationIdSupplier()
         {
             Assert.NotNull(_connection);
             Assert.NotNull(_management);
@@ -223,7 +226,7 @@ public async Task ResponderClientPingPongWithCustomReplyToQueueAndCorrelationIdS
         /// </summary>
         /// <exception cref="InvalidOperationException"></exception>
         [Fact]
-        public async Task ResponderClientOverridingTheRequestAndResponsePostProcessor()
+        public async Task ResponderRequesterOverridingTheRequestAndResponsePostProcessor()
         {
             Assert.NotNull(_connection);
             Assert.NotNull(_management);
@@ -278,7 +281,7 @@ public async Task ResponderClientOverridingTheRequestAndResponsePostProcessor()
         }
 
         [Fact]
-        public async Task RpcClientMultiThreadShouldBeSafe()
+        public async Task RequesterMultiThreadShouldBeSafe()
         {
             Assert.NotNull(_connection);
             const int messagesToSend = 99;
@@ -350,7 +353,7 @@ Task<IMessage> RpcHandler(IResponder.IContext context, IMessage request)
         /// The RPC client `PublishAsync` should raise a timeout exception if the server does not reply within the timeout
         /// </summary>
         [Fact]
-        public async Task RpcClientShouldRaiseTimeoutError()
+        public async Task RequesterShouldRaiseTimeoutError()
         {
             Assert.NotNull(_connection);