From 7b7b5d10c12898b56e98c3cecf5dfdaa66c6e3a7 Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Tue, 30 Dec 2025 08:41:00 +0300 Subject: [PATCH 1/4] initial --- SW.Bus.RabbitMqExtensions/IConsumeExtended.cs | 8 + SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs | 10 ++ SW.Bus.RabbitMqExtensions/QueueOptions.cs | 14 ++ .../SW.Bus.RabbitMqExtensions.csproj | 13 ++ SW.Bus.SampleWeb/SW.Bus.SampleWeb.csproj | 2 +- SW.Bus.sln | 6 + SW.Bus/BasicPublisher.cs | 14 +- SW.Bus/BusOptions.cs | 7 +- SW.Bus/ConsumerDefinition.cs | 39 +++-- SW.Bus/ConsumerDiscovery.cs | 42 ++++- SW.Bus/ConsumerRunner.cs | 2 +- SW.Bus/ConsumersService.cs | 158 +++++++++++++++++- SW.Bus/Publisher.cs | 18 +- SW.Bus/QueueOptions.cs | 16 -- SW.Bus/SW.Bus.csproj | 10 +- 15 files changed, 305 insertions(+), 54 deletions(-) create mode 100644 SW.Bus.RabbitMqExtensions/IConsumeExtended.cs create mode 100644 SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs create mode 100644 SW.Bus.RabbitMqExtensions/QueueOptions.cs create mode 100644 SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj delete mode 100644 SW.Bus/QueueOptions.cs diff --git a/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs b/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs new file mode 100644 index 0000000..fbfe4a3 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs @@ -0,0 +1,8 @@ +using SW.PrimitiveTypes; + +namespace SW.Bus.RabbitMqExtensions; + +public interface IConsumeExtended : IConsume +{ + Task> GetQueOptions(); +} \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs b/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs new file mode 100644 index 0000000..7cabac8 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs @@ -0,0 +1,10 @@ +using SW.PrimitiveTypes; + +namespace SW.Bus.RabbitMqExtensions; + +public interface IRabbitMqPublish : IPublish +{ + Task Publish(TMessage message, byte priority); + Task Publish(string messageTypeName, string message, byte priority); + Task Publish(string messageTypeName, byte[] message, byte priority); +} \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/QueueOptions.cs b/SW.Bus.RabbitMqExtensions/QueueOptions.cs new file mode 100644 index 0000000..a0b8280 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/QueueOptions.cs @@ -0,0 +1,14 @@ +namespace SW.Bus.RabbitMqExtensions; + +public class QueueOptions +{ + public ushort? Prefetch { get; set; } + public int? RetryCount { get; set; } + public uint? RetryAfterSeconds { get; set; } + public int? Priority { get; set; } + public int? MaxPriority { get; set; } + public IDictionary? ConsumerArgs => Priority is null or 0 ? null : new Dictionary + { + { "x-priority", Priority}, + }; +} \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj new file mode 100644 index 0000000..dc59e3a --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj @@ -0,0 +1,13 @@ + + + + net8.0 + enable + enable + + + + + + + diff --git a/SW.Bus.SampleWeb/SW.Bus.SampleWeb.csproj b/SW.Bus.SampleWeb/SW.Bus.SampleWeb.csproj index c44169b..9626d52 100644 --- a/SW.Bus.SampleWeb/SW.Bus.SampleWeb.csproj +++ b/SW.Bus.SampleWeb/SW.Bus.SampleWeb.csproj @@ -7,7 +7,7 @@ - + diff --git a/SW.Bus.sln b/SW.Bus.sln index 5d61340..9737b00 100644 --- a/SW.Bus.sln +++ b/SW.Bus.sln @@ -14,6 +14,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution .editorconfig = .editorconfig EndProjectSection EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SW.Bus.RabbitMqExtensions", "SW.Bus.RabbitMqExtensions\SW.Bus.RabbitMqExtensions.csproj", "{4C85A34C-FBDA-4BBF-938D-86A46F9E5593}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -32,6 +34,10 @@ Global {2FB51BFB-82A8-40EF-8FEB-D9F12993F2E1}.Debug|Any CPU.Build.0 = Debug|Any CPU {2FB51BFB-82A8-40EF-8FEB-D9F12993F2E1}.Release|Any CPU.ActiveCfg = Release|Any CPU {2FB51BFB-82A8-40EF-8FEB-D9F12993F2E1}.Release|Any CPU.Build.0 = Release|Any CPU + {4C85A34C-FBDA-4BBF-938D-86A46F9E5593}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {4C85A34C-FBDA-4BBF-938D-86A46F9E5593}.Debug|Any CPU.Build.0 = Debug|Any CPU + {4C85A34C-FBDA-4BBF-938D-86A46F9E5593}.Release|Any CPU.ActiveCfg = Release|Any CPU + {4C85A34C-FBDA-4BBF-938D-86A46F9E5593}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/SW.Bus/BasicPublisher.cs b/SW.Bus/BasicPublisher.cs index f8f024c..e28f2b8 100644 --- a/SW.Bus/BasicPublisher.cs +++ b/SW.Bus/BasicPublisher.cs @@ -24,7 +24,7 @@ public BasicPublisher(IModel model, BusOptions busOptions, RequestContext reques this.requestContext = requestContext; } - async public Task Publish(TMessage message, string exchange) + async public Task Publish(TMessage message, string exchange, byte? priority = null) { var serializerOptions = new JsonSerializerOptions() { @@ -32,15 +32,15 @@ async public Task Publish(TMessage message, string exchange) }; var body = JsonSerializer.Serialize(message,message.GetType(), serializerOptions); - await Publish(message.GetType().Name, body,exchange); + await Publish(message.GetType().Name, body,exchange, priority); } - public async Task Publish(string messageTypeName, string message,string exchange) + public async Task Publish(string messageTypeName, string message,string exchange, byte? priority = null) { try { var body = Encoding.UTF8.GetBytes(message); - await Publish(messageTypeName, body,exchange); + await Publish(messageTypeName, body,exchange, priority); } catch (Exception e) { @@ -49,11 +49,15 @@ public async Task Publish(string messageTypeName, string message,string exchange } - public Task Publish(string messageTypeName, byte[] message,string exchange) + public Task Publish(string messageTypeName, byte[] message,string exchange, byte? priority = null) { IBasicProperties props = null; props = model.CreateBasicProperties(); props.Headers = new Dictionary(); + if (priority.HasValue) + { + props.Priority = priority.Value; + } if (requestContext.IsValid && busOptions.Token.IsValid) { var jwt = busOptions.Token.WriteJwt((ClaimsIdentity)requestContext.User.Identity); diff --git a/SW.Bus/BusOptions.cs b/SW.Bus/BusOptions.cs index 2650b38..2f53be8 100644 --- a/SW.Bus/BusOptions.cs +++ b/SW.Bus/BusOptions.cs @@ -1,6 +1,7 @@ using SW.HttpExtensions; using System; using System.Collections.Generic; +using SW.Bus.RabbitMqExtensions; namespace SW.Bus { @@ -34,6 +35,7 @@ public BusOptions(string environment) public ushort DefaultQueuePrefetch { get; set; } public ushort DefaultRetryCount { get; set; } public uint DefaultRetryAfter { get; set; } + public int DefaultMaxPriority { get; set; } public string NodeId { get; set; } public int ListenRetryCount { get; set; } public ushort ListenRetryAfter { get; set; } @@ -66,14 +68,15 @@ public BusOptions(string environment) public string NodeDeadLetterExchange => $"{versionPrefix}{environment}{(string.IsNullOrWhiteSpace(ApplicationName) ? "" : $".{ApplicationName}")}.node.deadletter".ToLower(); - public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null) + public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null, int? maxPriority = null) { Options[queueName.ToLower()] = new QueueOptions { Prefetch = prefetch, RetryCount = retryCount, RetryAfterSeconds = retryAfterSeconds, - Priority = priority + Priority = priority, + MaxPriority = maxPriority }; } diff --git a/SW.Bus/ConsumerDefinition.cs b/SW.Bus/ConsumerDefinition.cs index c66ec13..5d51bd6 100644 --- a/SW.Bus/ConsumerDefinition.cs +++ b/SW.Bus/ConsumerDefinition.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Reflection; +using SW.Bus.RabbitMqExtensions; namespace SW.Bus { @@ -10,12 +11,19 @@ public class ConsumerDefinition private readonly BusOptions busOptions; private readonly QueueOptions queueOptions; - public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string nakedQueueName) + public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string nakedQueueName, QueueOptions explicitOptions = null) { this.queueNamePrefix = queueNamePrefix; this.busOptions = busOptions; NakedQueueName = nakedQueueName; - busOptions.Options.TryGetValue(NakedQueueName, out queueOptions); + if (explicitOptions != null) + { + queueOptions = explicitOptions; + } + else + { + busOptions.Options.TryGetValue(NakedQueueName, out queueOptions); + } } @@ -26,11 +34,13 @@ public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string public int RetryCount => queueOptions?.RetryCount ?? busOptions.DefaultRetryCount; public uint RetryAfter => queueOptions?.RetryAfterSeconds ?? busOptions.DefaultRetryAfter; public ushort QueuePrefetch => queueOptions?.Prefetch ?? busOptions.DefaultQueuePrefetch; + public int MaxPriority => Math.Min(queueOptions?.MaxPriority ?? busOptions.DefaultMaxPriority, 5); public string NakedQueueName { get; private set; } - public string QueueName => $"{queueNamePrefix}.{NakedQueueName}".ToLower(); + public string QueueName => $"{queueNamePrefix}.{NakedQueueName}{(MaxPriority > 0 ? $".p{MaxPriority}" : "")}".ToLower(); + public string LegacyQueueName => $"{queueNamePrefix}.{NakedQueueName}".ToLower(); public string RoutingKey => MessageTypeName.ToLower(); public string RetryRoutingKey => $"{NakedQueueName}.retry".ToLower(); - public string RetryQueueName => $"{queueNamePrefix}.{NakedQueueName}.retry".ToLower(); + public string RetryQueueName => $"{queueNamePrefix}.{NakedQueueName}.retry{(MaxPriority > 0 ? $".p{MaxPriority}" : "")}".ToLower(); public string BadRoutingKey => $"{NakedQueueName}.bad".ToLower(); public string BadQueueName => $"{queueNamePrefix}.{NakedQueueName}.bad".ToLower(); public IDictionary RetryArgs => RetryCount == 0 ? null : new Dictionary @@ -40,12 +50,22 @@ public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string { "x-message-ttl", RetryAfter == 0 ? 100 : RetryAfter * 1000 } }; - public IDictionary ProcessArgs => new Dictionary + public IDictionary ProcessArgs { - { "x-dead-letter-exchange", busOptions.DeadLetterExchange }, - { "x-dead-letter-routing-key", RetryRoutingKey }, - - }; + get + { + var args = new Dictionary + { + { "x-dead-letter-exchange", busOptions.DeadLetterExchange }, + { "x-dead-letter-routing-key", RetryRoutingKey }, + }; + if (MaxPriority > 0) + { + args.Add("x-max-priority", MaxPriority); + } + return args; + } + } public static IDictionary BadArgs => new Dictionary { @@ -55,4 +75,3 @@ public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string } } - diff --git a/SW.Bus/ConsumerDiscovery.cs b/SW.Bus/ConsumerDiscovery.cs index cad6c2c..9761661 100644 --- a/SW.Bus/ConsumerDiscovery.cs +++ b/SW.Bus/ConsumerDiscovery.cs @@ -4,7 +4,9 @@ using System.Reflection; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using SW.PrimitiveTypes; +using SW.Bus.RabbitMqExtensions; namespace SW.Bus { @@ -12,11 +14,13 @@ public class ConsumerDiscovery { private readonly IServiceProvider sp; private readonly BusOptions busOptions; + private readonly ILogger logger; - public ConsumerDiscovery(IServiceProvider sp, BusOptions busOptions) + public ConsumerDiscovery(IServiceProvider sp, BusOptions busOptions, ILogger logger) { this.sp = sp; this.busOptions = busOptions; + this.logger = logger; } internal async Task> Load(bool consumersOnly = false) @@ -27,13 +31,39 @@ internal async Task> Load(bool consumersOnly = f using var scope = sp.CreateScope(); var consumers = scope.ServiceProvider.GetServices(); foreach (var svc in consumers) - foreach (var messageTypeName in await svc.GetMessageTypeNames()) + { + if (svc is IConsumeExtended consumeExtended) + { + var options = await consumeExtended.GetQueOptions(); + foreach (var kvp in options) + { + var messageTypeName = kvp.Key; + var extOptions = kvp.Value; + + if (extOptions.MaxPriority > 5) + { + logger.LogError($"MaxPriority for {messageTypeName} is {extOptions.MaxPriority}, which is greater than 5. It will be capped at 5."); + } - consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower()) + var queueOptions = extOptions; + + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower(), queueOptions) + { + ServiceType = svc.GetType(), + MessageTypeName = messageTypeName, + }); + } + } + else { - ServiceType = svc.GetType(), - MessageTypeName = messageTypeName, - }); + foreach (var messageTypeName in await svc.GetMessageTypeNames()) + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower()) + { + ServiceType = svc.GetType(), + MessageTypeName = messageTypeName, + }); + } + } if (consumersOnly) return consumerDefinitions; diff --git a/SW.Bus/ConsumerRunner.cs b/SW.Bus/ConsumerRunner.cs index 87893b2..ef45774 100644 --- a/SW.Bus/ConsumerRunner.cs +++ b/SW.Bus/ConsumerRunner.cs @@ -172,7 +172,7 @@ private async Task RunOnFail(object svc, MethodInfo failMethod, Exception ex, st { await (Task)failMethod.Invoke(svc, new Object[] { ex }); } - catch (Exception e) + catch (Exception) { logger.LogError(ex, $"Failed to run OnFail message, Message {message}"); } diff --git a/SW.Bus/ConsumersService.cs b/SW.Bus/ConsumersService.cs index 3b35d53..29acce8 100644 --- a/SW.Bus/ConsumersService.cs +++ b/SW.Bus/ConsumersService.cs @@ -3,6 +3,7 @@ using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -18,6 +19,7 @@ internal class ConsumersService : IHostedService private readonly ConsumerDiscovery consumerDiscovery; private readonly ConnectionFactory connectionFactory; private readonly IDictionary openModels; + private readonly ConcurrentDictionary drainingModels = new ConcurrentDictionary(); private readonly ConsumerRunner consumerRunner; private IConnection conn; @@ -38,6 +40,7 @@ public ConsumersService(ILogger logger, BusOptions busOptions, public Task StartAsync(CancellationToken cancellationToken) { Task.Run(() => StartBusAsync(cancellationToken), cancellationToken); + Task.Run(() => DrainingLoop(cancellationToken), cancellationToken); return Task.CompletedTask; } @@ -78,6 +81,8 @@ private void DeclareAndBind(IModel model, ConsumerDefinition c) { logger.LogInformation($"Declaring and binding: {c.QueueName}."); + CheckAndMigrateLegacyQueue(c); + // process queue model.QueueDeclare(c.QueueName, true, false, false, c.ProcessArgs); model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RoutingKey, null); @@ -209,9 +214,147 @@ private void ConnectionShutdown(object connection, ShutdownEventArgs args) } } + private void CheckAndMigrateLegacyQueue(ConsumerDefinition c) + { + var potentialLegacyQueues = new List(); + potentialLegacyQueues.Add(c.LegacyQueueName); + for (int i = 1; i <= 10; i++) + { + potentialLegacyQueues.Add($"{c.LegacyQueueName}.p{i}"); + } + if (busOptions.DefaultMaxPriority > 10) + { + potentialLegacyQueues.Add($"{c.LegacyQueueName}.p{busOptions.DefaultMaxPriority}"); + } + + foreach (var legacyQueueName in potentialLegacyQueues) + { + if (legacyQueueName == c.QueueName) continue; + if (drainingModels.ContainsKey(legacyQueueName)) continue; + try + { + using (var tempModel = conn.CreateModel()) + { + tempModel.QueueDeclarePassive(legacyQueueName); + } + + logger.LogInformation($"Legacy queue found: {legacyQueueName}. Starting migration."); + + using (var tempModel = conn.CreateModel()) + { + tempModel.QueueUnbind(legacyQueueName, busOptions.ProcessExchange, c.RoutingKey, null); + tempModel.QueueUnbind(legacyQueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null); + } + + if (openModels.ContainsKey(legacyQueueName)) + { + var model = openModels[legacyQueueName]; + openModels.Remove(legacyQueueName); + drainingModels.TryAdd(legacyQueueName, model); + logger.LogInformation($"Moved active consumer on {legacyQueueName} to draining mode."); + } + else + { + StartDraining(legacyQueueName, c); + } + } + catch (RabbitMQ.Client.Exceptions.OperationInterruptedException ex) + { + if (ex.ShutdownReason.ReplyCode != 404) + { + logger.LogError(ex, $"Error checking legacy queue {legacyQueueName}"); + } + } + catch (Exception ex) + { + logger.LogError(ex, $"Error checking legacy queue {legacyQueueName}"); + } + } + } - public async Task StopAsync(CancellationToken cancellationToken) + private void StartDraining(string queueName, ConsumerDefinition c) + { + var model = conn.CreateModel(); + if (drainingModels.TryAdd(queueName, model)) + { + var consumer = new AsyncEventingBasicConsumer(model); + consumer.Received += async (ch, ea) => + { + await consumerRunner.Run(ea, c, model); + }; + + var args = new Dictionary { { "x-priority", 1 } }; + model.BasicConsume(queueName, false, "", args, consumer); + + logger.LogInformation($"Started draining legacy queue: {queueName}"); + } + else + { + model.Dispose(); + } + } + + private async Task DrainingLoop(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken); + + if (drainingModels.IsEmpty || conn == null || !conn.IsOpen) continue; + + using (var model = conn.CreateModel()) + { + var queuesToRemove = new List(); + foreach (var queueName in drainingModels.Keys) + { + try + { + var result = model.QueueDeclarePassive(queueName); + if (result.MessageCount == 0) + { + model.QueueDelete(queueName); + queuesToRemove.Add(queueName); + logger.LogInformation($"Legacy queue {queueName} is empty and has been deleted."); + } + } + catch (Exception ex) + { + logger.LogWarning(ex, $"Error checking draining queue {queueName}"); + if (ex is RabbitMQ.Client.Exceptions.OperationInterruptedException oex && oex.ShutdownReason.ReplyCode == 404) + { + queuesToRemove.Add(queueName); + } + } + } + + foreach(var q in queuesToRemove) + { + if (drainingModels.TryRemove(q, out var drainingModel)) + { + try + { + drainingModel.Close(); + drainingModel.Dispose(); + } + catch + { + // ignored + } + } + } + } + } + catch (Exception ex) + { + logger.LogError(ex, "Error in DrainingLoop"); + } + } + } + + public Task StopAsync(CancellationToken cancellationToken) { foreach (var model in openModels.Values) @@ -226,9 +369,22 @@ public async Task StopAsync(CancellationToken cancellationToken) logger.LogWarning(ex, $"Failed to stop model."); } + foreach (var model in drainingModels.Values) + { + try + { + model.Dispose(); + } + catch + { + // ignored + } + } + nodeModel?.Dispose(); conn?.Close(); conn?.Dispose(); + return Task.CompletedTask; } } } diff --git a/SW.Bus/Publisher.cs b/SW.Bus/Publisher.cs index aefbc93..005d3f7 100644 --- a/SW.Bus/Publisher.cs +++ b/SW.Bus/Publisher.cs @@ -1,15 +1,11 @@ - -using RabbitMQ.Client; -using SW.HttpExtensions; +using RabbitMQ.Client; using SW.PrimitiveTypes; -using System.Collections.Generic; -using System.Security.Claims; -using System.Text; using System.Threading.Tasks; +using SW.Bus.RabbitMqExtensions; namespace SW.Bus { - internal class Publisher : IPublish + internal class Publisher : IRabbitMqPublish { private readonly BasicPublisher basicPublisher; private readonly string exchange; @@ -25,5 +21,13 @@ public Task Publish(string messageTypeName, string message) => public Task Publish(string messageTypeName, byte[] message) => basicPublisher.Publish(messageTypeName, message, exchange); + public Task Publish(TMessage message, byte priority) => + basicPublisher.Publish(message, exchange, priority); + + public Task Publish(string messageTypeName, string message, byte priority) => + basicPublisher.Publish(messageTypeName, message, exchange, priority); + + public Task Publish(string messageTypeName, byte[] message, byte priority) => + basicPublisher.Publish(messageTypeName, message, exchange, priority); } } \ No newline at end of file diff --git a/SW.Bus/QueueOptions.cs b/SW.Bus/QueueOptions.cs deleted file mode 100644 index 17db832..0000000 --- a/SW.Bus/QueueOptions.cs +++ /dev/null @@ -1,16 +0,0 @@ -using System.Collections.Generic; - -namespace SW.Bus -{ - public class QueueOptions - { - public ushort? Prefetch { get; set; } - public int? RetryCount { get; set; } - public uint? RetryAfterSeconds { get; set; } - public int? Priority { get; set; } - public IDictionary ConsumerArgs => Priority is null or 0 ? null : new Dictionary - { - { "x-priority", Priority}, - }; - } -} \ No newline at end of file diff --git a/SW.Bus/SW.Bus.csproj b/SW.Bus/SW.Bus.csproj index 41e22cd..d922c45 100644 --- a/SW.Bus/SW.Bus.csproj +++ b/SW.Bus/SW.Bus.csproj @@ -25,14 +25,14 @@ - + + + + + - - True - \ - True From 3da8c9a5586a3cf0b37e2a912c4374c8c71c878e Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Tue, 30 Dec 2025 10:29:35 +0300 Subject: [PATCH 2/4] refactor: simplify consumer options and queue handling --- SW.Bus.RabbitMqExtensions/ConsumerOptions.cs | 7 + SW.Bus.RabbitMqExtensions/IConsumeExtended.cs | 2 +- SW.Bus.RabbitMqExtensions/QueueOptions.cs | 5 +- .../SW.Bus.RabbitMqExtensions.csproj | 14 + SW.Bus/BusOptions.cs | 5 +- SW.Bus/ConsumerDefinition.cs | 120 ++--- SW.Bus/ConsumerDiscovery.cs | 149 +++--- SW.Bus/ConsumersService.cs | 503 +++++++----------- 8 files changed, 331 insertions(+), 474 deletions(-) create mode 100644 SW.Bus.RabbitMqExtensions/ConsumerOptions.cs diff --git a/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs b/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs new file mode 100644 index 0000000..9e21440 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/ConsumerOptions.cs @@ -0,0 +1,7 @@ +namespace SW.Bus.RabbitMqExtensions; + +public class ConsumerOptions +{ + public ushort? Prefetch { get; set; } + public int? Priority { get; set; } +} \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs b/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs index fbfe4a3..3ff43a5 100644 --- a/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs +++ b/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs @@ -4,5 +4,5 @@ namespace SW.Bus.RabbitMqExtensions; public interface IConsumeExtended : IConsume { - Task> GetQueOptions(); + Task> GetMessageTypeNamesWithOptions(); } \ No newline at end of file diff --git a/SW.Bus.RabbitMqExtensions/QueueOptions.cs b/SW.Bus.RabbitMqExtensions/QueueOptions.cs index a0b8280..4da32d3 100644 --- a/SW.Bus.RabbitMqExtensions/QueueOptions.cs +++ b/SW.Bus.RabbitMqExtensions/QueueOptions.cs @@ -1,12 +1,9 @@ namespace SW.Bus.RabbitMqExtensions; -public class QueueOptions +public class QueueOptions:ConsumerOptions { - public ushort? Prefetch { get; set; } public int? RetryCount { get; set; } public uint? RetryAfterSeconds { get; set; } - public int? Priority { get; set; } - public int? MaxPriority { get; set; } public IDictionary? ConsumerArgs => Priority is null or 0 ? null : new Dictionary { { "x-priority", Priority}, diff --git a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj index dc59e3a..4adfda3 100644 --- a/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj +++ b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj @@ -4,6 +4,20 @@ net8.0 enable enable + SimplyWorks.Bus.RabbitMqExtensions + SimplyWorks.Bus.RabbitMqExtensions + Simplify9 + Extensions for SimplyWorks.Bus library to support RabbitMQ specific features. + messagebus;rabbitmq;aspnetcore;messaging;pubsub;eventdriven;microservices;dotnet8 + MIT + https://github.com/simplify9/SW-Bus + https://github.com/simplify9/SW-Bus + README.md + icon.png + git + Copyright © 2020 Simplify9 + See https://github.com/simplify9/SW-Bus/releases for release notes and changelog. + diff --git a/SW.Bus/BusOptions.cs b/SW.Bus/BusOptions.cs index 2f53be8..a8bf90d 100644 --- a/SW.Bus/BusOptions.cs +++ b/SW.Bus/BusOptions.cs @@ -68,15 +68,14 @@ public BusOptions(string environment) public string NodeDeadLetterExchange => $"{versionPrefix}{environment}{(string.IsNullOrWhiteSpace(ApplicationName) ? "" : $".{ApplicationName}")}.node.deadletter".ToLower(); - public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null, int? maxPriority = null) + public void AddQueueOption(string queueName, ushort? prefetch = null, int? retryCount = null, uint? retryAfterSeconds = null,int? priority = null) { Options[queueName.ToLower()] = new QueueOptions { Prefetch = prefetch, RetryCount = retryCount, RetryAfterSeconds = retryAfterSeconds, - Priority = priority, - MaxPriority = maxPriority + Priority = priority }; } diff --git a/SW.Bus/ConsumerDefinition.cs b/SW.Bus/ConsumerDefinition.cs index 5d51bd6..8adaea3 100644 --- a/SW.Bus/ConsumerDefinition.cs +++ b/SW.Bus/ConsumerDefinition.cs @@ -1,77 +1,73 @@ using System; using System.Collections.Generic; using System.Reflection; +using RabbitMQ.Client.Events; using SW.Bus.RabbitMqExtensions; -namespace SW.Bus +namespace SW.Bus; + +public class ConsumerDefinition { - public class ConsumerDefinition + private readonly string queueNamePrefix; + private readonly BusOptions busOptions; + private readonly QueueOptions queueOptions; + + public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string nakedQueueName) { - private readonly string queueNamePrefix; - private readonly BusOptions busOptions; - private readonly QueueOptions queueOptions; + this.queueNamePrefix = queueNamePrefix; + this.busOptions = busOptions; + NakedQueueName = nakedQueueName; + busOptions.Options.TryGetValue(NakedQueueName, out queueOptions); + } - public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, string nakedQueueName, QueueOptions explicitOptions = null) - { - this.queueNamePrefix = queueNamePrefix; - this.busOptions = busOptions; - NakedQueueName = nakedQueueName; - if (explicitOptions != null) - { - queueOptions = explicitOptions; - } - else - { - busOptions.Options.TryGetValue(NakedQueueName, out queueOptions); - } - - } + public ConsumerDefinition(string queueNamePrefix, BusOptions busOptions, ConsumerOptions consumerOptions, + string nakedQueueName): this(queueNamePrefix, busOptions, nakedQueueName) + { + ArgumentNullException.ThrowIfNull(consumerOptions); + queueOptions ??= new QueueOptions(); + if (consumerOptions.Prefetch.HasValue) + queueOptions.Prefetch = consumerOptions.Prefetch; + if (consumerOptions.Priority.HasValue) + queueOptions.Priority = consumerOptions.Priority; + } + + public Type ServiceType { get; set; } + public Type MessageType { get; set; } + public string MessageTypeName { get; set; } + public MethodInfo Method { get; set; } + public int RetryCount => queueOptions?.RetryCount ?? busOptions.DefaultRetryCount; + public uint RetryAfter => queueOptions?.RetryAfterSeconds ?? busOptions.DefaultRetryAfter; + public ushort QueuePrefetch => queueOptions?.Prefetch ?? busOptions.DefaultQueuePrefetch; + public string NakedQueueName { get; private set; } + public string QueueName => $"{queueNamePrefix}.{NakedQueueName}".ToLower(); + public string RoutingKey => MessageTypeName.ToLower(); + public string RetryRoutingKey => $"{NakedQueueName}.retry".ToLower(); + public string RetryQueueName => $"{queueNamePrefix}.{NakedQueueName}.retry".ToLower(); + public string BadRoutingKey => $"{NakedQueueName}.bad".ToLower(); + public string BadQueueName => $"{queueNamePrefix}.{NakedQueueName}.bad".ToLower(); - public Type ServiceType { get; set; } - public Type MessageType { get; set; } - public string MessageTypeName { get; set; } - public MethodInfo Method { get; set; } - public int RetryCount => queueOptions?.RetryCount ?? busOptions.DefaultRetryCount; - public uint RetryAfter => queueOptions?.RetryAfterSeconds ?? busOptions.DefaultRetryAfter; - public ushort QueuePrefetch => queueOptions?.Prefetch ?? busOptions.DefaultQueuePrefetch; - public int MaxPriority => Math.Min(queueOptions?.MaxPriority ?? busOptions.DefaultMaxPriority, 5); - public string NakedQueueName { get; private set; } - public string QueueName => $"{queueNamePrefix}.{NakedQueueName}{(MaxPriority > 0 ? $".p{MaxPriority}" : "")}".ToLower(); - public string LegacyQueueName => $"{queueNamePrefix}.{NakedQueueName}".ToLower(); - public string RoutingKey => MessageTypeName.ToLower(); - public string RetryRoutingKey => $"{NakedQueueName}.retry".ToLower(); - public string RetryQueueName => $"{queueNamePrefix}.{NakedQueueName}.retry{(MaxPriority > 0 ? $".p{MaxPriority}" : "")}".ToLower(); - public string BadRoutingKey => $"{NakedQueueName}.bad".ToLower(); - public string BadQueueName => $"{queueNamePrefix}.{NakedQueueName}.bad".ToLower(); - public IDictionary RetryArgs => RetryCount == 0 ? null : new Dictionary + public IDictionary RetryArgs => RetryCount == 0 + ? null + : new Dictionary { - { "x-dead-letter-exchange", busOptions.ProcessExchange}, - { "x-dead-letter-routing-key", RetryRoutingKey}, + { "x-dead-letter-exchange", busOptions.ProcessExchange }, + { "x-dead-letter-routing-key", RetryRoutingKey }, { "x-message-ttl", RetryAfter == 0 ? 100 : RetryAfter * 1000 } }; - public IDictionary ProcessArgs - { - get - { - var args = new Dictionary - { - { "x-dead-letter-exchange", busOptions.DeadLetterExchange }, - { "x-dead-letter-routing-key", RetryRoutingKey }, - }; - if (MaxPriority > 0) - { - args.Add("x-max-priority", MaxPriority); - } - return args; - } - } + public IDictionary ProcessArgs => new Dictionary + { + { "x-dead-letter-exchange", busOptions.DeadLetterExchange }, + { "x-dead-letter-routing-key", RetryRoutingKey }, + }; - public static IDictionary BadArgs => new Dictionary - { - { "x-message-ttl", (uint)TimeSpan.FromDays(7).TotalMilliseconds } - }; - public IDictionary ConsumerArgs => queueOptions?.ConsumerArgs; - } + public static IDictionary BadArgs => new Dictionary + { + { "x-message-ttl", (uint)TimeSpan.FromDays(7).TotalMilliseconds } + }; -} + public IDictionary ConsumerArgs => queueOptions?.ConsumerArgs; + public int? ConsumerPriority => queueOptions?.Priority; + public string ConsumerTag { get; set; } + public AsyncEventingBasicConsumer ConsumerObject { get; set; } +} \ No newline at end of file diff --git a/SW.Bus/ConsumerDiscovery.cs b/SW.Bus/ConsumerDiscovery.cs index 9761661..ac5f8a2 100644 --- a/SW.Bus/ConsumerDiscovery.cs +++ b/SW.Bus/ConsumerDiscovery.cs @@ -4,103 +4,86 @@ using System.Reflection; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using SW.PrimitiveTypes; using SW.Bus.RabbitMqExtensions; +using SW.PrimitiveTypes; -namespace SW.Bus +namespace SW.Bus; + +public class ConsumerDiscovery(IServiceProvider sp, BusOptions busOptions) { - public class ConsumerDiscovery + internal async Task> Load(bool consumersOnly = false) { - private readonly IServiceProvider sp; - private readonly BusOptions busOptions; - private readonly ILogger logger; - - public ConsumerDiscovery(IServiceProvider sp, BusOptions busOptions, ILogger logger) - { - this.sp = sp; - this.busOptions = busOptions; - this.logger = logger; - } + var consumerDefinitions = new List(); + var queueNamePrefix = + $"{busOptions.ProcessExchange}{(string.IsNullOrWhiteSpace(busOptions.ApplicationName) ? "" : $".{busOptions.ApplicationName}")}"; - internal async Task> Load(bool consumersOnly = false) + using var scope = sp.CreateScope(); + var consumers = scope.ServiceProvider.GetServices(); + foreach (var svc in consumers) { - var consumerDefinitions = new List(); - var queueNamePrefix = $"{busOptions.ProcessExchange}{(string.IsNullOrWhiteSpace(busOptions.ApplicationName) ? "" : $".{busOptions.ApplicationName}")}"; - - using var scope = sp.CreateScope(); - var consumers = scope.ServiceProvider.GetServices(); - foreach (var svc in consumers) + if (svc is IConsumeExtended extendedSvc) { - if (svc is IConsumeExtended consumeExtended) + var messageTypesWithOptions = await extendedSvc.GetMessageTypeNamesWithOptions(); + foreach (var kvp in messageTypesWithOptions) { - var options = await consumeExtended.GetQueOptions(); - foreach (var kvp in options) + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, kvp.Value, + $"{svc.GetType().Name}.{kvp.Key}".ToLower()) { - var messageTypeName = kvp.Key; - var extOptions = kvp.Value; - - if (extOptions.MaxPriority > 5) - { - logger.LogError($"MaxPriority for {messageTypeName} is {extOptions.MaxPriority}, which is greater than 5. It will be capped at 5."); - } - - var queueOptions = extOptions; - - consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower(), queueOptions) - { - ServiceType = svc.GetType(), - MessageTypeName = messageTypeName, - }); - } - } - else - { - foreach (var messageTypeName in await svc.GetMessageTypeNames()) - consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower()) - { - ServiceType = svc.GetType(), - MessageTypeName = messageTypeName, - }); + ServiceType = svc.GetType(), + MessageTypeName = kvp.Key, + }); } } + else + { + foreach (var messageTypeName in await svc.GetMessageTypeNames()) - if (consumersOnly) - return consumerDefinitions; - var genericConsumers = scope.ServiceProvider.GetServices(); - foreach (var svc in genericConsumers) - foreach (var type in svc.GetType().GetTypeInfo().ImplementedInterfaces.Where(t => t.IsGenericType && t.GetGenericTypeDefinition() == typeof(IConsume<>))) - - consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{type.GetGenericArguments()[0].Name}".ToLower()) - { - ServiceType = svc.GetType(), - MessageType = type.GetGenericArguments()[0], - MessageTypeName = type.GetGenericArguments()[0].Name, - Method = type.GetMethod("Process"), - }); - - return consumerDefinitions; + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, + $"{svc.GetType().Name}.{messageTypeName}".ToLower()) + { + ServiceType = svc.GetType(), + MessageTypeName = messageTypeName, + }); + } } - - internal ICollection LoadListeners() - { - var consumerDefinitions = new List(); - using var scope = sp.CreateScope(); - - var genericConsumers = scope.ServiceProvider.GetServices(); - foreach (var svc in genericConsumers) - foreach (var type in svc.GetType().GetTypeInfo().ImplementedInterfaces.Where(t => t.IsGenericType && t.GetGenericTypeDefinition() == typeof(IListen<>))) - consumerDefinitions.Add(new ListenerDefinition - { - ServiceType = svc.GetType(), - MessageType = type.GetGenericArguments()[0], - MessageTypeName = type.GetGenericArguments()[0].Name, - Method = type.GetMethod("Process"), - FailMethod= type.GetMethod("OnFail") - }); + if (consumersOnly) return consumerDefinitions; - } + var genericConsumers = scope.ServiceProvider.GetServices(); + foreach (var svc in genericConsumers) + foreach (var type in svc.GetType().GetTypeInfo().ImplementedInterfaces + .Where(t => t.IsGenericType && t.GetGenericTypeDefinition() == typeof(IConsume<>))) + + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, + $"{svc.GetType().Name}.{type.GetGenericArguments()[0].Name}".ToLower()) + { + ServiceType = svc.GetType(), + MessageType = type.GetGenericArguments()[0], + MessageTypeName = type.GetGenericArguments()[0].Name, + Method = type.GetMethod("Process"), + }); + + return consumerDefinitions; + } + + internal ICollection LoadListeners() + { + var consumerDefinitions = new List(); + using var scope = sp.CreateScope(); + + var genericConsumers = scope.ServiceProvider.GetServices(); + foreach (var svc in genericConsumers) + foreach (var type in svc.GetType().GetTypeInfo().ImplementedInterfaces + .Where(t => t.IsGenericType && t.GetGenericTypeDefinition() == typeof(IListen<>))) + consumerDefinitions.Add(new ListenerDefinition + { + ServiceType = svc.GetType(), + MessageType = type.GetGenericArguments()[0], + MessageTypeName = type.GetGenericArguments()[0].Name, + Method = type.GetMethod("Process"), + FailMethod = type.GetMethod("OnFail") + }); + return consumerDefinitions; } -} +} \ No newline at end of file diff --git a/SW.Bus/ConsumersService.cs b/SW.Bus/ConsumersService.cs index 29acce8..ebf4209 100644 --- a/SW.Bus/ConsumersService.cs +++ b/SW.Bus/ConsumersService.cs @@ -3,389 +3,250 @@ using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; -namespace SW.Bus +namespace SW.Bus; + +internal class ConsumersService : IHostedService { - internal class ConsumersService : IHostedService + private readonly ILogger logger; + private readonly BusOptions busOptions; + private readonly ConsumerDiscovery consumerDiscovery; + private readonly ConnectionFactory connectionFactory; + private readonly IDictionary openModels; + private readonly ConsumerRunner consumerRunner; + + private IConnection conn; + private IModel nodeModel; + private ICollection consumerDefinitions; + + public ConsumersService(ILogger logger, BusOptions busOptions, + ConsumerDiscovery consumerDiscovery, ConnectionFactory connectionFactory, ConsumerRunner consumerRunner) { + this.logger = logger; + this.busOptions = busOptions; + this.consumerDiscovery = consumerDiscovery; + this.connectionFactory = connectionFactory; + this.consumerRunner = consumerRunner; - private readonly ILogger logger; - private readonly BusOptions busOptions; - private readonly ConsumerDiscovery consumerDiscovery; - private readonly ConnectionFactory connectionFactory; - private readonly IDictionary openModels; - private readonly ConcurrentDictionary drainingModels = new ConcurrentDictionary(); - private readonly ConsumerRunner consumerRunner; - - private IConnection conn; - private IModel nodeModel; - private ICollection consumerDefinitions; - public ConsumersService(ILogger logger, BusOptions busOptions, - ConsumerDiscovery consumerDiscovery, ConnectionFactory connectionFactory, ConsumerRunner consumerRunner) - { - this.logger = logger; - this.busOptions = busOptions; - this.consumerDiscovery = consumerDiscovery; - this.connectionFactory = connectionFactory; - this.consumerRunner = consumerRunner; + openModels = new Dictionary(); + } - openModels = new Dictionary(); - } + public Task StartAsync(CancellationToken cancellationToken) + { + Task.Run(() => StartBusAsync(cancellationToken), cancellationToken); + return Task.CompletedTask; + } - public Task StartAsync(CancellationToken cancellationToken) + private async Task StartBusAsync(CancellationToken cancellationToken) + { + try { - Task.Run(() => StartBusAsync(cancellationToken), cancellationToken); - Task.Run(() => DrainingLoop(cancellationToken), cancellationToken); - return Task.CompletedTask; + consumerDefinitions = await consumerDiscovery.Load(); - } + conn = connectionFactory.CreateConnection(); + conn.ConnectionShutdown += ConnectionShutdown; + DeclareAndBindListener(); - private async Task StartBusAsync(CancellationToken cancellationToken) - { - - try + using (var model = conn.CreateModel()) { - consumerDefinitions = await consumerDiscovery.Load(); - - conn = connectionFactory.CreateConnection(); - conn.ConnectionShutdown += ConnectionShutdown; - DeclareAndBindListener(); - - using (var model = conn.CreateModel()) - { - foreach (var c in consumerDefinitions) - DeclareAndBind(model,c); - - } - - foreach (var consumerDefinition in consumerDefinitions) - { - AttachConsumer(consumerDefinition); - } - - + foreach (var c in consumerDefinitions) + DeclareAndBind(model, c); } - catch (Exception ex) + + foreach (var consumerDefinition in consumerDefinitions) { - logger.LogError(ex, $"Starting {nameof(ConsumersService)}"); + AttachConsumer(consumerDefinition); } - } - - private void DeclareAndBind(IModel model, ConsumerDefinition c) + catch (Exception ex) { - logger.LogInformation($"Declaring and binding: {c.QueueName}."); - - CheckAndMigrateLegacyQueue(c); - - // process queue - model.QueueDeclare(c.QueueName, true, false, false, c.ProcessArgs); - model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RoutingKey, null); - model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null); - //model.QueueUnbind(); - // wait queue - - model.QueueDeclare(c.RetryQueueName, true, false, false, c.RetryArgs); - model.QueueBind(c.RetryQueueName, busOptions.DeadLetterExchange, c.RetryRoutingKey, null); - // bad queue - model.QueueDeclare(c.BadQueueName, true, false, false, ConsumerDefinition.BadArgs); - model.QueueBind(c.BadQueueName, busOptions.DeadLetterExchange, c.BadRoutingKey, null); + logger.LogError(ex, $"Starting {nameof(ConsumersService)}"); } + } - private void DeclareAndBindListener() - { - logger.LogInformation($"Declaring and binding node queue: {busOptions.NodeQueueName}."); - var listeners = consumerDiscovery.LoadListeners(); - - var repeated = listeners.GroupBy(nc => nc.MessageType).Select(grp => new - { - Count = grp.Count(), - MessageType = grp.Key - }).Where(mc=> mc.Count > 1).ToArray(); - - if (repeated.Any()) - throw new BusException("One node consumer is allowed for each message type. the following message(s) has more than one node consumer defined" + - $" {string.Join(',', repeated.Select(r=> r.MessageType.FullName))}"); - - nodeModel = conn.CreateModel(); - // process queue - nodeModel.QueueDeclare(busOptions.NodeQueueName, true, true, true, busOptions.NodeProcessArgs ); - nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRoutingKey, null); - nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRetryRoutingKey, null); - // wait queue - nodeModel.QueueDeclare(busOptions.NodeRetryQueueName, true, true, true, busOptions.NodeRetryArgs); - nodeModel.QueueBind(busOptions.NodeRetryQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeRetryRoutingKey, null); - // bad queue - nodeModel.QueueDeclare(busOptions.NodeBadQueueName, true, false, false, ConsumerDefinition.BadArgs); - nodeModel.QueueBind(busOptions.NodeBadQueueName, busOptions.NodeDeadLetterExchange, busOptions.NodeBadRoutingKey, null); - - var consumer = new AsyncEventingBasicConsumer(nodeModel); - consumer.Shutdown += (ch, args) => - { - try - { - logger.LogWarning($"Node Consumer RabbitMq connection shutdown. {args}"); - } - catch (Exception) - { - // ignored - } - return Task.CompletedTask; - }; - consumer.Received += async (ch, ea) => - { - await consumerRunner.RunNodeMessage(ea, nodeModel,listeners,RefreshConsumers ); - }; - - nodeModel.BasicQos(0, 1, false); + private void DeclareAndBind(IModel model, ConsumerDefinition c) + { + logger.LogInformation($"Declaring and binding: {c.QueueName}."); + + // process queue + model.QueueDeclare(c.QueueName, true, false, false, c.ProcessArgs); + model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RoutingKey, null); + model.QueueBind(c.QueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null); + //model.QueueUnbind(); + // wait queue + + model.QueueDeclare(c.RetryQueueName, true, false, false, c.RetryArgs); + model.QueueBind(c.RetryQueueName, busOptions.DeadLetterExchange, c.RetryRoutingKey, null); + // bad queue + model.QueueDeclare(c.BadQueueName, true, false, false, ConsumerDefinition.BadArgs); + model.QueueBind(c.BadQueueName, busOptions.DeadLetterExchange, c.BadRoutingKey, null); + } - nodeModel.BasicConsume(busOptions.NodeQueueName, false, consumer); + private void DeclareAndBindListener() + { + logger.LogInformation($"Declaring and binding node queue: {busOptions.NodeQueueName}."); + var listeners = consumerDiscovery.LoadListeners(); - } - private void AttachConsumer(ConsumerDefinition consumerDefinition) + var repeated = listeners.GroupBy(nc => nc.MessageType).Select(grp => new { - var model = conn.CreateModel(); - openModels.Add(consumerDefinition.QueueName, model); - - var consumer = new AsyncEventingBasicConsumer(model); - consumer.Shutdown += (ch, args) => - { - try - { - logger.LogWarning($"Consumer RabbitMq connection shutdown. {args}"); - } - catch (Exception) - { - // ignored - } - return Task.CompletedTask; - }; - consumer.Received += async (ch, ea) => - { - await consumerRunner.Run(ea, consumerDefinition, model); - }; - - model.BasicQos(0, consumerDefinition.QueuePrefetch, false); - - model.BasicConsume(consumerDefinition.QueueName, false, "", consumerDefinition.ConsumerArgs,consumer ); - - } - - private async Task RefreshConsumers() + Count = grp.Count(), + MessageType = grp.Key + }).Where(mc => mc.Count > 1).ToArray(); + + if (repeated.Any()) + throw new BusException( + "One node consumer is allowed for each message type. the following message(s) has more than one node consumer defined" + + $" {string.Join(',', repeated.Select(r => r.MessageType.FullName))}"); + + nodeModel = conn.CreateModel(); + // process queue + nodeModel.QueueDeclare(busOptions.NodeQueueName, true, true, true, busOptions.NodeProcessArgs); + nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRoutingKey, null); + nodeModel.QueueBind(busOptions.NodeQueueName, busOptions.NodeExchange, busOptions.NodeRetryRoutingKey, null); + // wait queue + nodeModel.QueueDeclare(busOptions.NodeRetryQueueName, true, true, true, busOptions.NodeRetryArgs); + nodeModel.QueueBind(busOptions.NodeRetryQueueName, busOptions.NodeDeadLetterExchange, + busOptions.NodeRetryRoutingKey, null); + // bad queue + nodeModel.QueueDeclare(busOptions.NodeBadQueueName, true, false, false, ConsumerDefinition.BadArgs); + nodeModel.QueueBind(busOptions.NodeBadQueueName, busOptions.NodeDeadLetterExchange, + busOptions.NodeBadRoutingKey, null); + + var consumer = new AsyncEventingBasicConsumer(nodeModel); + consumer.Shutdown += (ch, args) => { try { - consumerDefinitions = await consumerDiscovery.Load(true); - using (var model = conn.CreateModel()) - { - foreach (var c in consumerDefinitions) - { - if(openModels.ContainsKey(c.QueueName)) - continue; - DeclareAndBind(model,c); - } - } - foreach (var consumerDefinition in consumerDefinitions) - { - if(openModels.ContainsKey(consumerDefinition.QueueName)) - continue; - AttachConsumer(consumerDefinition); - } + logger.LogWarning($"Node Consumer RabbitMq connection shutdown. {args}"); } - catch (Exception ex) + catch (Exception) { - logger.LogError(ex, $"Starting {nameof(ConsumersService)}"); + // ignored } - } - private void ConnectionShutdown(object connection, ShutdownEventArgs args) + + return Task.CompletedTask; + }; + consumer.Received += async (ch, ea) => + { + await consumerRunner.RunNodeMessage(ea, nodeModel, listeners, RefreshConsumers); + }; + + nodeModel.BasicQos(0, 1, false); + + nodeModel.BasicConsume(busOptions.NodeQueueName, false, consumer); + } + + private void AttachConsumer(ConsumerDefinition consumerDefinition) + { + var model = conn.CreateModel(); + openModels.Add(consumerDefinition.QueueName, (model, consumerDefinition)); + + var consumer = new AsyncEventingBasicConsumer(model); + consumerDefinition.ConsumerObject = consumer; + consumer.Shutdown += (ch, args) => { try { - logger.LogWarning($"Consumer RabbitMq connection shutdown. {args.Cause}"); + logger.LogWarning($"Consumer RabbitMq connection shutdown. {args}"); } catch (Exception) { // ignored } - } - private void CheckAndMigrateLegacyQueue(ConsumerDefinition c) + return Task.CompletedTask; + }; + consumer.Received += async (ch, ea) => { await consumerRunner.Run(ea, consumerDefinition, model); }; + + model.BasicQos(0, consumerDefinition.QueuePrefetch, false); + + consumerDefinition.ConsumerTag = model.BasicConsume(consumerDefinition.QueueName, false, "", + consumerDefinition.ConsumerArgs, consumer); + } + + private async Task RefreshConsumers() + { + try { - var potentialLegacyQueues = new List(); - potentialLegacyQueues.Add(c.LegacyQueueName); - for (int i = 1; i <= 10; i++) - { - potentialLegacyQueues.Add($"{c.LegacyQueueName}.p{i}"); - } - if (busOptions.DefaultMaxPriority > 10) + consumerDefinitions = await consumerDiscovery.Load(true); + using (var model = conn.CreateModel()) { - potentialLegacyQueues.Add($"{c.LegacyQueueName}.p{busOptions.DefaultMaxPriority}"); + foreach (var c in consumerDefinitions) + { + if (openModels.ContainsKey(c.QueueName)) + continue; + DeclareAndBind(model, c); + } } - foreach (var legacyQueueName in potentialLegacyQueues) + foreach (var consumerDefinition in consumerDefinitions) { - if (legacyQueueName == c.QueueName) continue; - if (drainingModels.ContainsKey(legacyQueueName)) continue; - - try + var existing = openModels[consumerDefinition.QueueName]; + var model = existing.model; + if (model != null && + (existing.consumerDefinition.QueuePrefetch == consumerDefinition.QueuePrefetch + && existing.consumerDefinition.ConsumerPriority == consumerDefinition.ConsumerPriority)) + continue; + if (existing.model != null) { - using (var tempModel = conn.CreateModel()) - { - tempModel.QueueDeclarePassive(legacyQueueName); - } - - logger.LogInformation($"Legacy queue found: {legacyQueueName}. Starting migration."); - - using (var tempModel = conn.CreateModel()) + if (existing.consumerDefinition.ConsumerPriority != consumerDefinition.ConsumerPriority) { - tempModel.QueueUnbind(legacyQueueName, busOptions.ProcessExchange, c.RoutingKey, null); - tempModel.QueueUnbind(legacyQueueName, busOptions.ProcessExchange, c.RetryRoutingKey, null); + existing.model.BasicCancel(existing.consumerDefinition.ConsumerTag); + existing.consumerDefinition.ConsumerTag = model.BasicConsume(consumerDefinition.QueueName, + false, "", consumerDefinition.ConsumerArgs, consumerDefinition.ConsumerObject); + } - if (openModels.ContainsKey(legacyQueueName)) - { - var model = openModels[legacyQueueName]; - openModels.Remove(legacyQueueName); - drainingModels.TryAdd(legacyQueueName, model); - logger.LogInformation($"Moved active consumer on {legacyQueueName} to draining mode."); - } - else + if (existing.consumerDefinition.QueuePrefetch != consumerDefinition.QueuePrefetch) { - StartDraining(legacyQueueName, c); + existing.model.BasicQos(0, consumerDefinition.QueuePrefetch, false); } + + continue; } - catch (RabbitMQ.Client.Exceptions.OperationInterruptedException ex) - { - if (ex.ShutdownReason.ReplyCode != 404) - { - logger.LogError(ex, $"Error checking legacy queue {legacyQueueName}"); - } - } - catch (Exception ex) - { - logger.LogError(ex, $"Error checking legacy queue {legacyQueueName}"); - } + + AttachConsumer(consumerDefinition); } } - - private void StartDraining(string queueName, ConsumerDefinition c) + catch (Exception ex) { - var model = conn.CreateModel(); - if (drainingModels.TryAdd(queueName, model)) - { - var consumer = new AsyncEventingBasicConsumer(model); - consumer.Received += async (ch, ea) => - { - await consumerRunner.Run(ea, c, model); - }; - - var args = new Dictionary { { "x-priority", 1 } }; - model.BasicConsume(queueName, false, "", args, consumer); - - logger.LogInformation($"Started draining legacy queue: {queueName}"); - } - else - { - model.Dispose(); - } + logger.LogError(ex, $"Starting {nameof(ConsumersService)}"); } + } - private async Task DrainingLoop(CancellationToken cancellationToken) + private void ConnectionShutdown(object connection, ShutdownEventArgs args) + { + try { - while (!cancellationToken.IsCancellationRequested) - { - try - { - await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken); - - if (drainingModels.IsEmpty || conn == null || !conn.IsOpen) continue; - - using (var model = conn.CreateModel()) - { - var queuesToRemove = new List(); - foreach (var queueName in drainingModels.Keys) - { - try - { - var result = model.QueueDeclarePassive(queueName); - if (result.MessageCount == 0) - { - model.QueueDelete(queueName); - queuesToRemove.Add(queueName); - logger.LogInformation($"Legacy queue {queueName} is empty and has been deleted."); - } - } - catch (Exception ex) - { - logger.LogWarning(ex, $"Error checking draining queue {queueName}"); - if (ex is RabbitMQ.Client.Exceptions.OperationInterruptedException oex && oex.ShutdownReason.ReplyCode == 404) - { - queuesToRemove.Add(queueName); - } - } - } - - foreach(var q in queuesToRemove) - { - if (drainingModels.TryRemove(q, out var drainingModel)) - { - try - { - drainingModel.Close(); - drainingModel.Dispose(); - } - catch - { - // ignored - } - } - } - } - } - catch (Exception ex) - { - logger.LogError(ex, "Error in DrainingLoop"); - } - } + logger.LogWarning($"Consumer RabbitMq connection shutdown. {args.Cause}"); } - - public Task StopAsync(CancellationToken cancellationToken) + catch (Exception) { - - foreach (var model in openModels.Values) + // ignored + } + } - try - { - //model.Close(); - model.Dispose(); - } - catch (Exception ex) - { - logger.LogWarning(ex, $"Failed to stop model."); - } - foreach (var model in drainingModels.Values) + public async Task StopAsync(CancellationToken cancellationToken) + { + foreach (var (model, _) in openModels.Values) + + try { - try - { - model.Dispose(); - } - catch - { - // ignored - } + //model.Close(); + model.Dispose(); + } + catch (Exception ex) + { + logger.LogWarning(ex, $"Failed to stop model."); } - nodeModel?.Dispose(); - conn?.Close(); - conn?.Dispose(); - return Task.CompletedTask; - } + nodeModel?.Dispose(); + conn?.Close(); + conn?.Dispose(); } -} - +} \ No newline at end of file From bf935cd8fb05e9dbaac8d87c2ac1416a71e72988 Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Tue, 30 Dec 2025 10:32:17 +0300 Subject: [PATCH 3/4] refactor: update Publisher class to implement IPublish interface --- SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs | 10 ---------- SW.Bus/Publisher.cs | 11 +---------- 2 files changed, 1 insertion(+), 20 deletions(-) delete mode 100644 SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs diff --git a/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs b/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs deleted file mode 100644 index 7cabac8..0000000 --- a/SW.Bus.RabbitMqExtensions/IRabbitMqPublish.cs +++ /dev/null @@ -1,10 +0,0 @@ -using SW.PrimitiveTypes; - -namespace SW.Bus.RabbitMqExtensions; - -public interface IRabbitMqPublish : IPublish -{ - Task Publish(TMessage message, byte priority); - Task Publish(string messageTypeName, string message, byte priority); - Task Publish(string messageTypeName, byte[] message, byte priority); -} \ No newline at end of file diff --git a/SW.Bus/Publisher.cs b/SW.Bus/Publisher.cs index 005d3f7..4d06862 100644 --- a/SW.Bus/Publisher.cs +++ b/SW.Bus/Publisher.cs @@ -5,7 +5,7 @@ namespace SW.Bus { - internal class Publisher : IRabbitMqPublish + internal class Publisher : IPublish { private readonly BasicPublisher basicPublisher; private readonly string exchange; @@ -20,14 +20,5 @@ public Task Publish(string messageTypeName, string message) => basicPublisher.Publish(messageTypeName, message, exchange); public Task Publish(string messageTypeName, byte[] message) => basicPublisher.Publish(messageTypeName, message, exchange); - - public Task Publish(TMessage message, byte priority) => - basicPublisher.Publish(message, exchange, priority); - - public Task Publish(string messageTypeName, string message, byte priority) => - basicPublisher.Publish(messageTypeName, message, exchange, priority); - - public Task Publish(string messageTypeName, byte[] message, byte priority) => - basicPublisher.Publish(messageTypeName, message, exchange, priority); } } \ No newline at end of file From e863f3c4ec727d67ae09e5b6e9a50814c4308ae8 Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Tue, 30 Dec 2025 11:10:19 +0300 Subject: [PATCH 4/4] feat: extend consumer registration to include IConsumeExtended interface --- .github/workflows/nuget-publish.yml | 2 +- SW.Bus/IServiceCollectionExtensions.cs | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/.github/workflows/nuget-publish.yml b/.github/workflows/nuget-publish.yml index 1385bc1..bfa8385 100644 --- a/.github/workflows/nuget-publish.yml +++ b/.github/workflows/nuget-publish.yml @@ -34,7 +34,7 @@ jobs: - name: Pack and Push NuGet Package uses: simplify9/sw-workflows/actions/dotnet-pack-push@main with: - projects: "SW.Bus/SW.Bus.csproj" + projects: "SW.Bus/SW.Bus.csproj,SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj" configuration: "Release" version: ${{ steps.semver.outputs.version }} api-key: ${{ secrets.SWNUGETKEY }} diff --git a/SW.Bus/IServiceCollectionExtensions.cs b/SW.Bus/IServiceCollectionExtensions.cs index 7178a78..d8a3c77 100644 --- a/SW.Bus/IServiceCollectionExtensions.cs +++ b/SW.Bus/IServiceCollectionExtensions.cs @@ -4,6 +4,7 @@ using RabbitMQ.Client; using SW.HttpExtensions; using SW.PrimitiveTypes; +using SW.Bus.RabbitMqExtensions; using System; using System.Linq; using System.Reflection; @@ -107,6 +108,10 @@ public static IServiceCollection AddBusConsume(this IServiceCollection services, .FromAssemblies(assemblies) .AddClasses(classes => classes.AssignableTo()) .As().AsSelf().WithScopedLifetime()) + .Scan(scan => scan + .FromAssemblies(assemblies) + .AddClasses(classes => classes.AssignableTo()) + .As().AsSelf().WithScopedLifetime()) .Scan(scan => scan .FromAssemblies(assemblies) .AddClasses(classes => classes.AssignableTo(typeof(IConsume<>)))