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.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 new file mode 100644 index 0000000..3ff43a5 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/IConsumeExtended.cs @@ -0,0 +1,8 @@ +using SW.PrimitiveTypes; + +namespace SW.Bus.RabbitMqExtensions; + +public interface IConsumeExtended : IConsume +{ + Task> GetMessageTypeNamesWithOptions(); +} \ 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..4da32d3 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/QueueOptions.cs @@ -0,0 +1,11 @@ +namespace SW.Bus.RabbitMqExtensions; + +public class QueueOptions:ConsumerOptions +{ + public int? RetryCount { get; set; } + public uint? RetryAfterSeconds { 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..4adfda3 --- /dev/null +++ b/SW.Bus.RabbitMqExtensions/SW.Bus.RabbitMqExtensions.csproj @@ -0,0 +1,27 @@ + + + + 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.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..a8bf90d 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; } diff --git a/SW.Bus/ConsumerDefinition.cs b/SW.Bus/ConsumerDefinition.cs index c66ec13..8adaea3 100644 --- a/SW.Bus/ConsumerDefinition.cs +++ b/SW.Bus/ConsumerDefinition.cs @@ -1,58 +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) - { - this.queueNamePrefix = queueNamePrefix; - this.busOptions = busOptions; - NakedQueueName = nakedQueueName; - busOptions.Options.TryGetValue(NakedQueueName, out queueOptions); - - } - - 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 IDictionary RetryArgs => RetryCount == 0 ? null : new Dictionary - { - { "x-dead-letter-exchange", busOptions.ProcessExchange}, - { "x-dead-letter-routing-key", RetryRoutingKey}, - { "x-message-ttl", RetryAfter == 0 ? 100 : RetryAfter * 1000 } - }; + 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 IDictionary ProcessArgs => new Dictionary + public IDictionary RetryArgs => RetryCount == 0 + ? null + : new Dictionary { - { "x-dead-letter-exchange", busOptions.DeadLetterExchange }, + { "x-dead-letter-exchange", busOptions.ProcessExchange }, { "x-dead-letter-routing-key", RetryRoutingKey }, - + { "x-message-ttl", RetryAfter == 0 ? 100 : RetryAfter * 1000 } }; - public static IDictionary BadArgs => new Dictionary - { - { "x-message-ttl", (uint)TimeSpan.FromDays(7).TotalMilliseconds } - }; - public IDictionary ConsumerArgs => queueOptions?.ConsumerArgs; - } + 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 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 cad6c2c..ac5f8a2 100644 --- a/SW.Bus/ConsumerDiscovery.cs +++ b/SW.Bus/ConsumerDiscovery.cs @@ -4,73 +4,86 @@ using System.Reflection; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; +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; + var consumerDefinitions = new List(); + var queueNamePrefix = + $"{busOptions.ProcessExchange}{(string.IsNullOrWhiteSpace(busOptions.ApplicationName) ? "" : $".{busOptions.ApplicationName}")}"; - public ConsumerDiscovery(IServiceProvider sp, BusOptions busOptions) + using var scope = sp.CreateScope(); + var consumers = scope.ServiceProvider.GetServices(); + foreach (var svc in consumers) { - this.sp = sp; - this.busOptions = busOptions; - } - - internal async Task> Load(bool consumersOnly = false) - { - var consumerDefinitions = new List(); - var queueNamePrefix = $"{busOptions.ProcessExchange}{(string.IsNullOrWhiteSpace(busOptions.ApplicationName) ? "" : $".{busOptions.ApplicationName}")}"; + if (svc is IConsumeExtended extendedSvc) + { + var messageTypesWithOptions = await extendedSvc.GetMessageTypeNamesWithOptions(); + foreach (var kvp in messageTypesWithOptions) + { + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, kvp.Value, + $"{svc.GetType().Name}.{kvp.Key}".ToLower()) + { + ServiceType = svc.GetType(), + MessageTypeName = kvp.Key, + }); + } + } + else + { + foreach (var messageTypeName in await svc.GetMessageTypeNames()) - using var scope = sp.CreateScope(); - var consumers = scope.ServiceProvider.GetServices(); - foreach (var svc in consumers) - foreach (var messageTypeName in await svc.GetMessageTypeNames()) + consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, + $"{svc.GetType().Name}.{messageTypeName}".ToLower()) + { + ServiceType = svc.GetType(), + MessageTypeName = messageTypeName, + }); + } + } - consumerDefinitions.Add(new ConsumerDefinition(queueNamePrefix, busOptions, $"{svc.GetType().Name}.{messageTypeName}".ToLower()) - { - ServiceType = svc.GetType(), - MessageTypeName = messageTypeName, - }); + 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<>))) - 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"), + }); - 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; + } - 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") - }); + internal ICollection LoadListeners() + { + var consumerDefinitions = new List(); + using var scope = sp.CreateScope(); - 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(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/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..ebf4209 100644 --- a/SW.Bus/ConsumersService.cs +++ b/SW.Bus/ConsumersService.cs @@ -8,228 +8,245 @@ 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 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); - 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}."); - - // 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 } - } + return Task.CompletedTask; + }; + consumer.Received += async (ch, ea) => { await consumerRunner.Run(ea, consumerDefinition, model); }; + model.BasicQos(0, consumerDefinition.QueuePrefetch, false); - public async Task StopAsync(CancellationToken cancellationToken) - { - - foreach (var model in openModels.Values) + consumerDefinition.ConsumerTag = model.BasicConsume(consumerDefinition.QueueName, false, "", + consumerDefinition.ConsumerArgs, consumer); + } - try + private async Task RefreshConsumers() + { + try + { + consumerDefinitions = await consumerDiscovery.Load(true); + using (var model = conn.CreateModel()) + { + foreach (var c in consumerDefinitions) { - //model.Close(); - model.Dispose(); + if (openModels.ContainsKey(c.QueueName)) + continue; + DeclareAndBind(model, c); } - catch (Exception ex) + } + + foreach (var consumerDefinition in consumerDefinitions) + { + 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) { - logger.LogWarning(ex, $"Failed to stop model."); + if (existing.consumerDefinition.ConsumerPriority != consumerDefinition.ConsumerPriority) + { + existing.model.BasicCancel(existing.consumerDefinition.ConsumerTag); + existing.consumerDefinition.ConsumerTag = model.BasicConsume(consumerDefinition.QueueName, + false, "", consumerDefinition.ConsumerArgs, consumerDefinition.ConsumerObject); + + } + + if (existing.consumerDefinition.QueuePrefetch != consumerDefinition.QueuePrefetch) + { + existing.model.BasicQos(0, consumerDefinition.QueuePrefetch, false); + } + + continue; } - nodeModel?.Dispose(); - conn?.Close(); - conn?.Dispose(); + AttachConsumer(consumerDefinition); + } + } + catch (Exception ex) + { + logger.LogError(ex, $"Starting {nameof(ConsumersService)}"); } } -} + private void ConnectionShutdown(object connection, ShutdownEventArgs args) + { + try + { + logger.LogWarning($"Consumer RabbitMq connection shutdown. {args.Cause}"); + } + catch (Exception) + { + // ignored + } + } + + + public async Task StopAsync(CancellationToken cancellationToken) + { + foreach (var (model, _) in openModels.Values) + + try + { + //model.Close(); + model.Dispose(); + } + catch (Exception ex) + { + logger.LogWarning(ex, $"Failed to stop model."); + } + + nodeModel?.Dispose(); + conn?.Close(); + conn?.Dispose(); + } +} \ No newline at end of file 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<>))) diff --git a/SW.Bus/Publisher.cs b/SW.Bus/Publisher.cs index aefbc93..4d06862 100644 --- a/SW.Bus/Publisher.cs +++ b/SW.Bus/Publisher.cs @@ -1,11 +1,7 @@ - -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 { @@ -24,6 +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); - } } \ 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