Added integration test for re-queuing messages.

Author: Anton Vorontsov

src/RabbitMQ.Client.Core.DependencyInjection/Services/MessageHandlingService.cs

@@ -45,17 +45,15 @@ public async Task HandleMessageReceivingEvent(BasicDeliverEventArgs eventArgs, I
 
                 var matchingRoutes = GetMatchingRoutePatterns(eventArgs.Exchange, eventArgs.RoutingKey);
                 await ProcessMessage(eventArgs.Exchange, message, queueService, matchingRoutes).ConfigureAwait(false);
+                queueService.ConsumingChannel.BasicAck(eventArgs.DeliveryTag, false);
                 _logger.LogInformation($"Message processing finished successfully. Acknowledge has been sent with deliveryTag {eventArgs.DeliveryTag}.");
             }
             catch (Exception exception)
             {
+                queueService.ConsumingChannel.BasicAck(eventArgs.DeliveryTag, false);
                 _logger.LogError(new EventId(), exception, $"An error occurred while processing received message with the delivery tag {eventArgs.DeliveryTag}.");
                 await HandleFailedMessageProcessing(eventArgs, queueService).ConfigureAwait(false);
             }
-            finally
-            {
-                queueService.ConsumingChannel.BasicAck(eventArgs.DeliveryTag, false);
-            }
         }
 
         IEnumerable<string> GetMatchingRoutePatterns(string exchange, string routingKey)

tests/RabbitMQ.Client.Core.DependencyInjection.Tests/IntegrationTests/QueueServiceTests.cs

@@ -19,6 +19,7 @@ public class QueueServiceTests
         const string DefaultExchangeName = "exchange.name";
         const string FirstRoutingKey = "first.routing.key";
         const string SecondRoutingKey = "second.routing.key";
+        const int RequeueAttempts = 4;
 
         [Fact]
         public async Task ShouldProperlyPublishAndConsumeMessages()
@@ -47,7 +48,7 @@ public async Task ShouldProperlyPublishAndConsumeMessages()
                 .AddAsyncMessageHandlerTransient<StubAsyncMessageHandler>(SecondRoutingKey)
                 .AddAsyncNonCyclicMessageHandlerTransient<StubAsyncNonCyclicMessageHandler>(SecondRoutingKey);
 
-            var serviceProvider = serviceCollection.BuildServiceProvider();
+            await using var serviceProvider = serviceCollection.BuildServiceProvider();
             var queueService = serviceProvider.GetRequiredService<IQueueService>();
             queueService.StartConsuming();
 
@@ -66,6 +67,50 @@ public async Task ShouldProperlyPublishAndConsumeMessages()
             resetEvent.WaitOne(_globalTestsTimeout);
             callerMock.Verify(x => x.CallAsync(It.IsAny<string>()), Times.Exactly(2));
         }
+        
+        [Fact]
+        public async Task ShouldProperlyRequeueMessages()
+        {
+            var connectionFactoryMock = new Mock<RabbitMqConnectionFactory> { CallBase = true }
+                .As<IRabbitMqConnectionFactory>();
+
+            AsyncEventingBasicConsumer consumer = null;
+            connectionFactoryMock.Setup(x => x.CreateConsumer(It.IsAny<IModel>()))
+                .Returns<IModel>(channel =>
+                {
+                    consumer = new AsyncEventingBasicConsumer(channel);
+                    return consumer;
+                });
+
+            var callerMock = new Mock<IStubCaller>();
+
+            var serviceCollection = new ServiceCollection();
+            serviceCollection
+                .AddSingleton(connectionFactoryMock.Object)
+                .AddSingleton(callerMock.Object)
+                .AddRabbitMqClient(GetClientOptions())
+                .AddConsumptionExchange(DefaultExchangeName, GetExchangeOptions())
+                .AddMessageHandlerTransient<StubExceptionMessageHandler>(FirstRoutingKey);
+
+            await using var serviceProvider = serviceCollection.BuildServiceProvider();
+            var queueService = serviceProvider.GetRequiredService<IQueueService>();
+            queueService.StartConsuming();
+
+            using var resetEvent = new AutoResetEvent(false);
+            consumer.Received += (sender, @event) =>
+            {
+                resetEvent.Set();
+                return Task.CompletedTask;
+            };
+
+            await queueService.SendAsync(new { Message = "message" }, DefaultExchangeName, FirstRoutingKey);
+
+            for (var i = 1; i <= RequeueAttempts + 1; i++)
+            {
+                resetEvent.WaitOne(_globalTestsTimeout);
+            }
+            callerMock.Verify(x => x.Call(It.IsAny<string>()), Times.Exactly(RequeueAttempts + 1));
+        }
 
         static RabbitMqClientOptions GetClientOptions() =>
             new RabbitMqClientOptions
@@ -82,6 +127,8 @@ static RabbitMqExchangeOptions GetExchangeOptions() =>
             {
                 Type = "direct",
                 DeadLetterExchange = "exchange.dlx",
+                RequeueAttempts = RequeueAttempts,
+                RequeueTimeoutMilliseconds = 50,
                 Queues = new List<RabbitMqQueueOptions>
                 {
                     new RabbitMqQueueOptions

tests/RabbitMQ.Client.Core.DependencyInjection.Tests/Stubs/StubExceptionMessageHandler.cs

@@ -0,0 +1,21 @@
+using System;
+using RabbitMQ.Client.Core.DependencyInjection.MessageHandlers;
+
+namespace RabbitMQ.Client.Core.DependencyInjection.Tests.Stubs
+{
+    public class StubExceptionMessageHandler : IMessageHandler
+    {
+        readonly IStubCaller _caller;
+
+        public StubExceptionMessageHandler(IStubCaller caller)
+        {
+            _caller = caller;
+        }
+
+        public void Handle(string message, string routingKey)
+        {
+            _caller.Call($"{message}:{routingKey}");
+            throw new Exception();
+        }
+    }
+}