Merge pull request #47 from AntonyVorontsov/feature/retries-for-resending

WIP: Started making retries for failed messages.

Author: antonyvorontsov

docs/changelog.md

@@ -2,6 +2,16 @@
 
 All notable changes to this library will be documented in this file.
 
+## [4.2.0] - will be drafted
+
+### Added
+
+- Options that can allow configuring behaviour of re-queueing messages. New properties `RequeueTimeoutMilliseconds` and `RequeueAttempts` added to `RabbitMqExchangeOptions`.
+
+### Changed
+
+- **Breaking!** Now all `Send` or `SendAsync` methods of `IProducingService` (and `IQueueService`) with delay parameter use milliseconds instead of seconds.
+
 ## [4.1.1] - 2020-05-30
 
 ### Fixed

docs/exchange-configuration.md

@@ -42,6 +42,8 @@ And the `appsettings.json` file will be like this.
     "AutoDelete": false,
     "DeadLetterExchange": "default.dlx.exchange",
     "RequeueFailedMessages": true,
+    "RequeueTimeoutMilliseconds": 200,
+    "RequeueAttempts": 2,
     "Arguments": { "key": "value" },
     "Queues": [
       {
@@ -65,6 +67,8 @@ Exchanges can be configured with properties:
 - `AutoDelete` - an option for exchange auto deleting. The default value is `false`.
 - `Arguments` - a dictionary of additional arguments. The default value is `null`.
 - `RequeueFailedMessages` - an option that specifies behaviour of re-queueing failed messages with certain delay through the dead-letter-exchange. The default value is `true`. The mechanism of sending delayed messages is covered in the [documentation](message-production.md).
+- `RequeueTimeoutMilliseconds` - timeout in milliseconds after which the message will be re-queued. The default value is 200.
+- `RequeueAttempts` - a number of attempts which queueing service will try to re-queue a message. The default value is 2.
 - `DeadLetterExchange` - a value for dead-letter-exchange. The default value for the dead-letter-exchange name is `"default.dlx.exchange"`.
 - `Queues` - a collection of queues bound to the exchange.
 

docs/message-consumption.md

@@ -303,13 +303,13 @@ services.AddRabbitMqClient(clientConfiguration)
 
 ### Workflow of message handling
 
-The message handling process is organized as follows:
+The message handling process organized as follows:
 
 - `IQueueMessage` receives a message and delegates it to `IMessageHandlingService`.
 - `IMessageHandlingService` gets a message (as a byte array) and decodes it to the UTF8 string. It also checks if there are any message handlers in a combined collection of `IMessageHandler`, `IAsyncMessageHandler`, `INonCyclicMessageHandler` and `IAsyncNonCyclicMessageHandler` instances and forwards a message to them.
 - All subscribed message handlers (`IMessageHandler`, `IAsyncMessageHandler`, `INonCyclicMessageHandler`, `IAsyncNonCyclicMessageHandler`) process the given message in a given or a default order.
 - `IMessageHandlingService` acknowledges the message by its `DeliveryTag`.
-- If any exception occurs `IMessageHandlingService` acknowledges the message anyway and checks if the message has to be re-send. If exchange option `RequeueFailedMessages` is set `true` then `IMessageHandlingService` adds a header `"requeued"` to the message and sends it again with delay in 60 seconds. Mechanism of sending delayed messages covered in the message production [documentation](message-production.md).
+- If any exception occurs `IMessageHandlingService` acknowledges the message anyway and checks if the message has to be re-send. If exchange option `RequeueFailedMessages` is set `true` then `IMessageHandlingService` adds a header `"re-queue-attempts"` to the message and sends it again with delay in value of `RequeueTimeoutMilliseconds` (default is 200 milliseconds). The number of attempts is configurable and re-delivery will be made that many times as the value of `RequeueAttempts` property. Mechanism of sending delayed messages covered in the message production [documentation](message-production.md).
 - If any exception occurs within handling the message that has been already re-sent that message will not be re-send again (re-send happens only once).
 
 ### Batch message handlers

docs/message-production.md

@@ -90,24 +90,24 @@ queueService.Send(bytes, properties, exchangeName: "exchange.name", routingKey:
 await queueService.SendAsync(bytes, properties, exchangeName: "exchange.name", routingKey: "routing.key");
 ```
 
-You are also allowed to send messages with delay (in seconds). All of previously listed methods have an overload that takes a `delay` parameter.
+You are also allowed to send messages with delay (in milliseconds). All of previously listed methods have an overload that takes a `delay` parameter.
 
 ```c#
 // Objects
-queueService.Send(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
-await queueService.SendAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
+queueService.Send(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
+await queueService.SendAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
 
 // Json
-queueService.SendJson(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
-await queueService.SendJsonAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
+queueService.SendJson(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
+await queueService.SendJsonAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
 
 // Strings
-queueService.SendString(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
-await queueService.SendStringAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
+queueService.SendString(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
+await queueService.SendStringAsync(message, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
 
 // Bytes
-queueService.Send(bytes, properties, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
-await queueService.SendAsync(bytes, properties, exchangeName: "exchange.name", routingKey: "routing.key", secondsDelay: 10);
+queueService.Send(bytes, properties, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
+await queueService.SendAsync(bytes, properties, exchangeName: "exchange.name", routingKey: "routing.key", millisecondsDelay: 10);
 ```
 
 ### Mechanism of sending delayed messages
@@ -116,15 +116,15 @@ The implementation of sending deferred (delayed) messages in this project is qui
 The image below shows a model of the whole process of passing the message from the producer to the consumer.
 ![Model of sending delayed messages](./images/delayed-message-model.png)
 
-**Prerequisites.** Let's say that producer want to send a message to the exchange **"Exchange B"** with a routing key **"routing.key"**, and a delay in **30 seconds**.
+**Prerequisites.** Let's say that producer want to send a message to the exchange **"Exchange B"** with a routing key **"routing.key"**, and a delay in **30 milliseconds**.
  - Message goes to the exchange **"Exchange A"** whose responsibility is to manage delaying (storing) the message.
- - After that a queue with a compound name and special arguments is being created. Name consists of three parts: the routing key of the sent message, a word "delayed", and a number of delay seconds .
+ - After that a queue with a compound name and special arguments is being created. Name consists of three parts: the routing key of the sent message, a word "delayed", and a number of delay milliseconds .
 Queue arguments are as follows.
 ```
 x-dead-letter-exchange : Exchange В
 x-dead-letter-routing-key : routing.key
-x-message-ttl : secondsDelay * 1000
-x-expires : secondsDelay * 1000 + 60000
+x-message-ttl : millisecondsDelay
+x-expires : millisecondsDelay + 60000
 ```
  - A message, which gets in that queue, will have a specified ttl (time to live), and an exchange to which the message will be sent after expiration.
  - That new queue bounds to the **Exchange A**. That queue will be automatically deleted if there are no more messages in it within a minute.

readme.md

@@ -65,7 +65,7 @@ _queueService.Send(
     @object: messageObject,
     exchangeName: "exchange.name",
     routingKey: "routing.key",
-    secondsDelay: 10);
+    millisecondsDelay: 10);
 ```
 
  The mechanism of sending delayed messages described in the [documentation](./docs/message-production.md). Dive into it for more detailed information.
@@ -283,14 +283,14 @@ When the message collection is full to the size of `PrefetchCount` they are pass
 
 ## Advanced usage and nuances
 
-RabbitMQ client implemented in this library (class which implements `IQueueService`) opens two connections to the RabbitMQ server. One connection is used for message production and the other one is for message consumption.
+RabbitMQ client implemented in this library (class which implements `IQueueService`) opens two connections to the RabbitMQ server. One connection is used for message production, and the other one is for message consumption.
 This behavior covered in the [advanced usage documentation file](./docs/advanced-usage.md), dive into it deeply if you want to control the client behavior tighter.
 
-There is also an [example project](./examples/Examples.AdvancedConfiguration) that demonstrates an advances usage of the RabbitMQ client.
+There is also an [example project](./examples/Examples.AdvancedConfiguration) that demonstrates an advanced usage of the RabbitMQ client.
 
 ## Changelog
 
-All notable changes are being tracked in the [changelog](./docs/changelog.md) file.
+All notable changes covered in the [changelog](./docs/changelog.md) file.
 
 ## License
 

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/Configuration/RabbitMqExchangeOptions.cs

@@ -28,10 +28,20 @@ public class RabbitMqExchangeOptions
         public string DeadLetterExchange { get; set; } = "default.dlx.exchange";
 
         /// <summary>
-        /// Option to re-queue failed messages (once).
+        /// Option to re-queue failed messages.
         /// </summary>
         public bool RequeueFailedMessages { get; set; } = true;
 
+        /// <summary>
+        /// Re-queue message attempts.
+        /// </summary>
+        public int RequeueAttempts { get; set; } = 2;
+        
+        /// <summary>
+        /// Re-queue timeout in milliseconds.
+        /// </summary>
+        public int RequeueTimeoutMilliseconds { get; set; } = 200;
+
         /// <summary>
         /// Additional arguments.
         /// </summary>

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

@@ -34,8 +34,8 @@ public interface IProducingService
         /// <param name="object">Object message.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        void Send<T>(T @object, string exchangeName, string routingKey, int secondsDelay) where T : class;
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        void Send<T>(T @object, string exchangeName, string routingKey, int millisecondsDelay) where T : class;
 
         /// <summary>
         /// Send a message.
@@ -51,8 +51,8 @@ public interface IProducingService
         /// <param name="json"></param>
         /// <param name="exchangeName"></param>
         /// <param name="routingKey"></param>
-        /// <param name="secondsDelay">Delay time.</param>
-        void SendJson(string json, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        void SendJson(string json, string exchangeName, string routingKey, int millisecondsDelay);
 
         /// <summary>
         /// Send a message.
@@ -68,8 +68,8 @@ public interface IProducingService
         /// <param name="message">Message.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        void SendString(string message, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        void SendString(string message, string exchangeName, string routingKey, int millisecondsDelay);
 
         /// <summary>
         /// Send a message.
@@ -87,8 +87,8 @@ public interface IProducingService
         /// <param name="properties">Message properties.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        void Send(ReadOnlyMemory<byte> bytes, IBasicProperties properties, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        void Send(ReadOnlyMemory<byte> bytes, IBasicProperties properties, string exchangeName, string routingKey, int millisecondsDelay);
 
         /// <summary>
         /// Send a message asynchronously.
@@ -106,8 +106,8 @@ public interface IProducingService
         /// <param name="object">Object message.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        Task SendAsync<T>(T @object, string exchangeName, string routingKey, int secondsDelay) where T : class;
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        Task SendAsync<T>(T @object, string exchangeName, string routingKey, int millisecondsDelay) where T : class;
 
         /// <summary>
         /// Send a message asynchronously.
@@ -123,8 +123,8 @@ public interface IProducingService
         /// <param name="json">Json message.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        Task SendJsonAsync(string json, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        Task SendJsonAsync(string json, string exchangeName, string routingKey, int millisecondsDelay);
 
         /// <summary>
         /// Send a message asynchronously.
@@ -140,8 +140,8 @@ public interface IProducingService
         /// <param name="message">Message.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        Task SendStringAsync(string message, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        Task SendStringAsync(string message, string exchangeName, string routingKey, int millisecondsDelay);
 
         /// <summary>
         /// Send a message asynchronously.
@@ -159,7 +159,7 @@ public interface IProducingService
         /// <param name="properties">Message properties.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="routingKey">Routing key.</param>
-        /// <param name="secondsDelay">Delay time.</param>
-        Task SendAsync(ReadOnlyMemory<byte> bytes, IBasicProperties properties, string exchangeName, string routingKey, int secondsDelay);
+        /// <param name="millisecondsDelay">Delay time in milliseconds.</param>
+        Task SendAsync(ReadOnlyMemory<byte> bytes, IBasicProperties properties, string exchangeName, string routingKey, int millisecondsDelay);
     }
 }