x-sent-timestamp & ForwardAsync (#122)

* New header

* Добавлена установка нового заголовка

* Новый Forward Async

Sources/Contour/Headers.cs

@@ -67,6 +67,8 @@ public class Headers
         /// Максимальное количество сообщений в очереди.
         /// </summary>
         public static readonly string QueueMaxLength = "x-max-length";
+
+        public static readonly string SentTimestamp = "x-sent-timestamp";
 
         /// <summary>
         /// Коллекция заголовков, значения которых менять запрещено
@@ -187,6 +189,13 @@ public static string GetString(string key)
             return headers;
         }
 
+        public static void ApplySentTimestamp(IDictionary<string, object> headers)
+        {
+            var sent = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+
+            headers[SentTimestamp] = sent;
+        }
+
         /// <summary>
         /// Применяет к коллекции заголовков установку заголовка <c>Ttl</c>.
         /// </summary>

Sources/Contour/Receiving/DefaultConsumingContext.cs

@@ -82,6 +82,11 @@ public async Task ForwardAsync(string label)
             await this.Delivery.Forward(label.ToMessageLabel(), this.Message.Payload);
         }
 
+        public async Task ForwardAsync<TOut>(string label, TOut payload) where TOut : class
+        {
+            await this.Delivery.Forward(label.ToMessageLabel(), payload);
+        }
+
         /// <summary>
         /// Пересылает сообщение, устанавливая указанную метку.
         /// </summary>

Sources/Contour/Receiving/IConsumingContext.cs

@@ -46,6 +46,13 @@ public interface IConsumingContext<T>
         /// </summary>
         /// <param name="label">Новая метка, с которой пересылается сообщение.</param>
         Task ForwardAsync(string label);
+
+        /// <summary>
+        /// Пересылает сообщение, устанавливая указанную метку.
+        /// </summary>
+        /// <param name="label">Новая метка, с которой пересылается сообщение.</param>
+        /// <param name="payload">Новое содержимое сообщения.</param>
+        Task ForwardAsync<TOut>(string label, TOut payload) where TOut : class;
 
         /// <summary>
         /// Пересылает сообщение, устанавливая указанную метку.

Sources/Contour/Sending/AbstractSender.cs

@@ -96,6 +96,7 @@ public virtual bool CanRoute(MessageLabel label)
         /// <returns>Request processing task.</returns>
         public Task<T> Request<T>(object payload, IDictionary<string, object> headers) where T : class
         {
+            Headers.ApplySentTimestamp(headers);
             var message = new Message(this.Configuration.Label, headers, payload);
 
             var exchange = new MessageExchange(message, typeof(T));
@@ -138,6 +139,7 @@ public virtual bool CanRoute(MessageLabel label)
         [Obsolete("Необходимо использовать метод Send с указанием метки сообщения.")]
         public Task Send(object payload, IDictionary<string, object> headers)
         {
+            Headers.ApplySentTimestamp(headers);
             var message = new Message(this.Configuration.Label, headers, payload);
 
             return this.ProcessFilter(message, null);
@@ -185,6 +187,7 @@ public Task Send(object payload, PublishingOptions options)
         /// <returns>Request processing task.</returns>
         public Task<T> Request<T>(MessageLabel label, object payload, IDictionary<string, object> headers) where T : class
         {
+            Headers.ApplySentTimestamp(headers);
             if (!headers.ContainsKey(Headers.CorrelationId))
             {
                 headers[Headers.CorrelationId] = Guid.NewGuid().ToString("n");
@@ -215,6 +218,7 @@ public Task Send(object payload, PublishingOptions options)
         /// <returns>Задача выполнения отправки сообщения.</returns>
         public Task Send(MessageLabel label, object payload, IDictionary<string, object> headers, string connectionKey)
         {
+            Headers.ApplySentTimestamp(headers);
             var message = new Message(this.Configuration.Label.Equals(MessageLabel.Any) ? label : this.Configuration.Label, headers, payload);
 
             return this.ProcessFilter(message, connectionKey);