Add delayed message support for RabbitMQ.

Author: yang-xiaodong

src/DotNetCore.CAP.RabbitMQ/ICapPublisher.Extensions.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Threading;
+using System.Threading.Tasks;
+using DotNetCore.CAP.Messages;
+
+// ReSharper disable once CheckNamespace
+namespace DotNetCore.CAP
+{
+    public static class CapPublisherExtensions
+    {
+        /// <summary>
+        /// Schedule a message to be published at the feature time with callback name.
+        /// <para>SHOULD BE ENABLE RabbitMQ <b>rabbitmq_delayed_message_exchange</b>  PLUGINS.</para>
+        /// </summary>
+        /// <typeparam name="T">content object</typeparam>
+        /// <param name="delayTime">The delay for message to published.</param>
+        /// <param name="name">the topic name or exchange router key.</param>
+        /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
+        /// <param name="callbackName">callback subscriber name.</param>
+        /// <param name="cancellationToken"></param>
+        /// <param name="publisher"></param>
+        public static async Task PublishDelayAsync<T>(this ICapPublisher publisher, TimeSpan delayTime,
+            string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default)
+        {
+            var dic = new Dictionary<string, string?>
+            {
+                {Headers.CallbackName, callbackName},
+                {"x-delay", delayTime.TotalMilliseconds.ToString(CultureInfo.InvariantCulture)}
+            };
+
+            await publisher.PublishAsync(name, contentObj, dic, cancellationToken).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Schedule a message to be published at the feature time with custom headers.
+        /// <para>SHOULD BE ENABLE RabbitMQ <b>rabbitmq_delayed_message_exchange</b>  PLUGINS.</para>
+        /// </summary>
+        /// <typeparam name="T">content object</typeparam>
+        /// <param name="publisher"></param>
+        /// <param name="delayTime">The delay for message to published.</param>
+        /// <param name="name">the topic name or exchange router key.</param>
+        /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
+        /// <param name="headers">message additional headers.</param>
+        /// <param name="cancellationToken"></param>
+        public static async Task PublishDelayAsync<T>(this ICapPublisher publisher, TimeSpan delayTime,
+            string name, T? contentObj, Dictionary<string, string?> headers, CancellationToken cancellationToken = default)
+        {
+            headers.Add("x-delay", delayTime.TotalMilliseconds.ToString(CultureInfo.InvariantCulture));
+
+            await publisher.PublishAsync(name, contentObj, headers, cancellationToken).ConfigureAwait(false);
+        }
+    }
+}

src/DotNetCore.CAP.RabbitMQ/IConnectionChannelPool.Default.cs

@@ -6,6 +6,7 @@
 using System.Diagnostics;
 using System.Reflection;
 using System.Threading;
+using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using RabbitMQ.Client;
@@ -129,6 +130,8 @@ public virtual IModel Rent()
             try
             {
                 model = GetConnection().CreateModel();
+                model.ExchangeDeclare(Exchange, RabbitMQOptions.ExchangeType, true);
+                model.ConfirmSelect();
             }
             catch (Exception e)
             {

src/DotNetCore.CAP.RabbitMQ/ITransport.RabbitMQ.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Internal;
@@ -27,7 +28,7 @@ public RabbitMQTransport(
             _exchange = _connectionChannelPool.Exchange;
         }
 
-        public BrokerAddress BrokerAddress => new ("RabbitMQ", _connectionChannelPool.HostAddress);
+        public BrokerAddress BrokerAddress => new("RabbitMQ", _connectionChannelPool.HostAddress);
 
         public Task<OperateResult> SendAsync(TransportMessage message)
         {
@@ -36,17 +37,28 @@ public Task<OperateResult> SendAsync(TransportMessage message)
             {
                 channel = _connectionChannelPool.Rent();
 
-                //channel.ConfirmSelect();
-
                 var props = channel.CreateBasicProperties();
                 props.DeliveryMode = 2;
                 props.Headers = message.Headers.ToDictionary(x => x.Key, x => (object?)x.Value);
 
-                channel.ExchangeDeclare(_exchange, RabbitMQOptions.ExchangeType, true);
+                if (message.Headers.ContainsKey("x-delay"))
+                {
+                    var delayExchangeName = _exchange + "_delayed";
+                    channel.ExchangeDeclare(exchange: delayExchangeName, type: "x-delayed-message", durable: true, autoDelete: false,
+                        new Dictionary<string, object>
+                        {
+                            ["x-delayed-type"] = "topic"
+                        });
+                    channel.ExchangeBind(_exchange, delayExchangeName, message.GetName());
 
-                channel.BasicPublish(_exchange, message.GetName(), props, message.Body);
+                    channel.BasicPublish(delayExchangeName, message.GetName(), props, message.Body);
+                }
+                else
+                {
+                    channel.BasicPublish(_exchange, message.GetName(), props, message.Body);
+                }
 
-                //channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));
+                channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));
 
                 _logger.LogInformation("CAP message '{0}' published, internal id '{1}'", message.GetName(), message.GetId());
 

src/DotNetCore.CAP/Internal/IMessageSender.Default.cs

@@ -49,7 +49,7 @@ public async Task<OperateResult> SendAsync(MediumMessage message)
             {
                 var executedResult = await SendWithoutRetryAsync(message);
                 result = executedResult.Item2;
-                if (result == OperateResult.Success)
+                if (result.Equals(OperateResult.Success))
                 {
                     return result;
                 }

src/DotNetCore.CAP/Internal/ISubscribeDispatcher.Default.cs

@@ -23,6 +23,7 @@ internal class SubscribeDispatcher : ISubscribeDispatcher
         private readonly ILogger _logger;
         private readonly IServiceProvider _provider;
         private readonly CapOptions _options;
+        private readonly string? _hostName;
 
         // diagnostics listener
         // ReSharper disable once InconsistentNaming
@@ -39,6 +40,7 @@ public SubscribeDispatcher(
 
             _dataStorage = _provider.GetRequiredService<IDataStorage>();
             Invoker = _provider.GetRequiredService<ISubscribeInvoker>();
+            _hostName = GenerateHostnameInstanceId();
         }
 
         private ISubscribeInvoker Invoker { get; }
@@ -66,13 +68,13 @@ public async Task<OperateResult> DispatchAsync(MediumMessage message, ConsumerEx
             OperateResult result;
 
             //record instance id
-            message.Origin.Headers[Headers.ExecutionInstanceId] = GenerateHostnameInstanceId();
+            message.Origin.Headers[Headers.ExecutionInstanceId] = _hostName;
 
             do
             {
                 var (shouldRetry, operateResult) = await ExecuteWithoutRetryAsync(message, descriptor, cancellationToken);
                 result = operateResult;
-                if (result == OperateResult.Success)
+                if (result.Equals(OperateResult.Success))
                 {
                     return result;
                 }

src/DotNetCore.CAP/OperateResult.cs

@@ -3,78 +3,44 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 
 namespace DotNetCore.CAP
 {
     /// <summary>
     /// Represents the result of an consistent message operation.
     /// </summary>
-    public class OperateResult
+    public struct OperateResult : IEqualityComparer<OperateResult>
     {
-        // ReSharper disable once InconsistentNaming
+        private readonly OperateError? _operateError = null;
 
-        // ReSharper disable once FieldCanBeMadeReadOnly.Local
-        private readonly List<OperateError> _errors = new List<OperateError>();
+        public OperateResult(bool succeeded, Exception? exception = null, OperateError? error = null)
+        {
+            Succeeded = succeeded;
+            Exception = exception;
+            _operateError = error;
+        }
 
-        /// <summary>
-        /// Flag indicating whether if the operation succeeded or not.
-        /// </summary>
         public bool Succeeded { get; set; }
 
         public Exception? Exception { get; set; }
 
-        /// <summary>
-        /// An <see cref="IEnumerable{T}" /> of <see cref="OperateError" />s containing an errors
-        /// that occurred during the operation.
-        /// </summary>
-        /// <value>An <see cref="IEnumerable{T}" /> of <see cref="OperateError" />s.</value>
-        public IEnumerable<OperateError> Errors => _errors;
+        public static OperateResult Success => new(true);
 
-        /// <summary>
-        /// Returns an <see cref="OperateResult" /> indicating a successful identity operation.
-        /// </summary>
-        /// <returns>An <see cref="OperateResult" /> indicating a successful operation.</returns>
-        public static OperateResult Success { get; } = new OperateResult {Succeeded = true};
+        public static OperateResult Failed(Exception ex, OperateError? errors = null) => new(false, ex, errors);
 
-        /// <summary>
-        /// Creates an <see cref="OperateResult" /> indicating a failed operation, with a list of <paramref name="errors" /> if
-        /// applicable.
-        /// </summary>
-        /// <param name="ex">Operate Result exception</param>
-        /// <param name="errors">An optional array of <see cref="OperateError" />s which caused the operation to fail.</param>
-        /// <returns>
-        /// An <see cref="OperateResult" /> indicating a failed operation, with a list of <paramref name="errors" /> if
-        /// applicable.
-        /// </returns>
-        public static OperateResult Failed(Exception ex, params OperateError[] errors)
+        public override string ToString()
         {
-            var result = new OperateResult
-            {
-                Succeeded = false,
-                Exception = ex
-            };
-            if (errors != null)
-            {
-                result._errors.AddRange(errors);
-            }
+            return Succeeded ? "Succeeded" : $"Failed : {_operateError?.Code}";
+        }
 
-            return result;
+        public bool Equals(OperateResult x, OperateResult y)
+        {
+            return x.Succeeded == y.Succeeded;
         }
 
-        /// <summary>
-        /// Converts the value of the current <see cref="OperateResult" /> object to its equivalent string representation.
-        /// </summary>
-        /// <returns>A string representation of the current <see cref="OperateResult" /> object.</returns>
-        /// <remarks>
-        /// If the operation was successful the ToString() will return "Succeeded" otherwise it returned
-        /// "Failed : " followed by a comma delimited list of error codes from its <see cref="Errors" /> collection, if any.
-        /// </remarks>
-        public override string ToString()
+        public int GetHashCode(OperateResult obj)
         {
-            return Succeeded
-                ? "Succeeded"
-                : string.Format("{0} : {1}", "Failed", string.Join(",", Errors.Select(x => x.Code).ToList()));
+            return HashCode.Combine(obj._operateError, obj.Succeeded, obj.Exception);
         }
     }