using MassTransit; using Microsoft.Extensions.DependencyInjection; using Musify.Infrastructure.Configuration; using Musify.Infrastructure.MassTransit.Activities.Audio; using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Consumers; using Musify.Infrastructure.MassTransit.Logs; using Musify.Infrastructure.MassTransit.Activities; using Musify.Infrastructure.MassTransit.Activities.Arguments; using Musify.Infrastructure.MassTransit.Activities.Logs; using Musify.Infrastructure.MassTransit.Filters; using Musify.Infrastructure.MassTransit.RoutingSlip.Builders; using Musify.Infrastructure.MassTransit.Activities.Files; namespace Musify.Infrastructure.MassTransit { 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.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, 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 ConfigureActivityEndpoint( 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); } ); } } }