diff --git a/README.md b/README.md index b105c0d..d9f0d1e 100644 --- a/README.md +++ b/README.md @@ -289,6 +289,11 @@ To use events, first install the [NuGet package](https://www.nuget.org/packages/ dotnet add package Luxoft.Bss.Platform.Events ``` +> [!NOTE] +> For the new integration-events registration (`AddPlatformIntegrationEvents` overloads, input/output events, +> failed event processing, breaking changes and the queue-name `.v1` switch) see the +> [package README](src/Bss.Platform.Events/README.md). + ### Domain Events To use domain events, you need register it in DI diff --git a/src/Bss.Platform.Events.Abstractions/IEventTypeProvider.cs b/src/Bss.Platform.Events.Abstractions/IEventTypeProvider.cs new file mode 100644 index 0000000..5a6917b --- /dev/null +++ b/src/Bss.Platform.Events.Abstractions/IEventTypeProvider.cs @@ -0,0 +1,14 @@ +namespace Bss.Platform.Events.Abstractions; + +public interface IEventTypeProvider +{ + /// + /// Internal events are sent to the Rabbit->CAP queue and handled within the target system + /// + IReadOnlyDictionary InputEvents { get; } + + /// + /// External events are sent to the Rabbit exchange, but do not have a handler in the target system + /// + IReadOnlyDictionary OutputEvents { get; } +} diff --git a/src/Bss.Platform.Events.Abstractions/IFailedEventProcessor.cs b/src/Bss.Platform.Events.Abstractions/IFailedEventProcessor.cs new file mode 100644 index 0000000..dd28863 --- /dev/null +++ b/src/Bss.Platform.Events.Abstractions/IFailedEventProcessor.cs @@ -0,0 +1,6 @@ +namespace Bss.Platform.Events.Abstractions; + +public interface IFailedEventProcessor +{ + Task HandleAsync(TInputEvent? value, Exception ex, string? rawMessageBody); +} diff --git a/src/Bss.Platform.Events.Abstractions/IIntegrationEventPublisher.cs b/src/Bss.Platform.Events.Abstractions/IIntegrationEventPublisher.cs index 1e6cb80..d95f2ad 100644 --- a/src/Bss.Platform.Events.Abstractions/IIntegrationEventPublisher.cs +++ b/src/Bss.Platform.Events.Abstractions/IIntegrationEventPublisher.cs @@ -1,6 +1,8 @@ namespace Bss.Platform.Events.Abstractions; -public interface IIntegrationEventPublisher +public interface IIntegrationEventPublisher { - Task PublishAsync(IIntegrationEvent @event, CancellationToken cancellationToken); + Task PublishAsync(T @event, CancellationToken cancellationToken); } + +public interface IIntegrationEventPublisher : IIntegrationEventPublisher; diff --git a/src/Bss.Platform.Events/AssemblyInfo.cs b/src/Bss.Platform.Events/AssemblyInfo.cs new file mode 100644 index 0000000..56d4936 --- /dev/null +++ b/src/Bss.Platform.Events/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Tests.Unit")] diff --git a/src/Bss.Platform.Events/Bss.Platform.Events.csproj b/src/Bss.Platform.Events/Bss.Platform.Events.csproj index 7a22725..73ff420 100644 --- a/src/Bss.Platform.Events/Bss.Platform.Events.csproj +++ b/src/Bss.Platform.Events/Bss.Platform.Events.csproj @@ -5,11 +5,13 @@ + + diff --git a/src/Bss.Platform.Events/CapConsumerServiceSelector.cs b/src/Bss.Platform.Events/CapConsumerServiceSelector.cs deleted file mode 100644 index dd33ba5..0000000 --- a/src/Bss.Platform.Events/CapConsumerServiceSelector.cs +++ /dev/null @@ -1,53 +0,0 @@ -using System.Reflection; - -using Bss.Platform.Events.Abstractions; - -using DotNetCore.CAP; -using DotNetCore.CAP.Internal; - -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Options; - -namespace Bss.Platform.Events; - -public class CapConsumerServiceSelector(IServiceProvider serviceProvider, Assembly assembly) - : ConsumerServiceSelector(serviceProvider) -{ - protected override IEnumerable FindConsumersFromControllerTypes() => []; - - protected override IEnumerable FindConsumersFromInterfaceTypes(IServiceProvider provider) - { - var namePrefix = provider.GetRequiredService>().Value.TopicNamePrefix; - - return assembly - .ExportedTypes - .Where(x => typeof(IIntegrationEvent).IsAssignableFrom(x) && x is { IsInterface: false, IsAbstract: false }) - .Select(x => this.CreateExecutorDescriptor(typeof(CapConsumerExecutor<>).MakeGenericType(x), x, namePrefix)); - } - - private ConsumerExecutorDescriptor CreateExecutorDescriptor(Type executor, Type @event, string? namePrefix) - { - var subscribeAttribute = new CapSubscribeAttribute(@event.Name); - this.SetSubscribeAttribute(subscribeAttribute); - - var methodInfo = executor - .GetRuntimeMethods() - .Single(x => x.Name.Contains(nameof(CapConsumerExecutor.HandleAsync))); - - var methodParameters = methodInfo.GetParameters(); - return new ConsumerExecutorDescriptor - { - Attribute = subscribeAttribute, - ClassAttribute = null, - MethodInfo = methodInfo, - ImplTypeInfo = executor.GetTypeInfo(), - ServiceTypeInfo = null, - TopicNamePrefix = namePrefix, - Parameters = new List - { - new() { ParameterType = methodParameters[0].ParameterType, IsFromCap = false }, - new() { ParameterType = methodParameters[1].ParameterType, IsFromCap = true } - } - }; - } -} diff --git a/src/Bss.Platform.Events/DependencyInjection.cs b/src/Bss.Platform.Events/DependencyInjection.cs index f67d064..fa9c668 100644 --- a/src/Bss.Platform.Events/DependencyInjection.cs +++ b/src/Bss.Platform.Events/DependencyInjection.cs @@ -3,15 +3,22 @@ using Bss.Platform.Events.Abstractions; using Bss.Platform.Events.Interfaces; +using Bss.Platform.Events.Internal; using Bss.Platform.Events.Models; using Bss.Platform.Events.Publishers; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; using DotNetCore.CAP; +using DotNetCore.CAP.Filter; using DotNetCore.CAP.Internal; +using DotNetCore.CAP.Messages; +using DotNetCore.CAP.Serialization; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Mvc.Filters; +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Savorboard.CAP.InMemoryMessageQueue; @@ -26,71 +33,228 @@ public static IServiceCollection AddPlatformIntegrationEvents( this IServiceCollection services, Assembly eventsAssembly, Action? setup = null) - where TEventProcessor : class, IIntegrationEventProcessor + where TEventProcessor : class, IIntegrationEventProcessor, IIntegrationEventProcessor { services .AddSingleton() - .AddSingleton(x => new CapConsumerServiceSelector(x, eventsAssembly)) - .AddScoped() - .AddScoped( - serviceProvider => - { - var capTransaction = ActivatorUtilities.CreateInstance(serviceProvider); - capTransaction.DbTransaction = serviceProvider.GetRequiredService(); - return capTransaction; - }) - .AddCap( - x => - { - var eventsOptions = IntegrationEventsOptions.Default; - setup?.Invoke(eventsOptions); + .AddSingleton(x => new(x, eventsAssembly)); + services.AddPlatformIntegrationEventsInternal(setup); + services.TryAddLegacyEventPublisher(); + + return services; + } + + /// + /// A new way to register integration events, required to set up internal and external events
+ /// Used Bss.Platform.Mediation + /// + /// Automatically registered Legacy IIntegrationEventPublisher + public static IServiceCollection AddPlatformIntegrationEvents( + this IServiceCollection services, + Action> setupEvents, + Action setupOptions) + where TEventProcessor : class, IIntegrationEventProcessor => + services + .AddPlatformIntegrationEvents(setupEvents, setupOptions) + .TryAddLegacyEventPublisher(); - x.FailedRetryCount = eventsOptions.FailedRetryCount; - x.SucceedMessageExpiredAfter = (int)TimeSpan.FromDays(eventsOptions.RetentionDays).TotalSeconds; + /// + /// A new way to register integration events, required to set up internal and external events + /// + /// Automatically registered IIntegrationEventPublisher<TEvent> wrap it if you need + public static IServiceCollection AddPlatformIntegrationEvents( + this IServiceCollection services, + Action> setupEvents, + Action setupOptions) + where TEventProcessor : class, IIntegrationEventProcessor + where TEvent : class => + services.AddPlatformIntegrationEvents(setupEvents, setupOptions); - x.UseSqlServer( - o => - { - o.ConnectionString = eventsOptions.SqlServer.ConnectionString; - o.Schema = eventsOptions.SqlServer.Schema; - }); + /// + /// A new way to register integration events, required to set up internal and external events + /// + /// Automatically registered IIntegrationEventPublisher<TEvent> wrap it if you need + public static IServiceCollection AddPlatformIntegrationEvents( + this IServiceCollection services, + Action> setupEvents, + Action setupOptions) + where TEventProcessor : class, IIntegrationEventProcessor + where TInputEvent : class + where TOutputEvent : notnull + { + // TODO: via configuration + var typeProvider = new EventTypeProvider(); + setupEvents.Invoke(typeProvider); - x.UseDashboard(o => + services + .AddSingleton(typeProvider) + .AddSingleton(); + var eventsOptions = services.AddPlatformIntegrationEventsInternal(setupOptions); + services.Configure((RabbitMQOptions opt) => + { + // NOTE: required for rabbit messages generated outside of CAP + opt.CustomHeadersBuilder = (msg, sp) => + [ + new(Headers.MessageId, sp.GetRequiredService().NextId().ToString()), + new(Headers.MessageName, msg.RoutingKey), + new(Headers.Type, typeof(TInputEvent).Name) + ]; + }) + .TryAddScoped, IntegrationEventPublisherNew>(); + + services.AddSingleton, TEventProcessor>(); + // NOTE: register TEventProcessor for each type (required for CapConsumerExecutor) + typeProvider.InputEvents.Keys + .Select(t => typeof(IIntegrationEventProcessor<>).MakeGenericType(t)) + .ToList() + .ForEach(x => services.AddSingleton(x, sp => sp.GetRequiredService>())); + + if (eventsOptions.UseFailedEventProcessor) + { + services.AddSingleton(); + var (exchange, queue) = eventsOptions.DeadLetterOptions; + services.AddSingleton(new DeadLetterBindingsInitializer(exchange, queue)); + services.AddSingleton>(sp => sp.GetRequiredService()); + services.RemoveAll(typeof(ISerializer)); + services.AddSingleton(); + services.AddScoped>(); + } + + if (eventsOptions.MessageQueue.Enable) + { + services.AddExternalSystemQueueBindings(eventsOptions.MessageQueue.ExternalSystemBindingsSectionPath); + services.AddSingleton(); + services.AddSingleton(); + if (eventsOptions.MessageQueue.SchemaExportSettings != null) + { + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + } + + services.AddHostedService(); + } + return services; + } + + private static IServiceCollection TryAddLegacyEventPublisher(this IServiceCollection services) + { + services.TryAddScoped(); + services.TryAddScoped>(sp => sp.GetRequiredService()); + return services; + } + + /// + /// required for backward compatibility, added postfix ".v1" to queue name, like origin cap behavior without + /// + // TODO: remove or investigate when it's really really needed + private static Action SetLegacyQueueNameWithVersion(Action? setupOptions) => + opt => + { + setupOptions?.Invoke(opt); + var originMessageQueueName = opt.MessageQueue.QueueName; + opt.MessageQueue.QueueName = string.IsNullOrWhiteSpace(originMessageQueueName) ? $"{opt.MessageQueue.ExchangeName}.v1" : originMessageQueueName; + }; + + private static TOptions AddPlatformIntegrationEventsInternal( + this IServiceCollection services, + Action? setupEventOptions = null) + where TInputEvent : class + where TOptions : IntegrationEventsOptions, new() + { + var eventsOptions = new TOptions(); + setupEventOptions?.Invoke(eventsOptions); + setupEventOptions ??= _ => { }; + services.Configure(setupEventOptions); + + services + .AddScoped(serviceProvider => + { + var capTransaction = ActivatorUtilities.CreateInstance(serviceProvider); + capTransaction.DbTransaction = serviceProvider.GetRequiredService(); + return capTransaction; + }) + .AddCap(x => + { + x.JsonSerializerOptions.PropertyNameCaseInsensitive = true; + x.FailedRetryCount = eventsOptions.FailedRetryCount; + x.SucceedMessageExpiredAfter = (int)TimeSpan.FromDays(eventsOptions.RetentionDays).TotalSeconds; + + if (!string.IsNullOrEmpty(eventsOptions.SqlServer.ConnectionString)) + { + x.UseSqlServer(o => { - o.PathMatch = eventsOptions.DashboardPath; - if (eventsOptions.AuthorizationPredicate is { } authPredicate) - { - o.AllowAnonymousExplicit = false; - o.AuthorizationPolicy = AddDashboardAuthorizationPolicy(services, authPredicate); - } + o.ConnectionString = eventsOptions.SqlServer.ConnectionString; + o.Schema = eventsOptions.SqlServer.Schema; }); + } - if (!eventsOptions.MessageQueue.Enable) + x.UseDashboard(o => + { + o.PathMatch = eventsOptions.DashboardPath; + o.PathBase = eventsOptions.GatewayPrefix; + if (eventsOptions.AuthorizationPredicate is { } authPredicate) { - x.UseInMemoryMessageQueue(); - return; + o.AllowAnonymousExplicit = false; + o.AuthorizationPolicy = AddDashboardAuthorizationPolicy(services, authPredicate); } - - x.DefaultGroupName = eventsOptions.MessageQueue.ExchangeName; - x.UseRabbitMQ( - o => - { - o.HostName = eventsOptions.MessageQueue.Host; - o.Port = eventsOptions.MessageQueue.Port; - o.VirtualHost = eventsOptions.MessageQueue.VirtualHost; - o.Password = eventsOptions.MessageQueue.Secret; - o.UserName = eventsOptions.MessageQueue.UserName; - o.ExchangeName = eventsOptions.MessageQueue.ExchangeName; - o.BasicQosOptions = new RabbitMQOptions.BasicQos(1, true); - }); }); - return services; + var rabbitSettings = eventsOptions.MessageQueue; + if (rabbitSettings.Enable) + { + x.DefaultGroupName = string.IsNullOrWhiteSpace(rabbitSettings.QueueName) + ? rabbitSettings.ExchangeName + : rabbitSettings.QueueName; + + x.UseRabbitMQ(o => + { + o.HostName = rabbitSettings.Host; + o.Port = rabbitSettings.Port; + o.VirtualHost = rabbitSettings.VirtualHost; + o.Password = rabbitSettings.Secret; + o.UserName = rabbitSettings.UserName; + o.ExchangeName = rabbitSettings.ExchangeName; + o.BasicQosOptions = new(1, true); + }); + } + else + { + x.UseInMemoryMessageQueue(); + } + + eventsOptions.OverrideCapOptions?.Invoke(x); + }); + + return eventsOptions; } + private static void AddExternalSystemQueueBindings(this IServiceCollection services, string? sectionPath) + { + var optionsBuilder = services.AddOptions(); + if (!string.IsNullOrWhiteSpace(sectionPath)) + { + optionsBuilder.BindConfiguration(sectionPath); + // ConfigurationBinder drops Dictionary entries whose bound value is null, and both `null` and `[]` + // bind to null for an array value - so BindConfiguration above silently loses SystemBindings keys + // that have no explicit routing keys. Rebuild that dictionary from IConfiguration directly instead. + // Uses Configure (not PostConfigure) so consumers can still override via PostConfigure regardless + // of call order relative to this setup method - PostConfigure always runs after every Configure. + optionsBuilder.Configure((options, configuration) => + options.SystemBindings = ResolveSystemBindings(configuration, sectionPath)); + } + } + + internal static Dictionary ResolveSystemBindings(IConfiguration configuration, string sectionPath) => + configuration + .GetSection(sectionPath) + .GetSection(nameof(ExternalSystemBindingsOptions.SystemBindings)) + .GetChildren() + .ToDictionary(section => section.Key, section => section.Get()); + private static string AddDashboardAuthorizationPolicy(IServiceCollection services, Func> authPredicate) { - const string policyName = "bss-platform-dashboard-auth"; + const string policyName = "bss-platform-events-dashboard-auth"; services.AddAuthorizationBuilder() .AddPolicy( policyName, diff --git a/src/Bss.Platform.Events/EventTypeProvider.cs b/src/Bss.Platform.Events/EventTypeProvider.cs new file mode 100644 index 0000000..332c4bb --- /dev/null +++ b/src/Bss.Platform.Events/EventTypeProvider.cs @@ -0,0 +1,79 @@ +using System.Reflection; + +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Interfaces; + +namespace Bss.Platform.Events; + +public class EventTypeProvider : IEventTypeProvider, IIntegrationEventSetup +{ + public IReadOnlyDictionary InputEvents => this.inputTypes; + public IReadOnlyDictionary OutputEvents => this.outputTypes; + + // TODO: reverse inputTypes to Dictionary + private readonly Dictionary inputTypes = []; + private readonly Dictionary outputTypes = []; + + public IIntegrationEventSetup AddInputEvents(string prefix = "", params Assembly[] assemblies) + where TEvent : TIn + { + var newTypes = GetOrDefaultAssembly(assemblies) + .SelectMany(x => x.DefinedTypes) + .Where(this.IsAssignableAndSatisfyCondition) + .Except(this.inputTypes.Keys); + + foreach (var newType in newTypes) + { + this.inputTypes[newType] = $"{prefix}{newType.Name}"; + } + + return this; + } + + public IIntegrationEventSetup AddInputEvent(string routingKey) + where TEvent : class, TIn + { + var type = typeof(TEvent); + this.inputTypes[type] = routingKey; + return this; + } + + public IIntegrationEventSetup AddOutputEvents(string prefix = "", params Assembly[] assemblies) + where TEvent : TOut + { + var newTypes = GetOrDefaultAssembly(assemblies) + .SelectMany(x => x.DefinedTypes) + .Where(this.IsAssignableAndSatisfyCondition) + .Except(this.outputTypes.Keys); + + foreach (var newType in newTypes) + { + this.outputTypes[newType] = $"{prefix}{newType.Name}"; + } + + return this; + } + + public IIntegrationEventSetup AddOutputEvent(string routingKey) + where TEvent : class, TOut + { + var type = typeof(TEvent); + this.outputTypes[type] = routingKey; + return this; + } + + private static Assembly[] GetOrDefaultAssembly(Assembly[] assemblies) + { + if (assemblies.Length == 0) + { + assemblies = [typeof(TEvent).Assembly]; + } + + return assemblies; + } + + public virtual bool IsAssignableAndSatisfyCondition(TypeInfo typeInfo) => + typeInfo is { IsInterface: false, IsAbstract: false, IsNested: false } + && typeof(TAssignableTo).IsAssignableFrom(typeInfo) + && !typeInfo.Name.Contains('<'); +} diff --git a/src/Bss.Platform.Events/Interfaces/IIntegrationEventProcessor.cs b/src/Bss.Platform.Events/Interfaces/IIntegrationEventProcessor.cs index 25af453..8ff2064 100644 --- a/src/Bss.Platform.Events/Interfaces/IIntegrationEventProcessor.cs +++ b/src/Bss.Platform.Events/Interfaces/IIntegrationEventProcessor.cs @@ -2,7 +2,9 @@ namespace Bss.Platform.Events.Interfaces; -public interface IIntegrationEventProcessor +public interface IIntegrationEventProcessor : IIntegrationEventProcessor; + +public interface IIntegrationEventProcessor { - Task ProcessAsync(IIntegrationEvent @event, CancellationToken token); + Task ProcessAsync(T @event, CancellationToken token); } diff --git a/src/Bss.Platform.Events/Interfaces/IIntegrationEventSetup.cs b/src/Bss.Platform.Events/Interfaces/IIntegrationEventSetup.cs new file mode 100644 index 0000000..34c45d8 --- /dev/null +++ b/src/Bss.Platform.Events/Interfaces/IIntegrationEventSetup.cs @@ -0,0 +1,41 @@ +using System.Reflection; + +namespace Bss.Platform.Events.Interfaces; + +public interface IIntegrationEventSetup +{ + /// + /// Add multiple events implemented or inherited TInternalBase with the prefix + /// + /// + /// prefix to add before type name + /// <TInternalBase>("INT.") -> INT.TInternal + /// + /// assemblies to find types, if not passed - will be used assembly contains TInternalBase + IIntegrationEventSetup AddInputEvents(string prefix = "", params Assembly[] assemblies) + where TInputBase : TIn; + + /// + /// Add a single event with the routing key, overrides if it already exists (added by + /// ) + /// + IIntegrationEventSetup AddInputEvent(string routingKey) where TInput : class, TIn; + + /// + /// Add multiple events implemented or inherited TExternalBase with the prefix + /// + /// + /// prefix to add before type name + /// <TExternalBase>("SYS.") -> SYS.TExternal + /// + /// assemblies to find types, if not passed - will be used assembly contains TExternalBase + IIntegrationEventSetup AddOutputEvents(string prefix, params Assembly[] assemblies) where TOutputBase : TOut; + + /// + /// Add a single event with the routing key, overrides if it already exists (added by + /// ) + /// + IIntegrationEventSetup AddOutputEvent(string routingKey) where TOutput : class, TOut; + + bool IsAssignableAndSatisfyCondition(TypeInfo typeInfo); +} diff --git a/src/Bss.Platform.Events/Interfaces/IRabbitInitializer.cs b/src/Bss.Platform.Events/Interfaces/IRabbitInitializer.cs new file mode 100644 index 0000000..651363e --- /dev/null +++ b/src/Bss.Platform.Events/Interfaces/IRabbitInitializer.cs @@ -0,0 +1,12 @@ +using RabbitMQ.Client; + +namespace Bss.Platform.Events.Interfaces; + +/// +/// Runs one-time RabbitMQ topology setup (queues, bindings, etc.) against a channel rented at startup.
+/// Register additional implementations via services.AddSingleton<IRabbitInitializer, TInitializer>(). +/// +public interface IRabbitInitializer +{ + Task InitializeAsync(IModel model, CancellationToken cancellationToken); +} diff --git a/src/Bss.Platform.Events/CapConsumerExecutor.cs b/src/Bss.Platform.Events/Internal/CapConsumerExecutor.cs similarity index 65% rename from src/Bss.Platform.Events/CapConsumerExecutor.cs rename to src/Bss.Platform.Events/Internal/CapConsumerExecutor.cs index ca29eb7..ff492fa 100644 --- a/src/Bss.Platform.Events/CapConsumerExecutor.cs +++ b/src/Bss.Platform.Events/Internal/CapConsumerExecutor.cs @@ -1,10 +1,8 @@ -using Bss.Platform.Events.Abstractions; using Bss.Platform.Events.Interfaces; -namespace Bss.Platform.Events; +namespace Bss.Platform.Events.Internal; -internal class CapConsumerExecutor(IIntegrationEventProcessor eventProcessor) - where TEvent : IIntegrationEvent +internal class CapConsumerExecutor(IIntegrationEventProcessor eventProcessor) { public Task HandleAsync(TEvent @event, CancellationToken cancellationToken) => eventProcessor.ProcessAsync(@event, cancellationToken); } diff --git a/src/Bss.Platform.Events/Internal/CapConsumerServiceSelector.cs b/src/Bss.Platform.Events/Internal/CapConsumerServiceSelector.cs new file mode 100644 index 0000000..5d6e276 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/CapConsumerServiceSelector.cs @@ -0,0 +1,83 @@ +using System.Reflection; + +using Bss.Platform.Events.Abstractions; + +using DotNetCore.CAP; +using DotNetCore.CAP.Internal; + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Bss.Platform.Events.Internal; + +public class CapConsumerServiceSelectorLegacy(IServiceProvider serviceProvider, Assembly assembly) + : CapConsumerServiceSelectorBase(serviceProvider) +{ + protected override IEnumerable GetInternalEventTypes() => + assembly + .ExportedTypes + .Where(x => typeof(IIntegrationEvent).IsAssignableFrom(x) && x is { IsInterface: false, IsAbstract: false }); + + protected override CapSubscribeAttribute ProvideCapSubscribeAttribute(Type eventType) + { + var subscribeAttribute = new CapSubscribeAttribute(eventType.Name); + this.SetSubscribeAttribute(subscribeAttribute); + return subscribeAttribute; + } +} + +public class CapConsumerServiceSelectorNew(IServiceProvider serviceProvider, IEventTypeProvider eventTypeProvider, IOptions capOptions) + : CapConsumerServiceSelectorBase(serviceProvider) +{ + private readonly string queueName = capOptions.Value.DefaultGroupName; + protected override IEnumerable GetInternalEventTypes() => + eventTypeProvider.InputEvents.Keys; + + protected override CapSubscribeAttribute ProvideCapSubscribeAttribute(Type eventType) => + new(eventTypeProvider.InputEvents[eventType]) { Group = this.queueName }; +} + +public abstract class CapConsumerServiceSelectorBase(IServiceProvider serviceProvider) + : ConsumerServiceSelector(serviceProvider) +{ + protected abstract IEnumerable GetInternalEventTypes(); + + protected abstract CapSubscribeAttribute ProvideCapSubscribeAttribute(Type eventType); + + protected override IEnumerable FindConsumersFromControllerTypes() => + []; + + protected override IEnumerable FindConsumersFromInterfaceTypes(IServiceProvider provider) + { + var namePrefix = provider.GetRequiredService>().Value.TopicNamePrefix; + + return this.GetInternalEventTypes() + .Select(x => this.CreateExecutorDescriptor(typeof(CapConsumerExecutor<>).MakeGenericType(x), x, namePrefix)); + } + + + private ConsumerExecutorDescriptor CreateExecutorDescriptor(Type executor, Type @event, string? namePrefix) + { + var subscribeAttribute = this.ProvideCapSubscribeAttribute(@event); + + var methodInfo = executor + .GetRuntimeMethods() + .Single(x => x.Name.Contains(nameof(CapConsumerExecutor<>.HandleAsync))); + + var methodParameters = methodInfo.GetParameters(); + return new ConsumerExecutorDescriptor + { + Attribute = subscribeAttribute, + ClassAttribute = null, + MethodInfo = methodInfo, + ImplTypeInfo = executor.GetTypeInfo(), + ServiceTypeInfo = null, + TopicNamePrefix = namePrefix, + Parameters = new List + { + new() { ParameterType = methodParameters[0].ParameterType, IsFromCap = false }, + new() { ParameterType = methodParameters[1].ParameterType, IsFromCap = true } + } + }; + } +} diff --git a/src/Bss.Platform.Events/Internal/CapExceptionFilter.cs b/src/Bss.Platform.Events/Internal/CapExceptionFilter.cs new file mode 100644 index 0000000..3f995da --- /dev/null +++ b/src/Bss.Platform.Events/Internal/CapExceptionFilter.cs @@ -0,0 +1,48 @@ +using System.Text.Json; + +using Bss.Platform.Events.Abstractions; + +using DotNetCore.CAP; +using DotNetCore.CAP.Filter; +using DotNetCore.CAP.Serialization; + +using Microsoft.Extensions.Options; + +namespace Bss.Platform.Events.Internal; + +internal sealed class CapExceptionFilter( + IOptions capOptions, + IEnumerable> failsProcessors, + ISerializer serializer) + : SubscribeFilter where TInputEvent : class +{ + private const string ErrorDetailsHeader = "x-cap-failure-details"; + + // NOTE: -1 value because the CAP incremented after filters and handle 0 retries case + private int LatestRetryCount => Math.Max(capOptions.Value.FailedRetryCount - 1, 0); + + public override Task OnSubscribeExceptionAsync(ExceptionContext context) + { + if (context.MediumMessage.Retries != this.LatestRetryCount) + { + return Task.CompletedTask; + } + + var ex = context.Exception; + var errorText = ex.GetBaseException().Message; + + var details = new CapFailureDetails(ex.GetType().FullName ?? ex.GetType().Name, errorText, ex.StackTrace); + context.DeliverMessage.Headers[ErrorDetailsHeader] = JsonSerializer.Serialize(details); + context.DeliverMessage.Headers.TryGetValue(RawCapturingSerializer.RawBodyHeader, out var rawMessageBody); + + var payloadParam = context.ConsumerDescriptor.Parameters.SingleOrDefault(p => !p.IsFromCap); + var value = context.DeliverMessage.Value; + var payload = payloadParam is not null && value is not null && serializer.IsJsonType(value) + ? serializer.Deserialize(value, payloadParam.ParameterType) + : value; + + return Task.WhenAll(failsProcessors.Select(x => x.HandleAsync(payload as TInputEvent, ex, rawMessageBody))); + } + + internal sealed record CapFailureDetails(string ExceptionType, string Message, string? StackTrace); +} diff --git a/src/Bss.Platform.Events/Internal/DeadLetterBindingsInitializer.cs b/src/Bss.Platform.Events/Internal/DeadLetterBindingsInitializer.cs new file mode 100644 index 0000000..c63a4ce --- /dev/null +++ b/src/Bss.Platform.Events/Internal/DeadLetterBindingsInitializer.cs @@ -0,0 +1,16 @@ +using Bss.Platform.Events.Interfaces; + +using RabbitMQ.Client; + +namespace Bss.Platform.Events.Internal; + +internal sealed class DeadLetterBindingsInitializer(string exchange, string queue) : IRabbitInitializer +{ + public Task InitializeAsync(IModel model, CancellationToken cancellationToken) + { + model.ExchangeDeclare(exchange, ExchangeType.Fanout, true); + model.QueueDeclare(queue, true, false, false, null); + model.QueueBind(queue, exchange, string.Empty); + return Task.CompletedTask; + } +} diff --git a/src/Bss.Platform.Events/Internal/DeadLetterProcessor.cs b/src/Bss.Platform.Events/Internal/DeadLetterProcessor.cs new file mode 100644 index 0000000..f7c9af2 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/DeadLetterProcessor.cs @@ -0,0 +1,68 @@ +using System.Text; + +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Models; + +using DotNetCore.CAP; +using DotNetCore.CAP.RabbitMQ; + +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +using RabbitMQ.Client; + +namespace Bss.Platform.Events.Internal; + +internal sealed partial class DeadLetterProcessor( + IConnectionChannelPool connectionChannelPool, + IOptions capOptions, + IOptions eventOptions, + ILogger logger, + IEventTypeProvider registeredTypes) : IFailedEventProcessor +{ + private readonly string exchangeName = eventOptions.Value.DeadLetterOptions.ExchangeName; + private readonly string originMessageQueueName = capOptions.Value.DefaultGroupName; + + public Task HandleAsync(object? value, Exception ex, string? rawMessageBody) + { + var routingKey = value != null && registeredTypes?.InputEvents.TryGetValue(value.GetType(), out var registeredRoutingKey) == true + ? registeredRoutingKey + : $"unknown message: {value?.GetType().Name ?? ""}"; + + this.SendDeadLetter(routingKey, ex, rawMessageBody); + return Task.CompletedTask; + } + + internal void SendDeadLetter(string routingKey, Exception ex, string? rawMessageBody) + { + try + { + using var channel = connectionChannelPool.Rent(); + + var props = channel.CreateBasicProperties(); + props.DeliveryMode = 2; + + props.Headers = new Dictionary + { + ["error"] = ex.GetBaseException().Message, + ["queue"] = this.originMessageQueueName, + ["routingKey"] = routingKey, + ["stacktrace"] = ex.StackTrace ?? string.Empty + }; + + channel.BasicPublish(this.exchangeName, string.Empty, props, Encoding.UTF8.GetBytes(rawMessageBody ?? string.Empty)); + + if (channel.NextPublishSeqNo > 0) + { + channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5)); + } + } + catch (Exception exception) + { + this.LogError(routingKey, exception.GetType().Name, exception.Message); + } + } + + [LoggerMessage(LogLevel.Error, Message = "Fail send deadletter {RoutingKey}, {ErrorType}, {ErrorText}")] + partial void LogError(string routingKey, string errorType, string errorText); +} diff --git a/src/Bss.Platform.Events/Internal/ExternalSystemBindingsResolver.cs b/src/Bss.Platform.Events/Internal/ExternalSystemBindingsResolver.cs new file mode 100644 index 0000000..d49d949 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/ExternalSystemBindingsResolver.cs @@ -0,0 +1,34 @@ +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Models; + +using Microsoft.Extensions.Options; + +namespace Bss.Platform.Events.Internal; + +public class ExternalSystemBindingsResolver( + IEventTypeProvider eventTypeProvider, + IOptions options) : IExternalSystemBindingsResolver +{ + /// + /// Resolves which output-event routing keys each configured queue should be bound to.
+ /// null/empty values fall back to the default + /// set: all except those matching . + /// + public IReadOnlyDictionary> ResolveQueueBindings() + { + var defaultRoutingKeys = eventTypeProvider.OutputEvents.Values + .Distinct() + .Where(routingKey => !options.Value.ExcludeOutputEvents.Any(pattern => WildcardMatcher.IsMatch(routingKey, pattern))) + .ToArray(); + + return options.Value.SystemBindings.ToDictionary( + x => x.Key, + x => (IReadOnlyList)(x.Value is { Length: > 0 } explicitRoutingKeys ? explicitRoutingKeys : defaultRoutingKeys)); + } + + public IReadOnlyDictionary ResolveOutputEventsForExport() + { + var allUniqOutputRoutingKeys = this.ResolveQueueBindings().Values.SelectMany(x => x).Distinct(); + return eventTypeProvider.OutputEvents.Where(x => allUniqOutputRoutingKeys.Contains(x.Value)).ToDictionary(x => x.Value, x => x.Key); + } +} diff --git a/src/Bss.Platform.Events/Internal/ExternalSystemQueueBindingsInitializer.cs b/src/Bss.Platform.Events/Internal/ExternalSystemQueueBindingsInitializer.cs new file mode 100644 index 0000000..999b776 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/ExternalSystemQueueBindingsInitializer.cs @@ -0,0 +1,31 @@ +using Bss.Platform.Events.Interfaces; +using Bss.Platform.Events.Models; + +using Microsoft.Extensions.Options; + +using RabbitMQ.Client; + +namespace Bss.Platform.Events.Internal; + +internal sealed class ExternalSystemQueueBindingsInitializer( + IExternalSystemBindingsResolver bindingsResolver, + IOptions eventOptions) : IRabbitInitializer +{ + public Task InitializeAsync(IModel model, CancellationToken cancellationToken) + { + var exchangeName = eventOptions.Value.MessageQueue.ExchangeName; + foreach (var (queue, routingKeys) in bindingsResolver.ResolveQueueBindings()) + { + cancellationToken.ThrowIfCancellationRequested(); + + model.QueueDeclare(queue, true, false, false, null); + + foreach (var routingKey in routingKeys.Distinct()) + { + model.QueueBind(queue, exchangeName, routingKey); + } + } + + return Task.CompletedTask; + } +} diff --git a/src/Bss.Platform.Events/Internal/IExternalSystemBindingsResolver.cs b/src/Bss.Platform.Events/Internal/IExternalSystemBindingsResolver.cs new file mode 100644 index 0000000..f78bab4 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/IExternalSystemBindingsResolver.cs @@ -0,0 +1,10 @@ +using Bss.Platform.Events.Models; + +namespace Bss.Platform.Events.Internal; + +internal interface IExternalSystemBindingsResolver +{ + IReadOnlyDictionary> ResolveQueueBindings(); + + IReadOnlyDictionary ResolveOutputEventsForExport(); +} diff --git a/src/Bss.Platform.Events/Internal/RabbitExportEventsSchemaInitializer.cs b/src/Bss.Platform.Events/Internal/RabbitExportEventsSchemaInitializer.cs new file mode 100644 index 0000000..a35193e --- /dev/null +++ b/src/Bss.Platform.Events/Internal/RabbitExportEventsSchemaInitializer.cs @@ -0,0 +1,16 @@ +using Bss.Platform.Events.Interfaces; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +using RabbitMQ.Client; + +namespace Bss.Platform.Events.Internal; + +public sealed class RabbitExportEventsSchemaInitializer(RabbitEventsSchemaExporter exporter) : IRabbitInitializer +{ + public Task InitializeAsync(IModel model, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + exporter.Export(model); + return Task.CompletedTask; + } +} diff --git a/src/Bss.Platform.Events/Internal/RabbitInitializersHostedService.cs b/src/Bss.Platform.Events/Internal/RabbitInitializersHostedService.cs new file mode 100644 index 0000000..ad9f70e --- /dev/null +++ b/src/Bss.Platform.Events/Internal/RabbitInitializersHostedService.cs @@ -0,0 +1,35 @@ +using Bss.Platform.Events.Interfaces; + +using DotNetCore.CAP.RabbitMQ; + +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace Bss.Platform.Events.Internal; + +internal sealed partial class RabbitInitializersHostedService( + IConnectionChannelPool connectionChannelPool, + IEnumerable initializers, + ILogger logger) : IHostedService +{ + public async Task StartAsync(CancellationToken cancellationToken) + { + foreach (var initializer in initializers) + { + try + { + using var channel = connectionChannelPool.Rent(); + await initializer.InitializeAsync(channel, cancellationToken); + } + catch (Exception ex) + { + this.LogInitializerFailed(initializer.GetType().Name, ex); + } + } + } + + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + [LoggerMessage(LogLevel.Error, Message = "Rabbit initializer {InitializerName} failed to run")] + partial void LogInitializerFailed(string initializerName, Exception ex); +} diff --git a/src/Bss.Platform.Events/Internal/RabbitSchemaExportSettings.cs b/src/Bss.Platform.Events/Internal/RabbitSchemaExportSettings.cs new file mode 100644 index 0000000..f91580b --- /dev/null +++ b/src/Bss.Platform.Events/Internal/RabbitSchemaExportSettings.cs @@ -0,0 +1,33 @@ +using System.Reflection; + +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Models; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +using DotNetCore.CAP; + +using Microsoft.Extensions.Options; + +namespace Bss.Platform.Events.Internal; + +internal class RabbitSchemaExportSettings( + IOptions capOptions, + IOptions rabbitOptions, + IOptions eventOptions, + IEventTypeProvider eventTypeProvider, + IExternalSystemBindingsResolver bindingsResolver) + : IRabbitSchemaExportSettings +{ + public string ExchangeName => rabbitOptions.Value.ExchangeName; + + public string FromQueueName => capOptions.Value.DefaultGroupName; + + public string System => + eventOptions.Value.MessageQueue.SchemaExportSettings?.System is { Length: > 0 } systemNameFromConfiguration + ? systemNameFromConfiguration + : IRabbitSchemaExportSettings.SystemEntryAssemblyName; + + public IReadOnlyDictionary InputEvents => eventTypeProvider.InputEvents.ToDictionary(x => x.Value, x => x.Key); + + public IReadOnlyDictionary OutputEvents => bindingsResolver.ResolveOutputEventsForExport(); +} diff --git a/src/Bss.Platform.Events/Internal/RawCapturingSerializer.cs b/src/Bss.Platform.Events/Internal/RawCapturingSerializer.cs new file mode 100644 index 0000000..89f711d --- /dev/null +++ b/src/Bss.Platform.Events/Internal/RawCapturingSerializer.cs @@ -0,0 +1,61 @@ +using System.Text; + +using DotNetCore.CAP; +using DotNetCore.CAP.Messages; +using DotNetCore.CAP.Serialization; + +using Microsoft.Extensions.Options; + +namespace Bss.Platform.Events.Internal; + +/// +/// Wraps the default CAP serializer to preserve the raw wire body in a message header before it gets +/// deserialized into the subscriber's parameter type (which may have fewer fields than the original message). +/// Stored as a header (not a process-local cache) so it survives retries and the DB-backed retry processor, +/// which can re-execute a failed message on any instance in a multi-pod deployment. +/// Also dead-letters messages that fail deserialization itself, since CAP never routes those into +/// - they are acked off the broker after a single attempt. +/// +internal sealed class RawCapturingSerializer(IOptions capOptions, DeadLetterProcessor deadLetterProcessor) : ISerializer +{ + internal const string RawBodyHeader = "x-cap-raw-body"; + + private readonly JsonUtf8Serializer inner = new(capOptions); + + public ValueTask SerializeAsync(Message message) => + this.inner.SerializeAsync(message); + + public async ValueTask DeserializeAsync(TransportMessage transportMessage, Type? valueType) + { + if (transportMessage.Body.Length > 0 && !transportMessage.Headers.ContainsKey(RawBodyHeader)) + { + transportMessage.Headers[RawBodyHeader] = Encoding.UTF8.GetString(transportMessage.Body.Span); + } + + Message message; + try + { + message = await this.inner.DeserializeAsync(transportMessage, valueType); + } + catch (Exception ex) + { + transportMessage.Headers.TryGetValue(RawBodyHeader, out var rawBody); + deadLetterProcessor.SendDeadLetter(transportMessage.GetName(), ex, rawBody); + throw; + } + + return message; + } + + public string Serialize(Message message) => + this.inner.Serialize(message); + + public Message? Deserialize(string json) => + this.inner.Deserialize(json); + + public object? Deserialize(object value, Type valueType) => + this.inner.Deserialize(value, valueType); + + public bool IsJsonType(object jsonObject) => + this.inner.IsJsonType(jsonObject); +} diff --git a/src/Bss.Platform.Events/Internal/WildcardMatcher.cs b/src/Bss.Platform.Events/Internal/WildcardMatcher.cs new file mode 100644 index 0000000..af212f4 --- /dev/null +++ b/src/Bss.Platform.Events/Internal/WildcardMatcher.cs @@ -0,0 +1,18 @@ +using System.Text.RegularExpressions; + +namespace Bss.Platform.Events.Internal; + +internal static class WildcardMatcher +{ + /// + /// Matches against , where * matches any substring.
+ /// Case-insensitive. A pattern without * requires an exact match. + /// + public static bool IsMatch(string value, string pattern) => + pattern.Contains('*') + ? Regex.IsMatch(value, ToRegexPattern(pattern), RegexOptions.IgnoreCase, TimeSpan.FromSeconds(1)) + : string.Equals(value, pattern, StringComparison.OrdinalIgnoreCase); + + private static string ToRegexPattern(string pattern) => + $"^{string.Join("[\\s\\S]*", pattern.Split('*').Select(Regex.Escape))}$"; +} diff --git a/src/Bss.Platform.Events/Models/ExternalSystemBindingsOptions.cs b/src/Bss.Platform.Events/Models/ExternalSystemBindingsOptions.cs new file mode 100644 index 0000000..4a3f19f --- /dev/null +++ b/src/Bss.Platform.Events/Models/ExternalSystemBindingsOptions.cs @@ -0,0 +1,17 @@ +namespace Bss.Platform.Events.Models; + +public class ExternalSystemBindingsOptions +{ + /// + /// Key is a queue name to declare, value is an explicit list of output-event routing keys to bind it to.
+ /// null or an empty array means the default set: all + /// except those matching . + /// + public Dictionary SystemBindings { get; set; } = new(); + + /// + /// Exact routing keys or *-masks excluded from the default set.
+ /// Applied only when a queue in has no explicit routing keys, to filter from all output events. + /// + public string[] ExcludeOutputEvents { get; set; } = []; +} diff --git a/src/Bss.Platform.Events/Models/IntegrationEventsMessageQueueOptions.cs b/src/Bss.Platform.Events/Models/IntegrationEventsMessageQueueOptions.cs index 5f1a5fa..a0f81c0 100644 --- a/src/Bss.Platform.Events/Models/IntegrationEventsMessageQueueOptions.cs +++ b/src/Bss.Platform.Events/Models/IntegrationEventsMessageQueueOptions.cs @@ -4,9 +4,12 @@ public class IntegrationEventsMessageQueueOptions { public bool Enable { get; set; } + // TODO: remove + public bool EnableSchemaExport => this.SchemaExportSettings != null; + public string Host { get; set; } = default!; - public int Port { get; set; } + public int Port { get; set; } = 5672; public string UserName { get; set; } = default!; @@ -15,4 +18,35 @@ public class IntegrationEventsMessageQueueOptions public string VirtualHost { get; set; } = default!; public string ExchangeName { get; set; } = default!; + + public string QueueName { get; set; } = default!; + + /// + /// RabbitMQ queue used by schema export initializer + /// to publish a one-time event schema payload during startup, + /// if null, then that mechanism is disabled + /// + public SchemaExportSettings? SchemaExportSettings { get; set; } = new(); + + /// + /// Provide a path to section satisfied or configure ExternalSystemBindingsOptions by yourself,
+ /// but the mapping dictionary has caveats (null and empty values skipped by default, using this parameter, you will avoid that) + /// + /// + /// Expected configuration: + /// + /// { + /// "ExternalSystemBindings": { + /// "ExcludeOutputEvents": ["EXT.Debug*", "EXT.Internal.SomeEvent"], + /// "SystemBindings": { + /// "system1-allEvents-queue-except-excluded": null, + /// "system2-allEvents-queue-except-excluded": [], + /// "system3-fixedEvents-queue-exclude-not-applied": ["EXT.OrderCreated", "EXT.OrderCancelled", "EXT.Debug.Some"] + /// } + /// } + /// } + /// + /// + + public string? ExternalSystemBindingsSectionPath { get; set; } } diff --git a/src/Bss.Platform.Events/Models/IntegrationEventsOptions.cs b/src/Bss.Platform.Events/Models/IntegrationEventsOptions.cs index ea700e8..cdd7288 100644 --- a/src/Bss.Platform.Events/Models/IntegrationEventsOptions.cs +++ b/src/Bss.Platform.Events/Models/IntegrationEventsOptions.cs @@ -1,19 +1,28 @@ +using DotNetCore.CAP; + using Microsoft.AspNetCore.Http; namespace Bss.Platform.Events.Models; public class IntegrationEventsOptions { - public string DashboardPath { get; set; } = default!; + public string DashboardPath { get; set; } = "/admin/events"; - public int FailedRetryCount { get; set; } + public string GatewayPrefix { get; set; } = string.Empty; - public int RetentionDays { get; set; } + public int FailedRetryCount { get; set; } = 5; - public IntegrationEventsSqlServerOptions SqlServer { get; set; } = default!; + public int RetentionDays { get; set; } = 15; + + /// + /// required to fill connection string in options, otherwise MS SQL db won't connect + /// + public IntegrationEventsSqlServerOptions SqlServer { get; set; } = new() { Schema = "events" }; + + public IntegrationEventsMessageQueueOptions MessageQueue { get; set; } = new() { Enable = true }; + + public Action? OverrideCapOptions { get; set; } - public IntegrationEventsMessageQueueOptions MessageQueue { get; set; } = default!; - /// /// Any condition to check that a user should get access to events dashboard /// @@ -21,14 +30,4 @@ public class IntegrationEventsOptions /// AuthorizationPolicyPredicate = (httpContext) => httpContext.RequestServices.GetRequiredService<ICurrentUser>().IsAdminAsync() /// public Func>? AuthorizationPredicate { get; set; } - - public static IntegrationEventsOptions Default => - new() - { - DashboardPath = "/admin/events", - FailedRetryCount = 5, - RetentionDays = 15, - SqlServer = new IntegrationEventsSqlServerOptions { Schema = "events" }, - MessageQueue = new IntegrationEventsMessageQueueOptions { Enable = true } - }; } diff --git a/src/Bss.Platform.Events/Models/IntegrationEventsSqlServerOptions.cs b/src/Bss.Platform.Events/Models/IntegrationEventsSqlServerOptions.cs index f38a179..8825cf3 100644 --- a/src/Bss.Platform.Events/Models/IntegrationEventsSqlServerOptions.cs +++ b/src/Bss.Platform.Events/Models/IntegrationEventsSqlServerOptions.cs @@ -2,6 +2,9 @@ namespace Bss.Platform.Events.Models; public class IntegrationEventsSqlServerOptions { + /// + /// When empty - MS SQL db won't connect + /// public string ConnectionString { get; set; } = default!; public string Schema { get; set; } = default!; diff --git a/src/Bss.Platform.Events/Models/RabbitIntegrationEventsOptions.cs b/src/Bss.Platform.Events/Models/RabbitIntegrationEventsOptions.cs new file mode 100644 index 0000000..d610d3d --- /dev/null +++ b/src/Bss.Platform.Events/Models/RabbitIntegrationEventsOptions.cs @@ -0,0 +1,17 @@ +namespace Bss.Platform.Events.Models; + +public class RabbitIntegrationEventsOptions : IntegrationEventsOptions +{ + /// + /// When enable will be registered CAP filter to handle failed event on the last attempt + deadlettering + /// Allow registering multiple scoped implementations of + /// IFailedEventProcessor<TInputEvent> + /// + public bool UseFailedEventProcessor { get; set; } = true; + + /// + /// set exchange and queue names for deadlettering (routing key for deadletter message always empty)
+ /// used only when is true + /// + public (string ExchangeName, string QueueName) DeadLetterOptions { get; set; } = ("deadletters", "deadletters"); +} diff --git a/src/Bss.Platform.Events/Models/SchemaExportSettings.cs b/src/Bss.Platform.Events/Models/SchemaExportSettings.cs new file mode 100644 index 0000000..d30781f --- /dev/null +++ b/src/Bss.Platform.Events/Models/SchemaExportSettings.cs @@ -0,0 +1,12 @@ +using System.Reflection; + +namespace Bss.Platform.Events.Models; + +public class SchemaExportSettings +{ + public string QueueName { get; set; } = "ToDocumenter"; + + public string RoutingKey { get; set; } = "RabbitEventSchemas"; + + public string System { get; set; } = string.Empty; +} diff --git a/src/Bss.Platform.Events/Publishers/IntegrationEventPublisher.cs b/src/Bss.Platform.Events/Publishers/IntegrationEventPublisher.cs index 1d0273c..95f60bb 100644 --- a/src/Bss.Platform.Events/Publishers/IntegrationEventPublisher.cs +++ b/src/Bss.Platform.Events/Publishers/IntegrationEventPublisher.cs @@ -1,19 +1,58 @@ using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Interfaces; using DotNetCore.CAP; namespace Bss.Platform.Events.Publishers; -public class IntegrationEventPublisher(ICapPublisher capPublisher, ICapTransaction capTransaction) : IIntegrationEventPublisher +public class IntegrationEventPublisherLegacy(ICapPublisher capPublisher, ICapTransaction capTransaction) + : IntegrationEventPublisherBase(capPublisher, capTransaction), IIntegrationEventPublisher { - public Task PublishAsync(IIntegrationEvent @event, CancellationToken cancellationToken) + private readonly ICapPublisher capPublisher = capPublisher; + + protected override Task PublishInternalAsync(IIntegrationEvent @event, CancellationToken cancellationToken) => + this.capPublisher.PublishAsync(@event.GetType().Name, @event, cancellationToken: cancellationToken); +} + +public class IntegrationEventPublisherNew(ICapPublisher capPublisher, ICapTransaction capTransaction, IEventTypeProvider eventTypeProvider) + : IntegrationEventPublisherBase(capPublisher, capTransaction) + where T: notnull +{ + private readonly ICapPublisher capPublisher = capPublisher; + + protected override async Task PublishInternalAsync(T @event, CancellationToken cancellationToken) + { + if (eventTypeProvider.InputEvents.TryGetValue(@event.GetType(), out var internalRoutingKey)) + { + await this.capPublisher.PublishAsync(internalRoutingKey, @event, cancellationToken: cancellationToken); + } + + if (eventTypeProvider.OutputEvents.TryGetValue(@event.GetType(), out var externalRoutingKey) + && externalRoutingKey != internalRoutingKey) + { + await this.capPublisher.PublishAsync(externalRoutingKey, @event, cancellationToken: cancellationToken); + } + + if (internalRoutingKey == null && externalRoutingKey == null) + { + throw new($"No routing key found for event type {@event.GetType().FullName}"); + } + } +} + +public abstract class IntegrationEventPublisherBase(ICapPublisher capPublisher, ICapTransaction capTransaction) + : IIntegrationEventPublisher +{ + public Task PublishAsync(T @event, CancellationToken cancellationToken) { if (capPublisher.Transaction is not null && capPublisher.Transaction != capTransaction) { - throw new Exception("There cannot be different CAP transactions within the same scope"); + throw new("There cannot be different CAP transactions within the same scope"); } capPublisher.Transaction = capTransaction; - return capPublisher.PublishAsync(@event.GetType().Name, @event, cancellationToken: cancellationToken); + return this.PublishInternalAsync(@event, cancellationToken); } + + protected abstract Task PublishInternalAsync(T @event, CancellationToken cancellationToken); } diff --git a/src/Bss.Platform.Events/README.md b/src/Bss.Platform.Events/README.md new file mode 100644 index 0000000..792ac2d --- /dev/null +++ b/src/Bss.Platform.Events/README.md @@ -0,0 +1,397 @@ +# Luxoft.Bss.Platform.Events + +Integration & domain events for microservices, built on top of [DotNetCore.CAP](https://github.com/dotnetcore/CAP) +(outbox pattern over RabbitMQ + SQL Server) and integrated with +[`Bss.Platform.Mediation`](../../README.md#custom-mediation). + +```shell +dotnet add package Luxoft.Bss.Platform.Events +``` + +> This document describes the changes introduced **after** the base version `1.6.8` (package `1.6.9`). +> For the general "getting started" guide see the [Events section](../../README.md#events) of the root README. + +## Table of contents + +- [What changed (overview)](#what-changed-overview) +- [Registration methods ( + `AddPlatformIntegrationEvents` overloads)](#registration-methods-addplatformintegrationevents-overloads) + - [1. Legacy — assembly scan](#1-legacy--assembly-scan) + - [2. Setup-based with `IIntegrationEvent` base](#2-setup-based-with-iintegrationevent-base) + - [3. Setup-based with a custom base type](#3-setup-based-with-a-custom-base-type) + - [4. Setup-based with separate input/output base types](#4-setup-based-with-separate-inputoutput-base-types) + - [Which overload to use](#which-overload-to-use) +- [Input vs output events](#input-vs-output-events) +- [Publishing events](#publishing-events) +- [Failed event processing](#failed-event-processing) +- [External system queue bindings](#external-system-queue-bindings) +- [Options](#options) +- [Breaking changes](#breaking-changes) +- [Switching to the new registration (queue name & `.v1`)](#switching-to-the-new-registration-queue-name--v1) +- [Other notes / things to review](#other-notes--things-to-review) + +## What changed (overview) + +The integration-events module was reworked to support explicit, per-type event registration and to produce +"external" events for other systems in addition to consuming its own. + +Highlights: + +- **New setup-based registration** via `Action>` — you now list the event types + (and their routing keys) explicitly instead of relying only on an assembly scan. +- **Input vs output events** — a distinction between events that are *consumed & handled* inside the service + (`Input`) and events that are only *produced* for other systems (`Output`). See + [`IEventTypeProvider`](../Bss.Platform.Events.Abstractions/IEventTypeProvider.cs). +- **Generic publisher / processor** — `IIntegrationEventPublisher` and `IIntegrationEventProcessor`; + the old non-generic interfaces now inherit from the `IIntegrationEvent` closed versions. +- **Failed event processing** — opt-in CAP filter (`UseFailedEventProcessor`) that invokes any number of + `IFailedEventProcessor` implementations on the last retry. +- **Custom RabbitMQ headers** — messages produced by the new selector carry `MessageId` (snowflake), + `MessageName` (routing key) and `Type` headers so consumers built outside of CAP can read them. +- **Options cleanup & new options** — `IntegrationEventsOptions.Default` removed in favour of inline defaults; + new `GatewayPrefix`, `QueueName`, `OverrideCapOptions`, `UseFailedEventProcessor`; `SqlServer.ConnectionString` + and RabbitMQ are now optional (fallback to no-DB / in-memory queue), default RabbitMQ `Port` is `5672`. +- **Queue-name behaviour change** — the new (multi-generic) overloads no longer append `.v1` to the queue name. + This is the most important behavioural change — see + [Switching to the new registration](#switching-to-the-new-registration-queue-name--v1). + +## Registration methods (`AddPlatformIntegrationEvents` overloads) + +There are now four overloads. All of them ultimately register CAP (SQL Server outbox + RabbitMQ or in-memory), +the dashboard and a publisher; they differ in **how event types are discovered** and **which base type / publisher** +is used. + +### 1. Legacy — assembly scan + +```C# +services.AddPlatformIntegrationEvents( + Assembly eventsAssembly, + Action? setup = null); +// where TEventProcessor : class, IIntegrationEventProcessor +``` + +- Scans `eventsAssembly` for every non-abstract type assignable to `IIntegrationEvent` and subscribes to each + using the **type name** as the routing key. +- `TEventProcessor` is the non-generic `IIntegrationEventProcessor` (handles `IIntegrationEvent`). +- Registers the non-generic legacy `IIntegrationEventPublisher`. +- **Appends `.v1` to the queue name** to preserve the historical CAP group name. + +Use this when upgrading an existing service and you don't want to change anything else. + +### 2. Setup-based with `IIntegrationEvent` base + +```C# +services.AddPlatformIntegrationEvents( + Action> setupEvents, + Action setupOptions); +// where TEventProcessor : class, IIntegrationEventProcessor +``` + +- Same base type (`IIntegrationEvent`) as the legacy method, but you register the events **explicitly** via the + setup action (see [Input vs output events](#input-vs-output-events)). +- Registers the legacy `IIntegrationEventPublisher` for backward compatibility. + +Use this when you want the new explicit registration / output events but keep `IIntegrationEvent` as your base. + +### 3. Setup-based with a custom base type + +```C# +services.AddPlatformIntegrationEvents( + Action> setupEvents, + Action setupOptions); +// where TEventProcessor : class, IIntegrationEventProcessor +// where TEvent : notnull +``` + +- Your own base type `TEvent` for both consumed and produced events (input == output). +- Registers `IIntegrationEventPublisher`. +- **Does not append `.v1`** to the queue name. + +Use this when your events do not implement `IIntegrationEvent` and input/output share one base type. + +### 4. Setup-based with separate input/output base types + +```C# +services.AddPlatformIntegrationEvents( + Action> setupEvents, + Action setupOptions); +// where TEventProcessor : class, IIntegrationEventProcessor +// where TInputEvent : notnull +// where TOutputEvent : notnull +``` + +- The most flexible overload. `TInputEvent` is the base of events **consumed & handled** by this service; + `TOutputEvent` is the base of events **only produced** for other systems. +- Registers `IIntegrationEventPublisher` (wrap it if you need a narrower contract). +- **Does not append `.v1`** to the queue name. +- Overloads (3) and (2) both delegate to this one. + +Use this when produced and consumed events have different base contracts. + +### Which overload to use + +| Situation | Overload | +|----------------------------------------------------------------------|------------------------------| +| Just upgrading the package, no code changes wanted | **(1)** Legacy assembly scan | +| Want explicit registration / output events, keep `IIntegrationEvent` | **(2)** | +| Custom base type, input == output | **(3)** | +| Different base types for consumed vs produced events | **(4)** | + +## Input vs output events + +The setup action exposes [`IIntegrationEventSetup`](Interfaces/IIntegrationEventSetup.cs): + +```C# +services.AddPlatformIntegrationEvents( + events => events + // consumed & handled inside this service (Rabbit -> CAP -> processor): + .AddInputEvents("INT.") // scan assembly of IMyInputEvent, routing key "INT." + TypeName + .AddInputEvent("custom.routing.key") // single type, explicit routing key (overrides scan) + // only produced for other systems (published to the Rabbit exchange, no local handler): + .AddOutputEvents("EXT.", typeof(SomeOtherEvent).Assembly) + .AddOutputEvent("some.routing.key"), + options => + { + options.SqlServer.ConnectionString = "..."; + options.MessageQueue.ExchangeName = "integration.events"; + options.MessageQueue.Host = "..."; + // ... + }); +``` + +- **Input events** are subscribed in CAP and dispatched to `IIntegrationEventProcessor` + (a single processor instance is reused for every input type). +- **Output events** are not subscribed; they only get a routing key so the publisher can emit them. +- `AddInputEvents` / `AddOutputEvents` scan the assembly containing the base type (or the assemblies you pass) + and build the routing key as `prefix + TypeName`. `AddInputEvent` / `AddOutputEvent` register a single type + with an explicit routing key and **override** any entry added by the scan. + +## Publishing events + +```C# +public class Handler(IIntegrationEventPublisher publisher) : IRequestHandler +{ + public Task Handle(Command request, CancellationToken ct) => + publisher.PublishAsync(new MyEvent(), ct); +} +``` + +- The new publisher (`IntegrationEventPublisherNew`) resolves the routing key from the type provider: + if the type is registered as input it is published to the internal routing key, and if it is also registered + as output (with a different key) it is additionally published to the external routing key. If the type is + registered in neither, `PublishAsync` throws. +- The legacy publisher (`IntegrationEventPublisherLegacy`, exposed as `IIntegrationEventPublisher`) keeps the old + behaviour of publishing with the routing key equal to the event's type name. + +## Failed event processing + +Enable it via options and register one or more `IFailedEventProcessor` implementations: + +```C# +options.UseFailedEventProcessor = true; + +services.AddScoped(); +``` + +```C# +public interface IFailedEventProcessor +{ + Task HandleAsync(object? value, Exception ex); +} +``` + +- When enabled, a CAP `ISubscribeFilter` (`CapExceptionFilter`) is registered. +- It fires **only on the last retry** (accounting for the `0`-retries case) and invokes **all** registered + `IFailedEventProcessor` instances. +- The failed payload is deserialized to the handler's parameter type when the CAP value is JSON; otherwise the raw + value is passed. Failure details (`ExceptionType`, `Message`, `StackTrace`) are also written to the + `x-cap-failure-details` message header. + +## External system queue bindings + +Output events (see [Input vs output events](#input-vs-output-events)) are published to the exchange, but by default +nothing declares queues/bindings for external subscribers — someone has to create the queue and bind it to the +routing keys they care about. `ExternalSystemBindingsOptions` lets you declare N queues (one per external system) +and have them bound automatically at startup, with per-queue overrides and mass exclusion. + +The feature is **opt-in**: it does nothing unless enabled via `MessageQueue.EnableExternalSystemBindings(...)`. It +also requires an `IEventTypeProvider` (i.e. one of the setup-based registration overloads (2)-(4), not the legacy +assembly-scan one) — `ExternalSystemQueueBindingsInitializer` takes it as a mandatory dependency, so enabling the +feature on the legacy overload fails fast at startup with a standard DI resolution error instead of doing nothing silently. + +```C# +options.MessageQueue.EnableExternalSystemBindings(); // binds from the default section "RabbitCap:ExternalSystemBindings" +// or bind from a different section: +options.MessageQueue.EnableExternalSystemBindings("MyApp:ExternalSystemBindings"); +``` + +Pass an empty string explicitly (`EnableExternalSystemBindings("")`) to register the pipeline **without** binding +`ExternalSystemBindingsOptions` from configuration at all — use this if you configure `SystemBindings` entirely in +code (see [Overriding a specific system in code](#overriding-a-specific-system-in-code-with-di)). + +```json +{ + "RabbitCap": { + "ExternalSystemBindings": { + "ExcludeOutputEvents": ["EXT.Debug*", "EXT.Internal.SomeEvent"], + "SystemBindings": { + "crm-queue": null, + "billing-queue": [], + "analytics-queue": ["EXT.OrderCreated", "EXT.OrderCancelled"] + } + } + } +} +``` + +- `SystemBindings` — key is the queue name to declare, value is the list of output-event routing keys to bind it to. +- `null` or `[]` → the queue gets the **default set**: every registered output event, except those matching + `ExcludeOutputEvents` (exact match or `*`-mask, e.g. `"EXT.Debug*"`). +- A non-empty array is an **explicit override** — bound exactly as listed, ignoring `ExcludeOutputEvents` entirely. + It is not validated against the registered output events (you can list keys that don't exist in `OutputEvents`, + e.g. ones published from another service into the same exchange). + +### Overriding a specific system in code (with DI) + +`ExternalSystemBindingsOptions` is its own `IOptions`, so you can layer a code-based, DI-aware override on top of +whatever came from configuration — useful when one system's binding list depends on a service you already registered: + +```C# +services.AddOptions() + .PostConfigure((options, catalog) => + { + options.SystemBindings["billing-queue"] = catalog.GetBillingRoutingKeys(); + }); +``` + +`PostConfigure` always runs after configuration-based binding, regardless of registration order, and supports +injecting up to five dependencies via the `PostConfigure` overloads. + +### How the queues get declared + +At startup, `RabbitInitializersHostedService` rents a single RabbitMQ channel (the same `IConnectionChannelPool` +`DeadLetterProcessor` uses) and runs every registered `IRabbitInitializer.InitializeAsync(IModel, CancellationToken)` against it — +`ExternalSystemQueueBindingsInitializer` is the one that declares queues from `SystemBindings` and binds them. +Each initializer runs in isolation: if `InitializeAsync` itself throws (e.g. a `QueueDeclare` conflict with an +existing queue), it is logged as an `ERROR` with the initializer's type name, but it does **not** stop the other +initializers or fail application startup. This is different from a missing `IEventTypeProvider` (see above), which +fails during DI graph construction — before any initializer runs — and is not caught by this try/catch. You can +register your own `IRabbitInitializer` implementations the same way +(e.g. `services.AddSingleton()`) for other one-time RabbitMQ topology setup. + +## Rabbit event schema export + +Enable via `MessageQueue.EnableSchemaExport = true`. + +At startup, `RabbitEventSchemaExportInitializer` publishes a single JSON payload to +`MessageQueue.SchemaExportExchangeName` (fanout) / `MessageQueue.SchemaExportQueueName`: + +```json +{ + "output": { "...": "json schema object" }, + "input": { "...": "json schema object" }, + "systemName": "string", + "exchange": "string", + "queue": "string" +} +``` + +- `input` is generated from `IEventTypeProvider.InputEvents` as-is. +- `output` is generated from `IEventTypeProvider.OutputEvents`, filtered by + `ExternalSystemBindingsOptions.ExcludeOutputEvents` using the same wildcard matching rules as + `ExternalSystemQueueBindingsInitializer`. +- `systemName` uses `MessageQueue.SchemaExportSystemName`, or falls back to the effective consumer queue + (`MessageQueue.QueueName` if set, otherwise `MessageQueue.ExchangeName`). + +## Options + +[`IntegrationEventsOptions`](Models/IntegrationEventsOptions.cs): + +| Option | Description | Type | Default | +|--------------------------------|---------------------------------------------------------------------------------------------------------------------|----------------------------------|-----------------| +| **DashboardPath** | Dashboard relative path | string | `/admin/events` | +| **GatewayPrefix** | `PathBase` for the dashboard (when hosted behind a gateway prefix) | string | `""` | +| **FailedRetryCount** | Number of message retries | int | `5` | +| **RetentionDays** | Successful message retention period | int | `15` | +| **SqlServer.ConnectionString** | MS SQL connection string. **When empty, SQL Server storage is not configured**, but you can provide EF.Core storage | string | *(empty)* | +| **SqlServer.Schema** | Schema for event tables | string | `events` | +| **MessageQueue.Enable** | Dev only. When `false`, uses the in-memory queue instead of RabbitMQ | bool | `true` | +| **MessageQueue.EnableSchemaExport** | Publishes one startup schema payload (`input`/`output` JSON schemas + `systemName`/`exchange`/`queue`) to RabbitMQ | bool | `false` | +| **MessageQueue.Port** | RabbitMQ port | int | `5672` | +| **MessageQueue.QueueName** | Explicit CAP group / queue name (overrides the exchange-name default) | string | *(unset)* | +| **MessageQueue.SchemaExportExchangeName** | Fanout exchange used for schema export payload publication | string | `events.schema.export` | +| **MessageQueue.SchemaExportQueueName** | Queue declared and bound to `SchemaExportExchangeName` for schema export payloads | string | `events.schema.export` | +| **MessageQueue.SchemaExportSystemName** | `systemName` field in schema payload; when empty, effective queue name is used | string | `""` | +| **MessageQueue.EnableExternalSystemBindings(sectionPath)** | Opts into external system queue bindings; binds `ExternalSystemBindingsOptions` from `sectionPath` (pass `""` to configure purely in code). See [External system queue bindings](#external-system-queue-bindings) | method | not called (feature disabled); `sectionPath` defaults to `"RabbitCap:ExternalSystemBindings"` | +| **UseFailedEventProcessor** | Register the CAP filter that calls `IFailedEventProcessor` on the last retry | bool | `false` | +| **OverrideCapOptions** | Escape hatch to mutate `CapOptions` directly (applied last) | `Action?` | `null` | +| **AuthorizationPredicate** | Predicate controlling access to the events dashboard | `Func>?` | `null` | + +## Breaking changes + +1. **`IntegrationEventsOptions.Default` removed.** Defaults are now applied directly on the properties. If you + referenced `IntegrationEventsOptions.Default`, drop it — a plain `new IntegrationEventsOptions()` is already + populated. +2. **`IIntegrationEventPublisher` is now `IIntegrationEventPublisher`.** The non-generic + interface still exists (as a derived marker), so injecting `IIntegrationEventPublisher` keeps working. Publishing + custom base types uses `IIntegrationEventPublisher`. +3. **`IIntegrationEventProcessor` is now `IIntegrationEventProcessor`.** Existing non-generic + implementations still compile. +4. **`IntegrationEventPublisher` renamed** to `IntegrationEventPublisherLegacy` (+ new `IntegrationEventPublisherNew` + and `IntegrationEventPublisherBase`). Breaking only if you referenced the concrete class. +5. **`CapConsumerServiceSelector` renamed & moved.** It is now `CapConsumerServiceSelectorLegacy` / + `CapConsumerServiceSelectorNew` in the `Bss.Platform.Events.Internal` namespace. Breaking only if you referenced it. +6. **Queue name no longer gets `.v1`** with the new (multi-generic) overloads — see the next section. This changes + the RabbitMQ queue a service binds to, so it is a runtime/behavioural breaking change even though it compiles. +7. **`IEventTypeProvider` moved** to `Bss.Platform.Events.Abstractions` (it was briefly under + `Bss.Platform.Events.Interfaces`). +8. **New dependency:** the package now references `Bss.Platform.Mediation.Abstractions` + (`IIntegrationEvent : INotification`). +9. **`LangVersion` raised to `14`** in `Directory.Build.props`. + +## Switching to the new registration (queue name & `.v1`) + +> **Read this before moving from the legacy overload to the new setup-based overloads.** + +Historically the CAP consumer group / RabbitMQ queue name was `"{ExchangeName}.v1"`. To preserve that, +the **legacy** overload automatically appends `.v1` to the queue name (via `SetLegacyQueueNameWithVersion`): + +- **Legacy overload (assembly scan):** queue name = `MessageQueue.QueueName` if set, otherwise `"{ExchangeName}.v1"`. +- **New overloads (custom / input-output generics):** queue name = `MessageQueue.QueueName` if set, otherwise + `"{ExchangeName}"` — **no `.v1` suffix**. + +> Note: currently the `.v1` behaviour is applied by two overloads (the legacy assembly-scan one **and** the +> setup-based `IIntegrationEvent` overload). This is being consolidated so that only the **legacy** overload — the +> one that does **not** take a `setupEvents` action — keeps the `.v1` behaviour, purely for drop-in upgrade +> compatibility. + +### If you want to keep the old queue name + +When you move to a new overload but need to keep binding to the existing `"{ExchangeName}.v1"` queue, set the queue +name explicitly: + +```C# +services.AddPlatformIntegrationEvents( + events => events.AddInputEvents(), + options => + { + options.MessageQueue.ExchangeName = "integration.events"; + // keep the historical queue name so we don't create/bind a new queue: + options.MessageQueue.QueueName = $"{options.MessageQueue.ExchangeName}.v1"; + // ... + }); +``` + +If you intentionally want a fresh queue (new naming), simply leave `QueueName` unset and the queue will be +`"{ExchangeName}"`. + +## Other notes / things to review + +- `IIntegrationEventSetup.AddOutputEvents` has no default value for `prefix` while `AddInputEvents` does + (`prefix = ""`) — a minor API inconsistency. +- In `EventTypeProvider`, both `AddInputEvents` and `AddOutputEvents` deduplicate with + `.Except(this.outputTypes.Keys)`. For output registration this excludes types against *itself* (not against + input types), so a type registered as both input and output is allowed — which the publisher relies on (it emits + to both routing keys when they differ). Worth a quick confirmation that this is intended. +- `SqlServer.ConnectionString` empty now silently means "no SQL Server storage" — make sure services that expect + the outbox actually set it. diff --git a/src/Bss.Platform.RabbitMq.Consumer/Bss.Platform.RabbitMq.Consumer.csproj b/src/Bss.Platform.RabbitMq.Consumer/Bss.Platform.RabbitMq.Consumer.csproj index 9bd902b..8231b79 100644 --- a/src/Bss.Platform.RabbitMq.Consumer/Bss.Platform.RabbitMq.Consumer.csproj +++ b/src/Bss.Platform.RabbitMq.Consumer/Bss.Platform.RabbitMq.Consumer.csproj @@ -3,6 +3,7 @@ Luxoft.Bss.Platform.RabbitMq.Consumer + diff --git a/src/Bss.Platform.RabbitMq.Consumer/DependencyInjection.cs b/src/Bss.Platform.RabbitMq.Consumer/DependencyInjection.cs index 01c9a5a..ae78d0c 100644 --- a/src/Bss.Platform.RabbitMq.Consumer/DependencyInjection.cs +++ b/src/Bss.Platform.RabbitMq.Consumer/DependencyInjection.cs @@ -4,6 +4,7 @@ using Bss.Platform.RabbitMq.Consumer.Internal; using Bss.Platform.RabbitMq.Consumer.Services; using Bss.Platform.RabbitMq.Consumer.Settings; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -12,8 +13,8 @@ namespace Bss.Platform.RabbitMq.Consumer; public static class DependencyInjection { - public const string RoutingMessageProviderKey = nameof(RoutingMessageProviderKey); - + public const string RoutingConsumedMessagesProviderKey = nameof(RoutingConsumedMessagesProviderKey); + public static IServiceCollection AddPlatformRabbitMqSqlServerConsumerLock(this IServiceCollection services, string connectionString) => services .AddSingleton() @@ -38,6 +39,8 @@ public static IServiceCollection AddPlatformRabbitMqConsumer( } return services + .AddSingleton() + .AddKeyedSingleton(ExportEventsSchemaInitializer.SettingsKey) .Configure(consumerSettingsSection) .AddSingleton() .AddSingleton() @@ -60,7 +63,7 @@ public static IServiceCollection AddPlatformRabbitMqConsumerWithMessages(configuration, internalBuilder.RegisteredMessages); } - + /// /// Add consumer with default serialization (case in-sensitive), and find and register events marked by attribute /// @@ -76,6 +79,37 @@ public static IServiceCollection AddPlatformRabbitMqConsumerWithMessages(configuration, messages); } + /// + /// Allowed to register types for input/output events and system name, only provided values for exporter + /// and override autoregistered input types, but the exporter registered in + ///
    • + ///
    • + ///
    • + ///     + public static IServiceCollection RegisterRabbitTypesForExporter( + this IServiceCollection services, + string? systemName = null, + IReadOnlyDictionary? inputEvents = null, + IReadOnlyDictionary? outputEvents = null) + { + if (systemName != null) + { + services.AddKeyedSingleton(ExportEventsSchemaInitializer.RabbitSchemaExportSettings.SystemNameKey, systemName); + } + + if (inputEvents != null) + { + services.AddKeyedSingleton(ExportEventsSchemaInitializer.RabbitSchemaExportSettings.InputEventTypeKey, inputEvents); + } + + if (outputEvents != null) + { + services.AddKeyedSingleton(ExportEventsSchemaInitializer.RabbitSchemaExportSettings.OutputEventTypeKey, outputEvents); + } + + return services; + } + private static IServiceCollection AddRabbitAndEvents( this IServiceCollection services, IConfiguration configuration, @@ -93,9 +127,8 @@ private static IServiceCollection AddRabbitAndEvents( $"Unexpected message type '{x.MessageType.Name}' with routing key '{x.RoutingKey}', allow only {typeof(TEvent).Name} and its subtypes"), StringComparer.OrdinalIgnoreCase); - services.AddKeyedSingleton(RoutingMessageProviderKey, routeMessages); - services.PostConfigure( - opts => + services.AddKeyedSingleton(RoutingConsumedMessagesProviderKey, routeMessages); + services.PostConfigure(opts => { if (opts.RoutingKeys.Length > 0) { diff --git a/src/Bss.Platform.RabbitMq.Consumer/Interfaces/IRabbitMqMessageProcessor.cs b/src/Bss.Platform.RabbitMq.Consumer/Interfaces/IRabbitMqMessageProcessor.cs index bbf41b5..76ae9af 100644 --- a/src/Bss.Platform.RabbitMq.Consumer/Interfaces/IRabbitMqMessageProcessor.cs +++ b/src/Bss.Platform.RabbitMq.Consumer/Interfaces/IRabbitMqMessageProcessor.cs @@ -1,8 +1,12 @@ -using RabbitMQ.Client; +using System.Text.Json; + +using RabbitMQ.Client; namespace Bss.Platform.RabbitMq.Consumer.Interfaces; public interface IRabbitMqMessageProcessor { Task ProcessAsync(IBasicProperties properties, string routingKey, string message, CancellationToken token); + + protected static readonly JsonSerializerOptions CaseInsensitiveJsonSerializerOptions = new() { PropertyNameCaseInsensitive = true }; } diff --git a/src/Bss.Platform.RabbitMq.Consumer/Services/ExportEventsSchemaInitializer.cs b/src/Bss.Platform.RabbitMq.Consumer/Services/ExportEventsSchemaInitializer.cs new file mode 100644 index 0000000..4f6887f --- /dev/null +++ b/src/Bss.Platform.RabbitMq.Consumer/Services/ExportEventsSchemaInitializer.cs @@ -0,0 +1,55 @@ +using System.Collections.ObjectModel; + +using Bss.Platform.RabbitMq.Consumer.Interfaces; +using Bss.Platform.RabbitMq.Consumer.Settings; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +using RabbitMQ.Client; + +namespace Bss.Platform.RabbitMq.Consumer.Services; + +public sealed class ExportEventsSchemaInitializer( + [FromKeyedServices(ExportEventsSchemaInitializer.SettingsKey)] + IRabbitSchemaExportSettings settings) + : IRabbitMqInitializer +{ + public const string SettingsKey = $"{nameof(ExportEventsSchemaInitializer)}.{nameof(SettingsKey)}"; + + public void Initialize(IModel model) + { + new RabbitEventsSchemaExporter(settings).Export(model); + } + + internal sealed class RabbitSchemaExportSettings( + IOptions options, + [FromKeyedServices(RabbitSchemaExportSettings.InputEventTypeKey)] + Dictionary? inputTypes = null, + [FromKeyedServices(DependencyInjection.RoutingConsumedMessagesProviderKey)] + Dictionary? inputAutoTypes = null, + [FromKeyedServices(RabbitSchemaExportSettings.OutputEventTypeKey)] + Dictionary? outputTypes = null, + [FromKeyedServices(RabbitSchemaExportSettings.SystemNameKey)] + string? systemName = null) : IRabbitSchemaExportSettings + { + public const string InputEventTypeKey = $"{nameof(RabbitSchemaExportSettings)}.{nameof(InputEventTypeKey)}"; + + public const string OutputEventTypeKey = $"{nameof(RabbitSchemaExportSettings)}.{nameof(OutputEventTypeKey)}"; + + public const string SystemNameKey = $"{nameof(RabbitSchemaExportSettings)}.{nameof(SystemNameKey)}"; + + public string FromQueueName => options.Value.Queue; + + public string ExchangeName => options.Value.Exchange; + + public string System => systemName ?? IRabbitSchemaExportSettings.SystemEntryAssemblyName; + + public bool IsEnabled => inputTypes != null || inputAutoTypes != null || outputTypes != null; + + public IReadOnlyDictionary InputEvents => inputTypes ?? inputAutoTypes ?? []; + + public IReadOnlyDictionary OutputEvents => outputTypes ?? []; + } +} diff --git a/src/Bss.Platform.RabbitMq.Consumer/Services/RabbitMqMessageProcessor.cs b/src/Bss.Platform.RabbitMq.Consumer/Services/RabbitMqMessageProcessor.cs index fe83ba7..558d51e 100644 --- a/src/Bss.Platform.RabbitMq.Consumer/Services/RabbitMqMessageProcessor.cs +++ b/src/Bss.Platform.RabbitMq.Consumer/Services/RabbitMqMessageProcessor.cs @@ -12,11 +12,9 @@ namespace Bss.Platform.RabbitMq.Consumer.Services; internal class RabbitMqMessageProcessor( IRabbitMqEventProcessor rabbitEventProcessor, ILogger> logger, - [FromKeyedServices(DependencyInjection.RoutingMessageProviderKey)] + [FromKeyedServices(DependencyInjection.RoutingConsumedMessagesProviderKey)] Dictionary registeredHandlers) : IRabbitMqMessageProcessor { - private static readonly JsonSerializerOptions CaseInsensitiveJsonSerializerOptions = new() { PropertyNameCaseInsensitive = true }; - public Task ProcessAsync(IBasicProperties properties, string routingKey, string message, CancellationToken token) { if (!registeredHandlers.TryGetValue(routingKey, out var handlerType)) @@ -26,7 +24,7 @@ public Task ProcessAsync(IBasicProperties properties, string routingKey, string throw new InvalidOperationException(error.Replace("{RoutingKey}", routingKey)); } - var request = JsonSerializer.Deserialize(message, handlerType, CaseInsensitiveJsonSerializerOptions); + var request = JsonSerializer.Deserialize(message, handlerType, IRabbitMqMessageProcessor.CaseInsensitiveJsonSerializerOptions); if (request is null) { const string error = "The request with routing key '{RoutingKey}' could not be deserialized."; diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/DependencyInjection.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/DependencyInjection.cs index 4376982..6633048 100644 --- a/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/DependencyInjection.cs +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/DependencyInjection.cs @@ -5,6 +5,8 @@ namespace Bss.Platform.RabbitMq.JsonSchemaGenerator; public static class DependencyInjection { + // TODO: update obsolete message (link to readme?) + [Obsolete("Use new consumer based or provide RabbitEventsSchemaExporter instead")] public static IApplicationBuilder UseRabbitJsonSchemaGenerator( this IApplicationBuilder app, Action? setup = null) @@ -14,7 +16,7 @@ public static IApplicationBuilder UseRabbitJsonSchemaGenerator( var consumedEvents = app.ApplicationServices - .GetKeyedService>(Consumer.DependencyInjection.RoutingMessageProviderKey) + .GetKeyedService>(Consumer.DependencyInjection.RoutingConsumedMessagesProviderKey) ?.Select(x => (x.Key, x.Value)) ?? []; diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/GenerateSchemaMiddleware.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/GenerateSchemaMiddleware.cs index 2a37277..9ac1b42 100644 --- a/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/GenerateSchemaMiddleware.cs +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGenerator/GenerateSchemaMiddleware.cs @@ -1,7 +1,6 @@ -using Microsoft.AspNetCore.Http; +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; -using NJsonSchema; -using NJsonSchema.Generation; +using Microsoft.AspNetCore.Http; namespace Bss.Platform.RabbitMq.JsonSchemaGenerator; @@ -12,43 +11,14 @@ public async Task InvokeAsync(HttpContext context) if (context.Request.Method == "GET" && context.Request.Path.Value.Equals(path, StringComparison.InvariantCultureIgnoreCase)) { - await this.GenerateSchema(context); + var schemaContainer = new RabbitEventsSchemaGenerator().GenerateSchema(eventsDict); + + context.Response.StatusCode = 200; + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync(schemaContainer.ToJson()); return; } await next(context); } - - private async Task GenerateSchema(HttpContext context) - { - var settings = new SystemTextJsonSchemaGeneratorSettings - { - FlattenInheritanceHierarchy = true, - GenerateAbstractProperties = false, - AllowReferencesWithProperties = false - }; - - var schemaContainer = new JsonSchema(); - var appender = new JsonSchemaAppender(schemaContainer, new MappedNameGenerator(eventsDict)); - var generator = new NJsonSchema.Generation.JsonSchemaGenerator(settings); - - var jsonSchemas = eventsDict.Select(x => x.Value).Select(generator.Generate); - foreach (var schema in jsonSchemas) - { - appender.AppendSchema(schema, null); - } - - context.Response.StatusCode = 200; - context.Response.ContentType = "application/json"; - await context.Response.WriteAsync(schemaContainer.ToJson()); - } -} - -file class MappedNameGenerator(Dictionary eventsDict) : ITypeNameGenerator -{ - private readonly Dictionary mapping = eventsDict.DistinctBy(x => x.Value) - .ToDictionary(x => x.Value.Name, x => x.Key); - - public string Generate(JsonSchema schema, string? typeNameHint, IEnumerable reservedTypeNames) => - this.mapping.GetValueOrDefault(schema.Title ?? throw new("JsonSchema title is null"), schema.Title); } diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/AssemblyInfo.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/AssemblyInfo.cs new file mode 100644 index 0000000..56d4936 --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Tests.Unit")] diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase.csproj b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase.csproj new file mode 100644 index 0000000..2cb991e --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase.csproj @@ -0,0 +1,10 @@ + + + Luxoft.Bss.Platform.RabbitMq.JsonSchemaGeneratorBase + Bss.Platform.RabbitMq.JsonSchemaGeneratorBase + + + + + + diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/IRabbitSchemaExportSettings.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/IRabbitSchemaExportSettings.cs new file mode 100644 index 0000000..301b771 --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/IRabbitSchemaExportSettings.cs @@ -0,0 +1,39 @@ +using System.Reflection; + +namespace Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +public interface IRabbitSchemaExportSettings +{ + string RoutingKey => "RabbitEventSchemas"; + + string ToQueueName => "ToDocumenter"; + + string System => SystemEntryAssemblyName; + + bool IsEnabled => true; + + static string SystemEntryAssemblyName => Assembly.GetEntryAssembly()?.GetName().Name ?? string.Empty; + + string FromQueueName { get; } + + string ExchangeName { get; } + + IReadOnlyDictionary InputEvents { get; } + + IReadOnlyDictionary OutputEvents { get; } +} + +public abstract class RabbitSchemaExportSettingsBase( + //[FromKeyedServices(InputEventTypeKey)] + Dictionary? consumedTypes = null) : IRabbitSchemaExportSettings +{ + public const string InputEventTypeKey = nameof(InputEventTypeKey); + + public abstract string FromQueueName { get; } + + public abstract string ExchangeName { get; } + + public IReadOnlyDictionary InputEvents => consumedTypes ?? []; + + public IReadOnlyDictionary OutputEvents => throw new NotImplementedException(); +} diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/MappedNameGenerator.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/MappedNameGenerator.cs new file mode 100644 index 0000000..996654a --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/MappedNameGenerator.cs @@ -0,0 +1,12 @@ +using NJsonSchema; + +namespace Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +internal class MappedNameGenerator(IReadOnlyDictionary eventsDict) : ITypeNameGenerator +{ + private readonly Dictionary mapping = eventsDict.DistinctBy(x => x.Value) + .ToDictionary(x => x.Value.Name, x => x.Key); + + public string Generate(JsonSchema schema, string? typeNameHint, IEnumerable reservedTypeNames) => + this.mapping.GetValueOrDefault(schema.Title ?? throw new("JsonSchema title is null"), schema.Title); +} diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaExporter.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaExporter.cs new file mode 100644 index 0000000..de550a9 --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaExporter.cs @@ -0,0 +1,62 @@ +using System.Text; +using System.Text.Json; + +using RabbitMQ.Client; + +namespace Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +public sealed class RabbitEventsSchemaExporter(IRabbitSchemaExportSettings settings) +{ + public void Export(IModel model) + { + if (!settings.IsEnabled) + { + return; + } + + var exportExchange = settings.ExchangeName; + var exportQueue = settings.ToQueueName; + + model.QueueDeclare(exportQueue, true, false, false, null); + model.QueueBind(exportQueue, exportExchange, settings.RoutingKey); + + var properties = model.CreateBasicProperties(); + properties.DeliveryMode = 2; + properties.ContentType = "application/json"; + + var payloadJson = this.BuildExportPayloadJson(); + model.BasicPublish(exportExchange, settings.RoutingKey, properties, Encoding.UTF8.GetBytes(payloadJson)); + if (model.NextPublishSeqNo > 0) + { + model.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5)); + } + } + + internal string BuildExportPayloadJson() + { + var inputSchema = GenerateSchemaJson(settings.InputEvents); + var outputSchema = GenerateSchemaJson(settings.OutputEvents); + + using var inputDocument = JsonDocument.Parse(inputSchema); + using var outputDocument = JsonDocument.Parse(outputSchema); + + var payload = new + { + output = outputDocument.RootElement.Clone(), + input = inputDocument.RootElement.Clone(), + systemName = settings.System, + exchange = settings.ExchangeName, + queue = settings.FromQueueName + }; + + return JsonSerializer.Serialize(payload); + } + + private static string GenerateSchemaJson(IReadOnlyDictionary eventsDict) + { + var schemaGenerator = new RabbitEventsSchemaGenerator(); + var schemaContainer = schemaGenerator.GenerateSchema(eventsDict); + + return schemaContainer.ToJson(); + } +} diff --git a/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaGenerator.cs b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaGenerator.cs new file mode 100644 index 0000000..20c0f30 --- /dev/null +++ b/src/Bss.Platform.RabbitMq.JsonSchemaGeneratorBase/RabbitEventsSchemaGenerator.cs @@ -0,0 +1,29 @@ +using NJsonSchema; +using NJsonSchema.Generation; + +namespace Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +public class RabbitEventsSchemaGenerator +{ + public JsonSchema GenerateSchema(IReadOnlyDictionary eventsDict) + { + var settings = new SystemTextJsonSchemaGeneratorSettings + { + FlattenInheritanceHierarchy = true, + GenerateAbstractProperties = false, + AllowReferencesWithProperties = false + }; + + var schemaContainer = new JsonSchema(); + var appender = new JsonSchemaAppender(schemaContainer, new MappedNameGenerator(eventsDict)); + var generator = new JsonSchemaGenerator(settings); + + var jsonSchemas = eventsDict.Select(x => x.Value).Select(generator.Generate); + foreach (var schema in jsonSchemas) + { + appender.AppendSchema(schema, null); + } + + return schemaContainer; + } +} diff --git a/src/Bss.Platform.sln b/src/Bss.Platform.sln index 60f9c37..c82dddc 100644 --- a/src/Bss.Platform.sln +++ b/src/Bss.Platform.sln @@ -58,6 +58,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "__Solution Items", "__Solut Directory.Packages.props = Directory.Packages.props EndProjectSection EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bss.Platform.RabbitMq.JsonSchemaGeneratorBase", "Bss.Platform.RabbitMq.JsonSchemaGeneratorBase\Bss.Platform.RabbitMq.JsonSchemaGeneratorBase.csproj", "{6574B96D-56B5-4AC1-BF8B-711D86A8E9AA}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -124,6 +126,10 @@ Global {9EE0DF52-2C67-4DD1-AA9D-64ABC5BD00D7}.Debug|Any CPU.Build.0 = Debug|Any CPU {9EE0DF52-2C67-4DD1-AA9D-64ABC5BD00D7}.Release|Any CPU.ActiveCfg = Release|Any CPU {9EE0DF52-2C67-4DD1-AA9D-64ABC5BD00D7}.Release|Any CPU.Build.0 = Release|Any CPU + {6574B96D-56B5-4AC1-BF8B-711D86A8E9AA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {6574B96D-56B5-4AC1-BF8B-711D86A8E9AA}.Debug|Any CPU.Build.0 = Debug|Any CPU + {6574B96D-56B5-4AC1-BF8B-711D86A8E9AA}.Release|Any CPU.ActiveCfg = Release|Any CPU + {6574B96D-56B5-4AC1-BF8B-711D86A8E9AA}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -144,6 +150,7 @@ Global {0FC70E24-62A9-44BF-A433-804C98514A09} = {0F54786D-AB46-46AC-88DF-BEB789A62C1F} {393C3A3D-E55F-4006-81A7-ED1CB3FAC1FF} = {40F06493-58CE-4E9D-BBB4-771BD26A9658} {9EE0DF52-2C67-4DD1-AA9D-64ABC5BD00D7} = {40F06493-58CE-4E9D-BBB4-771BD26A9658} + {6574B96D-56B5-4AC1-BF8B-711D86A8E9AA} = {0F54786D-AB46-46AC-88DF-BEB789A62C1F} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {9CEE3EB7-A785-4C3D-A1AE-9E82E66B9252} diff --git a/src/Directory.Build.props b/src/Directory.Build.props index c854181..eb238da 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -1,6 +1,7 @@ - net9.0 + net9.0 + 14 https://github.com/Luxoft/bss-platform git @@ -20,7 +21,9 @@ false false - false + true + false + $(NoWarn);1591 true NU1901;NU1902;NU1903;NU1904 diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index d7e8a96..fa02da2 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -31,6 +31,7 @@     + diff --git a/src/Tests.Unit/Platform/Events/DependencyInjectionSystemBindingsTests.cs b/src/Tests.Unit/Platform/Events/DependencyInjectionSystemBindingsTests.cs new file mode 100644 index 0000000..5e8b655 --- /dev/null +++ b/src/Tests.Unit/Platform/Events/DependencyInjectionSystemBindingsTests.cs @@ -0,0 +1,97 @@ +using System.Text; + +using Bss.Platform.Events; +using Bss.Platform.Events.Models; + +using FluentAssertions; + +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +using Xunit; + +namespace Tests.Unit.Platform.Events; + +public class DependencyInjectionSystemBindingsTests +{ + // matches the example from IntegrationEventsMessageQueueOptions.EnableExternalSystemBindings XML doc + private const string Json = """ + { + "RabbitCap": { + "ExternalSystemBindings": { + "ExcludeOutputEvents": ["EXT.Debug*", "EXT.Internal.SomeEvent"], + "SystemBindings": { + "system1-allEvents-queue-except-excluded": null, + "system2-allEvents-queue-except-excluded": [], + "system3-fixedEvents-queue-exclude-not-applied": ["EXT.OrderCreated", "EXT.OrderCancelled", "EXT.Debug.Some"] + } + } + } + } + """; + + private static IConfiguration BuildConfiguration() => + new ConfigurationBuilder() + .AddJsonStream(new MemoryStream(Encoding.UTF8.GetBytes(Json))) + .Build(); + + [Fact] + public void ResolveSystemBindings_preserves_all_keys_including_null_and_empty_array_values() + { + var configuration = BuildConfiguration(); + + var systemBindings = DependencyInjection.ResolveSystemBindings(configuration, "RabbitCap:ExternalSystemBindings"); + + // null and [] both mean "no explicit routing keys" (see ExternalSystemBindingsOptions.SystemBindings doc) + // and the JSON provider's raw representation of that distinction is an undocumented implementation + // detail that differs across framework versions - only assert both keys survive, not which shape they take. + systemBindings.Should().HaveCount(3); + systemBindings.Should().ContainKey("system1-allEvents-queue-except-excluded") + .WhoseValue.Should().BeNullOrEmpty(); + systemBindings.Should().ContainKey("system2-allEvents-queue-except-excluded") + .WhoseValue.Should().BeNullOrEmpty(); + systemBindings["system3-fixedEvents-queue-exclude-not-applied"] + .Should().BeEquivalentTo("EXT.OrderCreated", "EXT.OrderCancelled", "EXT.Debug.Some"); + } + + // reproduces AddExternalSystemQueueBindings's own registration (BindConfiguration + Configure) + // to guard the contract: a consumer's PostConfigure must always win, regardless of call order. + private static IServiceCollection RegisterLikeLibraryDoes(IConfiguration configuration) => + new ServiceCollection() + .AddSingleton(configuration) + .AddOptions() + .BindConfiguration("RabbitCap:ExternalSystemBindings") + .Configure((options, config) => + options.SystemBindings = DependencyInjection.ResolveSystemBindings(config, "RabbitCap:ExternalSystemBindings")) + .Services; + + [Fact] + public void Consumer_PostConfigure_registered_after_library_setup_overrides_SystemBindings() + { + var services = RegisterLikeLibraryDoes(BuildConfiguration()); + services.PostConfigure(o => o.SystemBindings = new() { ["overridden-queue"] = ["EXT.Custom"] }); + + var options = services.BuildServiceProvider().GetRequiredService>().Value; + + options.SystemBindings.Should().ContainSingle().Which.Should().BeEquivalentTo( + new KeyValuePair("overridden-queue", ["EXT.Custom"])); + } + + [Fact] + public void Consumer_PostConfigure_registered_before_library_setup_still_overrides_SystemBindings() + { + IServiceCollection services = new ServiceCollection(); + services.PostConfigure(o => o.SystemBindings = new() { ["overridden-queue"] = ["EXT.Custom"] }); + + foreach (var descriptor in RegisterLikeLibraryDoes(BuildConfiguration())) + { + services.Add(descriptor); + } + + var options = services.BuildServiceProvider().GetRequiredService>().Value; + + options.SystemBindings.Should().ContainSingle().Which.Should().BeEquivalentTo( + new KeyValuePair("overridden-queue", ["EXT.Custom"])); + } +} diff --git a/src/Tests.Unit/Platform/Events/ExternalSystemBindingsResolverTests.cs b/src/Tests.Unit/Platform/Events/ExternalSystemBindingsResolverTests.cs new file mode 100644 index 0000000..eba6043 --- /dev/null +++ b/src/Tests.Unit/Platform/Events/ExternalSystemBindingsResolverTests.cs @@ -0,0 +1,60 @@ +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Internal; +using Bss.Platform.Events.Models; + +using FluentAssertions; + +using Microsoft.Extensions.Options; + +using Xunit; + +namespace Tests.Unit.Platform.Events; + +public class ExternalSystemBindingsResolverTests +{ + private sealed class StaticOptionsSnapshot(TOptions value) : IOptionsSnapshot + where TOptions : class + { + public TOptions Value { get; } = value; + + public TOptions Get(string? name) => this.Value; + } + + private sealed record PublicOutputEvent(string Id); + + private sealed record InternalOutputEvent(string Id); + + private sealed class FakeEventTypeProvider( + IReadOnlyDictionary inputEvents, + IReadOnlyDictionary outputEvents) : IEventTypeProvider + { + public IReadOnlyDictionary InputEvents { get; } = inputEvents; + + public IReadOnlyDictionary OutputEvents { get; } = outputEvents; + } + + [Fact] + public void ResolveOutputEventsForExport_applies_exclude_patterns() + { + var provider = new FakeEventTypeProvider( + new Dictionary(), + new Dictionary + { + [typeof(PublicOutputEvent)] = "EXT.Public.A", + [typeof(InternalOutputEvent)] = "EXT.Internal.B" + }); + + var options = new ExternalSystemBindingsOptions + { + ExcludeOutputEvents = ["EXT.Internal*"], + SystemBindings = { ["public-queue"] = null } + }; + var resolver = new ExternalSystemBindingsResolver( + provider, + new StaticOptionsSnapshot(options)); + + var result = resolver.ResolveOutputEventsForExport(); + + result.Keys.Should().BeEquivalentTo("EXT.Public.A"); + } +} diff --git a/src/Tests.Unit/Platform/Events/ExternalSystemQueueBindingsInitializerTests.cs b/src/Tests.Unit/Platform/Events/ExternalSystemQueueBindingsInitializerTests.cs new file mode 100644 index 0000000..ff2e678 --- /dev/null +++ b/src/Tests.Unit/Platform/Events/ExternalSystemQueueBindingsInitializerTests.cs @@ -0,0 +1,139 @@ +using Bss.Platform.Events.Abstractions; +using Bss.Platform.Events.Internal; +using Bss.Platform.Events.Models; + +using FluentAssertions; + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +using Xunit; + +namespace Tests.Unit.Platform.Events; + +public class ExternalSystemQueueBindingsInitializerTests +{ + private sealed class StaticOptionsSnapshot(TOptions value) : IOptionsSnapshot + where TOptions : class + { + public TOptions Value { get; } = value; + + public TOptions Get(string? name) => this.Value; + } + + private sealed class FakeEventTypeProvider(params string[] outputRoutingKeys) : IEventTypeProvider + { + // arbitrary distinct types used only as dictionary keys - bindings resolver only reads OutputEvents + private static readonly Type[] DummyTypes = [typeof(object), typeof(string), typeof(int), typeof(bool), typeof(double), typeof(long)]; + + public IReadOnlyDictionary InputEvents { get; } = new Dictionary(); + + public IReadOnlyDictionary OutputEvents { get; } = + outputRoutingKeys.Select((key, i) => (key, i)).ToDictionary(x => DummyTypes[x.i], x => x.key); + } + + private static ServiceProvider BuildProvider(Dictionary systemBindings, bool registerEventTypeProvider) + { + var services = new ServiceCollection(); + services.AddOptions(); + services.AddOptions().Configure(o => o.SystemBindings = systemBindings); + if (registerEventTypeProvider) + { + services.AddSingleton(new FakeEventTypeProvider()); + } + + services.AddSingleton(); + services.AddSingleton(); + + return services.BuildServiceProvider(); + } + + private static ExternalSystemBindingsResolver CreateResolver( + ExternalSystemBindingsOptions options, + params string[] outputRoutingKeys) => + new(new FakeEventTypeProvider(outputRoutingKeys), new StaticOptionsSnapshot(options)); + + [Fact] + public void Resolves_when_IEventTypeProvider_is_registered() + { + var provider = BuildProvider(new Dictionary(), registerEventTypeProvider: true); + + var act = () => provider.GetRequiredService(); + + act.Should().NotThrow(); + } + + [Fact] + public void Throws_when_IEventTypeProvider_is_not_registered() + { + // the initializer requires IEventTypeProvider unconditionally, regardless of whether any bindings are configured + var provider = BuildProvider(new Dictionary(), registerEventTypeProvider: false); + + var act = () => provider.GetRequiredService(); + + act.Should().Throw(); + } + + [Fact] + public async Task InitializeAsync_does_nothing_when_no_bindings_configured() + { + var provider = BuildProvider(new Dictionary(), registerEventTypeProvider: true); + var initializer = provider.GetRequiredService(); + + var act = () => initializer.InitializeAsync(null!, CancellationToken.None); + + await act.Should().NotThrowAsync(); + } + + [Fact] + public void ResolveQueueBindings_uses_all_output_events_by_default() + { + var options = new ExternalSystemBindingsOptions { SystemBindings = { ["crm-queue"] = null } }; + var resolver = CreateResolver(options, "EXT.A", "EXT.B"); + + var result = resolver.ResolveQueueBindings(); + + result["crm-queue"].Should().BeEquivalentTo("EXT.A", "EXT.B"); + } + + [Fact] + public void ResolveQueueBindings_applies_exclude_masks_only_to_default_set() + { + var options = new ExternalSystemBindingsOptions + { + ExcludeOutputEvents = ["EXT.Debug*"], + SystemBindings = { ["crm-queue"] = null, ["billing-queue"] = ["EXT.Debug.Ping"] } + }; + var resolver = CreateResolver(options, "EXT.A", "EXT.Debug.Ping", "EXT.B"); + + var result = resolver.ResolveQueueBindings(); + + result["crm-queue"].Should().BeEquivalentTo("EXT.A", "EXT.B"); + result["billing-queue"].Should().BeEquivalentTo("EXT.Debug.Ping"); + } + + [Fact] + public void ResolveQueueBindings_explicit_list_overrides_default_without_validation() + { + var options = new ExternalSystemBindingsOptions + { + SystemBindings = { ["analytics-queue"] = ["EXT.NotRegisteredAnywhere"] } + }; + var resolver = CreateResolver(options, "EXT.A", "EXT.B"); + + var result = resolver.ResolveQueueBindings(); + + result["analytics-queue"].Should().BeEquivalentTo("EXT.NotRegisteredAnywhere"); + } + + [Fact] + public void ResolveQueueBindings_returns_empty_map_when_no_systems_configured() + { + var options = new ExternalSystemBindingsOptions(); + var resolver = CreateResolver(options, "EXT.A"); + + var result = resolver.ResolveQueueBindings(); + + result.Should().BeEmpty(); + } +} diff --git a/src/Tests.Unit/Platform/Events/RabbitInitializersHostedServiceTests.cs b/src/Tests.Unit/Platform/Events/RabbitInitializersHostedServiceTests.cs new file mode 100644 index 0000000..acad92e --- /dev/null +++ b/src/Tests.Unit/Platform/Events/RabbitInitializersHostedServiceTests.cs @@ -0,0 +1,70 @@ +using Bss.Platform.Events.Interfaces; +using Bss.Platform.Events.Internal; + +using DotNetCore.CAP.RabbitMQ; + +using FluentAssertions; + +using Microsoft.Extensions.Logging.Abstractions; + +using RabbitMQ.Client; + +using Xunit; + +namespace Tests.Unit.Platform.Events; + +public class RabbitInitializersHostedServiceTests +{ + private sealed class ThrowingInitializer : IRabbitInitializer + { + public bool WasCalled { get; private set; } + + public Task InitializeAsync(RabbitMQ.Client.IModel model, CancellationToken cancellationToken) + { + this.WasCalled = true; + throw new InvalidOperationException("boom"); + } + } + + private sealed class RecordingInitializer : IRabbitInitializer + { + public bool WasCalled { get; private set; } + + public Task InitializeAsync(RabbitMQ.Client.IModel model, CancellationToken cancellationToken) + { + this.WasCalled = true; + return Task.CompletedTask; + } + } + + private sealed class ConnectionChannelPool: IConnectionChannelPool { + public IConnection GetConnection() => null!; + + public IModel Rent() => null!; + + public bool Return(IModel context) => true; + + public string HostAddress => null!; + + public string Exchange => null!; + } + + [Fact] + public async Task RunInitializersAsync_continues_after_one_initializer_throws() + { + var throwing = new ThrowingInitializer(); + var recordingBefore = new RecordingInitializer(); + var recordingAfter = new RecordingInitializer(); + + var service = new RabbitInitializersHostedService( + connectionChannelPool: new ConnectionChannelPool(), + initializers: [recordingBefore, throwing, recordingAfter], + logger: NullLogger.Instance); + + await service.StartAsync(CancellationToken.None); + + recordingBefore.WasCalled.Should().BeTrue(); + throwing.WasCalled.Should().BeTrue(); + recordingAfter.WasCalled.Should().BeTrue(); + } +} diff --git a/src/Tests.Unit/Platform/Events/WildcardMatcherTests.cs b/src/Tests.Unit/Platform/Events/WildcardMatcherTests.cs new file mode 100644 index 0000000..9ccaeca --- /dev/null +++ b/src/Tests.Unit/Platform/Events/WildcardMatcherTests.cs @@ -0,0 +1,24 @@ +using Bss.Platform.Events.Internal; + +using FluentAssertions; + +using Xunit; + +namespace Tests.Unit.Platform.Events; + +public class WildcardMatcherTests +{ + [Theory] + [InlineData("EXT.OrderCreated", "EXT.OrderCreated", true)] + [InlineData("EXT.OrderCreated", "EXT.OrderCancelled", false)] + [InlineData("EXT.Debug.Ping", "EXT.Debug*", true)] + [InlineData("EXT.Something", "EXT.Debug*", false)] + [InlineData("EXT.Internal.Debug", "*.Debug", true)] + [InlineData("EXT.Internal.Debugger", "*.Debug", false)] + [InlineData("EXT.Internal.Debug.Extra", "EXT.*.Debug.*", true)] + [InlineData("anything", "*", true)] + [InlineData("EXT.OrderCreated", "ext.ordercreated", true)] + [InlineData("EXT.DEBUG.Ping", "ext.debug*", true)] + public void IsMatch_returns_expected_result(string value, string pattern, bool expected) => + WildcardMatcher.IsMatch(value, pattern).Should().Be(expected); +} diff --git a/src/Tests.Unit/Platform/RabbitMq/JsonSchemaGeneratorBase/RabbitEventsSchemaExporterTests.cs b/src/Tests.Unit/Platform/RabbitMq/JsonSchemaGeneratorBase/RabbitEventsSchemaExporterTests.cs new file mode 100644 index 0000000..06f4321 --- /dev/null +++ b/src/Tests.Unit/Platform/RabbitMq/JsonSchemaGeneratorBase/RabbitEventsSchemaExporterTests.cs @@ -0,0 +1,80 @@ +using System.Text.Json; + +using Bss.Platform.RabbitMq.JsonSchemaGeneratorBase; + +using FluentAssertions; + +using Xunit; + +namespace Tests.Unit.Platform.RabbitMq.JsonSchemaGeneratorBase; + +public class RabbitEventsSchemaExporterTests +{ + private sealed record SampleInputEvent(string Id); + + private sealed record PublicOutputEvent(string Id); + + private sealed class TestSchemaExportSettings : IRabbitSchemaExportSettings + { + public required string FromQueueName { get; init; } + + public required string ExchangeName { get; init; } + + public required string System { get; init; } + + public IReadOnlyDictionary InputEvents { get; init; } = new Dictionary(); + + public IReadOnlyDictionary OutputEvents { get; init; } = new Dictionary(); + } + + private static RabbitEventsSchemaExporter CreateExporter( + IReadOnlyDictionary inputEvents, + IReadOnlyDictionary outputEvents, + string exchange = "service.exchange", + string queue = "service.queue", + string systemName = "service.system") => + new( + new TestSchemaExportSettings + { + InputEvents = inputEvents, + OutputEvents = outputEvents, + ExchangeName = exchange, + FromQueueName = queue, + System = systemName + }); + + [Fact] + public void BuildExportPayloadJson_contains_required_fields_and_event_schemas() + { + var exporter = CreateExporter( + new Dictionary { ["IN.OrderAccepted"] = typeof(SampleInputEvent) }, + new Dictionary { ["EXT.OrderCreated"] = typeof(PublicOutputEvent) }, + "orders.exchange", + "orders.queue", + "orders"); + + using var payload = JsonDocument.Parse(exporter.BuildExportPayloadJson()); + + payload.RootElement.GetProperty("systemName").GetString().Should().Be("orders"); + payload.RootElement.GetProperty("exchange").GetString().Should().Be("orders.exchange"); + payload.RootElement.GetProperty("queue").GetString().Should().Be("orders.queue"); + payload.RootElement.GetProperty("input").GetRawText().Should().Contain("IN.OrderAccepted"); + payload.RootElement.GetProperty("output").GetRawText().Should().Contain("EXT.OrderCreated"); + } + + [Fact] + public void BuildExportPayloadJson_keeps_provided_empty_queue_and_system_name_values() + { + var exporter = CreateExporter( + new Dictionary { ["IN.Event"] = typeof(SampleInputEvent) }, + new Dictionary { ["EXT.Event"] = typeof(PublicOutputEvent) }, + "payments.exchange", + string.Empty, + string.Empty); + + using var payload = JsonDocument.Parse(exporter.BuildExportPayloadJson()); + + payload.RootElement.GetProperty("queue").GetString().Should().Be(string.Empty); + payload.RootElement.GetProperty("systemName").GetString().Should().Be(string.Empty); + } +} diff --git a/src/Tests.Unit/Tests.Unit.csproj b/src/Tests.Unit/Tests.Unit.csproj index f82a352..8b8b55d 100644 --- a/src/Tests.Unit/Tests.Unit.csproj +++ b/src/Tests.Unit/Tests.Unit.csproj @@ -5,6 +5,7 @@ + @@ -12,9 +13,11 @@ + + diff --git a/src/__SolutionItems/CommonAssemblyInfo.cs b/src/__SolutionItems/CommonAssemblyInfo.cs index ce2b74b..2d3b7b8 100644 --- a/src/__SolutionItems/CommonAssemblyInfo.cs +++ b/src/__SolutionItems/CommonAssemblyInfo.cs @@ -4,9 +4,9 @@ [assembly: AssemblyCompany("Luxoft")] [assembly: AssemblyCopyright("Copyright © Luxoft 2026")] -[assembly: AssemblyVersion("1.6.9.0")] -[assembly: AssemblyFileVersion("1.6.9.0")] -[assembly: AssemblyInformationalVersion("1.6.9.0")] +[assembly: AssemblyVersion("1.7.0.0")] +[assembly: AssemblyFileVersion("1.7.0.0")] +[assembly: AssemblyInformationalVersion("1.7.0.0")] #if DEBUG [assembly: AssemblyConfiguration("Debug")]