using MassTransit; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using Musify.Application.Contracts.Infrastructure; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.MassTransit.Activities; using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Audio; using Musify.Infrastructure.MassTransit.Activities.Files; using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Consumers; using Musify.Infrastructure.MassTransit.Filters; using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.RoutingSlip.Builders; namespace Musify.Infrastructure.MassTransit { public static class MassTransitDependencyInjection { public static IServiceCollection AddMassTransitClient(this IServiceCollection serviceDescriptors, IConfiguration configuration) { serviceDescriptors .AddOptionsWithValidateOnStart() .Bind(configuration.GetRequiredSection(MassTransitConfiguration.SectionName)) .ValidateDataAnnotations(); serviceDescriptors.AddSingleton(serviceProvider => serviceProvider.GetRequiredService>().Value); serviceDescriptors.AddScoped(); serviceDescriptors.AddMassTransit(options => { options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService())); } ); return serviceDescriptors; } public static IServiceCollection AddMassTransitConsumers(this IServiceCollection serviceDescriptors, IConfiguration configuration) { serviceDescriptors .AddOptionsWithValidateOnStart() .Bind(configuration.GetRequiredSection(MassTransitConfiguration.SectionName)) .ValidateDataAnnotations(); serviceDescriptors.AddSingleton(serviceProvider => serviceProvider.GetRequiredService>().Value); serviceDescriptors .AddOptionsWithValidateOnStart() .Bind(configuration.GetRequiredSection(WorkerConfiguration.SectionName)) .ValidateDataAnnotations(); serviceDescriptors.AddSingleton(serviceProvider => serviceProvider.GetRequiredService>().Value); serviceDescriptors.AddScoped(); serviceDescriptors.AddScoped(); serviceDescriptors.AddScoped(); serviceDescriptors.AddMassTransit(options => { options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); options.AddConsumer(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddExecuteActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.AddActivity(); options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) => { ConfigureRabbitMqHost( busFactoryConfigurator, busRegistrationContext.GetRequiredService()); busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext); busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator => endpointConfigurator.ConfigureConsumer(busRegistrationContext)); busFactoryConfigurator.ReceiveEndpoint(RemoveFileConsumer.QueueName, endpointConfigurator => endpointConfigurator.ConfigureConsumer(busRegistrationContext)); busFactoryConfigurator.ReceiveEndpoint(UpdateTrackAudioConsumer.QueueName, endpointConfigurator => endpointConfigurator.ConfigureConsumer(busRegistrationContext)); busFactoryConfigurator.ReceiveEndpoint(UpdateTrackPictureConsumer.QueueName, endpointConfigurator => endpointConfigurator.ConfigureConsumer(busRegistrationContext)); busFactoryConfigurator.ReceiveEndpoint(RoutingSlipCleanUpConsumer.QueueName, endpointConfigurator => endpointConfigurator.ConfigureConsumer(busRegistrationContext)); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, RemoveFileFromBucketActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, GenerateAudioWorkflowPathsActivity.ExecuteEndpointName); ConfigureExecuteActivityEndpoint( busFactoryConfigurator, busRegistrationContext, GeneratePictureWorkflowPathsActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, ResizePictureActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UpdatePlayListPictureActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UpdateTrackPictureActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, DownloadFileFromBucketActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, TranscodeDashAudioActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, TransferFilesToBucketActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UpdateTrackAudioActivity.ExecuteEndpointName); ConfigureActivityEndpoint( busFactoryConfigurator, busRegistrationContext, UploadFileToBucketActivity.ExecuteEndpointName); } ); } ); return serviceDescriptors; } static void ConfigureRabbitMqHost( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, MassTransitConfiguration massTransitConfiguration) { busFactoryConfigurator.Host(massTransitConfiguration.Address, 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 ConfigureActivityEndpoint( IRabbitMqBusFactoryConfigurator busFactoryConfigurator, IBusRegistrationContext busRegistrationContext, string endpointName) where TActivity : class, IActivity where TArguments : class where TLog : class { busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator => endpointConfigurator.ExecuteActivityHost( EndpointHelper.BuildCompensateActivityUri(endpointName), busRegistrationContext) ); busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_compensate", endpointConfigurator => endpointConfigurator.CompensateActivityHost(busRegistrationContext) ); } } }