using MassTransit; using Microsoft.Extensions.DependencyInjection; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Messaging.Activities; using Musify.Infrastructure.Messaging.Activities.Arguments; using Musify.Infrastructure.Messaging.Consumers; using Musify.Infrastructure.Messaging.Filters; namespace Musify.Infrastructure.Messaging { public static class MessagingDependencyInjection { public static IServiceCollection AddMassTransitClient(this IServiceCollection serviceDescriptors) { serviceDescriptors.AddMassTransit(options => { options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService())); }); return serviceDescriptors; } public static IServiceCollection AddMassTransitConsumers(this IServiceCollection serviceDescriptors) { serviceDescriptors.AddMassTransit(options => { // Register consumers options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); // Register execute activities options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => { ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService()); busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext); // Consumer endpoints busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); }); busFactoryConfigurator.ReceiveEndpoint(RemoveFileConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); }); busFactoryConfigurator.ReceiveEndpoint(TranscodeAudioFromTrackConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); }); busFactoryConfigurator.ReceiveEndpoint(UpdateTrackPictureConsumer.QueueName, endpointConfigurator => { endpointConfigurator.ConfigureConsumer(busRegistrationContext); }); // Activity endpoints ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, ResizePictureActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, RemoveFileFromBucketActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UpdatePlayListPictureActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UpdateTrackPictureActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, DownloadFileFromBucketActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, TranscodeDashAudioActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, TransferFilesToBucket.ExecuteEndpointName); }); }); return serviceDescriptors; } static void ConfigureRabbitMqHost( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, MessagingConfiguration massTransitConfiguration) { busFactoryConfigurator.Host(massTransitConfiguration.Host, options => { options.Username(massTransitConfiguration.Username); options.Password(massTransitConfiguration.Password); }); } static void ConfigureExecuteActivityEndpoint( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string endpointName) where TActivity : class, IExecuteActivity where TArguments : class { busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator => { endpointConfigurator.ExecuteActivityHost(busRegistrationContext); }); } static void ConfigureActivityEndpoints( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string endpointName) where TActivity : class, IActivity where TArguments : class where TLog : class { busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator => { endpointConfigurator.ExecuteActivityHost(busRegistrationContext); }); busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_compensate", endpointConfigurator => { endpointConfigurator.CompensateActivityHost(busRegistrationContext); }); } } }