using MassTransit; using Musify.Infrastructure.MassTransit.Consumers; using Musify.Infrastructure.Persistence; namespace Musify.Infrastructure.MassTransit { public static partial class MassTransitDependencyInjection { static void ConfigureConsumerEndpoint( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string queueName) where TConsumer : class, IConsumer { busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator => { endpointConfigurator.UseMessageRetry(retryConfigurator => retryConfigurator.Exponential( retryLimit: 5, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(2))); endpointConfigurator.UseEntityFrameworkOutbox(busRegistrationContext); endpointConfigurator.ConfigureConsumer(busRegistrationContext); }); } static void ConfigureExecuteActivityEndpoint( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string endpointName) where TActivity : class, IExecuteActivity where TArguments : class { busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator => { endpointConfigurator.UseMessageRetry(retryConfigurator => retryConfigurator.Exponential( retryLimit: 5, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(2))); endpointConfigurator.UseEntityFrameworkOutbox(busRegistrationContext); endpointConfigurator.ExecuteActivityHost(busRegistrationContext); }); } static void ConfigureActivityEndpoint( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string endpointName) where TActivity : class, IActivity where TArguments : class where TLog : class { busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.ExecuteQueue(endpointName), endpointConfigurator => { endpointConfigurator.UseMessageRetry(retryConfigurator => retryConfigurator.Exponential( retryLimit: 5, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(2))); endpointConfigurator.UseEntityFrameworkOutbox(busRegistrationContext); endpointConfigurator.ExecuteActivityHost( EndpointHelper.BuildCompensateActivityUri(endpointName), busRegistrationContext); }); busFactoryConfigurator.ReceiveEndpoint(MassTransitEndpointNames.CompensateQueue(endpointName), endpointConfigurator => { endpointConfigurator.UseMessageRetry(retryConfigurator => retryConfigurator.Exponential( retryLimit: 5, minInterval: TimeSpan.FromSeconds(1), maxInterval: TimeSpan.FromSeconds(30), intervalDelta: TimeSpan.FromSeconds(2))); endpointConfigurator.UseEntityFrameworkOutbox(busRegistrationContext); endpointConfigurator.CompensateActivityHost(busRegistrationContext); }); } } }