Added logic for re-queuing messages multiple times.

Author: Anton Vorontsov

src/RabbitMQ.Client.Core.DependencyInjection/Configuration/RabbitMqClientOptions.cs

@@ -22,7 +22,7 @@ public class RabbitMqClientOptions
         /// Collection of RabbitMQ server host names.
         /// It can be used when RabbitMQ HA cluster is running, and you want to connect multiple hosts.
         /// If HostNames collection is null or empty then HostName will be used to create connection.
-        /// Otherwise HostNames collection will be used and HostName property value will be ignored.
+        /// Otherwise, HostNames collection will be used and HostName property value will be ignored.
         /// </summary>
         /// <remarks>
         /// Has the second priority between properties TcpEndpoints, HostNames and HostName.

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

@@ -16,9 +16,6 @@ namespace RabbitMQ.Client.Core.DependencyInjection.Services
     /// </summary>
     public class MessageHandlingService : IMessageHandlingService
     {
-        // TODO: Убрать это.
-        const int ResendTimeout = 60;
-
         readonly IEnumerable<RabbitMqExchange> _exchanges;
         readonly IEnumerable<MessageHandlerContainer> _messageHandlerContainers;
         readonly ILogger<MessageHandlingService> _logger;
@@ -204,26 +201,47 @@ async Task HandleFailedMessageProcessing(BasicDeliverEventArgs eventArgs, IQueue
                 _logger.LogWarning($"DeadLetterExchange has not been configured for an exchange \"{eventArgs.Exchange}\". The message won't be re-queued.");
                 return;
             }
+
+            if (exchange.Options.RequeueTimeoutMilliseconds < 1)
+            {
+                _logger.LogWarning($"The value RequeueTimeoutMilliseconds for an exchange \"{eventArgs.Exchange}\" less than 1 millisecond. Configuration is invalid. The message won't be re-queued.");
+                return;
+            }
 
-            // TODO: Проверять параметры.
-            // TODO: добавить методы расширения для delayed сообщений.
-            // TODO: перепроверить, как будут создаваться очереди для delayed сообщений - переделать на milliseconds в именовании.
-            // TODO: Закинуть это дальше.
+            if (exchange.Options.RequeueAttempts < 1)
+            {
+                _logger.LogWarning($"The value RequeueAttempts for an exchange \"{eventArgs.Exchange}\" less than 1. Configuration is invalid. The message won't be re-queued.");
+                return;
+            }
+
             if (eventArgs.BasicProperties.Headers is null)
             {
                 eventArgs.BasicProperties.Headers = new Dictionary<string, object>();
             }
 
             if (!eventArgs.BasicProperties.Headers.ContainsKey("re-queue-attempts"))
             {
-                eventArgs.BasicProperties.Headers.Add("re-queue-attempts", true);
-                await queueService.SendAsync(eventArgs.Body, eventArgs.BasicProperties, eventArgs.Exchange, eventArgs.RoutingKey, ResendTimeout);
-                _logger.LogInformation("The failed message has been re-queued.");
+                eventArgs.BasicProperties.Headers.Add("re-queue-attempts", 1);
+                await RequeueMessage(eventArgs, queueService, exchange.Options.RequeueTimeoutMilliseconds);
+                return;
+            }
+            
+            var currentAttempt = (int)eventArgs.BasicProperties.Headers["re-queue-attempts"];
+            if (currentAttempt < exchange.Options.RequeueAttempts)
+            {
+                eventArgs.BasicProperties.Headers["re-queue-attempts"] = currentAttempt + 1;
+                await RequeueMessage(eventArgs, queueService, exchange.Options.RequeueTimeoutMilliseconds);
             }
             else
             {
-                _logger.LogInformation("The failed message would not be re-queued.");
+                _logger.LogInformation("The failed message would not be re-queued. Attempts limit exceeded.");   
             }
         }
+
+        async Task RequeueMessage(BasicDeliverEventArgs eventArgs, IQueueService queueService, int timeoutMilliseconds)
+        {
+            await queueService.SendAsync(eventArgs.Body, eventArgs.BasicProperties, eventArgs.Exchange, eventArgs.RoutingKey, timeoutMilliseconds);
+            _logger.LogInformation("The failed message has been re-queued.");
+        }
     }
 }

tests/RabbitMQ.Client.Core.DependencyInjection.Tests/UnitTests/MessageHandlingServiceTests.cs

@@ -228,10 +228,10 @@ static IEnumerable<MessageHandlerOrderingModel> GetMessageHandlerOrderingModels(
 
         static IEnumerable<MessageHandlerOrderingContainerTestModel> GetTestingOrderingModels(
             HandleMessageReceivingEventTestDataModel testDataModel,
-            Mock<IMessageHandler> messageHandlerMock,
-            Mock<IAsyncMessageHandler> asyncMessageHandlerMock,
-            Mock<INonCyclicMessageHandler> nonCyclicMessageHandlerMock,
-            Mock<IAsyncNonCyclicMessageHandler> asyncNonCyclicMessageHandlerMock)
+            IMock<IMessageHandler> messageHandlerMock,
+            IMock<IAsyncMessageHandler> asyncMessageHandlerMock,
+            IMock<INonCyclicMessageHandler> nonCyclicMessageHandlerMock,
+            IMock<IAsyncNonCyclicMessageHandler> asyncNonCyclicMessageHandlerMock)
         {
             var collection = new List<MessageHandlerOrderingContainerTestModel>
             {
@@ -265,13 +265,8 @@ static IEnumerable<MessageHandlerOrderingContainerTestModel> GetTestingOrderingM
             var orderedCollection = collection.OrderByDescending(x => x.OrderValue)
                 .ThenByDescending(x => x.MessageHandler.GetHashCode())
                 .ToList();
-            foreach (var item in orderedCollection)
+            foreach (var item in orderedCollection.Where(item => item.ShouldTrigger))
             {
-                if (!item.ShouldTrigger)
-                {
-                    continue;
-                }
-
                 item.CallOrder = callOrder++;
             }
             return orderedCollection;