using MassTransit; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Musify.Application.Contracts; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.MassTransit.Sagas; using Musify.Infrastructure.Persistence; namespace Musify.Infrastructure.MassTransit; public static partial class MassTransitDependencyInjection { public static IServiceCollection AddMassTransitClient(this IServiceCollection serviceDescriptors, IConfiguration configuration) { RegisterValidatedOptions( serviceDescriptors, configuration, MassTransitConfiguration.SectionName); serviceDescriptors.AddScoped(); serviceDescriptors.AddMassTransit(options => { options.AddEntityFrameworkOutbox(outbox => { outbox.UsePostgres(); outbox.UseBusOutbox(); outbox.QueryDelay = TimeSpan.FromSeconds(1); outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30); }); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService())); }); return serviceDescriptors; } public static IServiceCollection AddMassTransitConsumers(this IServiceCollection serviceDescriptors, IConfiguration configuration) { RegisterValidatedOptions( serviceDescriptors, configuration, MassTransitConfiguration.SectionName); RegisterValidatedOptions( serviceDescriptors, configuration, WorkerConfiguration.SectionName); RegisterRoutingSlipBuilders(serviceDescriptors); serviceDescriptors.AddMassTransit(options => { options.AddEntityFrameworkOutbox(outbox => { outbox.UsePostgres(); outbox.UseBusOutbox(); outbox.QueryDelay = TimeSpan.FromSeconds(1); outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30); }); options.SetKebabCaseEndpointNameFormatter(); RegisterConsumersAndActivities(options); options.AddSagaStateMachine() .EntityFrameworkRepository(repository => { repository.ConcurrencyMode = ConcurrencyMode.Pessimistic; repository.ExistingDbContext(); repository.UsePostgres(); }); options.AddSagaStateMachine() .EntityFrameworkRepository(repository => { repository.ConcurrencyMode = ConcurrencyMode.Pessimistic; repository.ExistingDbContext(); repository.UsePostgres(); }); options.AddSagaStateMachine() .EntityFrameworkRepository(repository => { repository.ConcurrencyMode = ConcurrencyMode.Pessimistic; repository.ExistingDbContext(); repository.UsePostgres(); }); options.AddConfigureEndpointsCallback((registrationContext, _, endpointConfigurator) => { UseStandardRetry(endpointConfigurator); endpointConfigurator.UseEntityFrameworkOutbox(registrationContext); }); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => { ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService()); busFactoryConfigurator.ConfigureEndpoints(busRegistrationContext); }); }); return serviceDescriptors; } private static void UseStandardRetry(IReceiveEndpointConfigurator endpointConfigurator) => endpointConfigurator.UseMessageRetry(retryConfigurator => retryConfigurator.Exponential( retryLimit: 5, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(2))); private static void ConfigureRabbitMqHost( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, MassTransitConfiguration massTransitConfiguration) { busFactoryConfigurator.Host(massTransitConfiguration.Address, options => { options.Username(massTransitConfiguration.Username); options.Password(massTransitConfiguration.Password); }); } }