using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.MassTransit.Consumers; using Elsa.MassTransit.Features; using Elsa.MassTransit.Options; using Elsa.Workflows.Contracts; using MassTransit; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; namespace Elsa.MassTransit.RabbitMq.Features; /// /// Configures MassTransit to use the RabbitMQ transport. /// [DependsOn(typeof(MassTransitFeature))] public class RabbitMqServiceBusFeature : FeatureBase { /// public RabbitMqServiceBusFeature(IModule module) : base(module) { } /// A RabbitMQ connection string. public string? ConnectionString { get; set; } /// Configures the RabbitMQ transport options. public Action? TransportOptions { get; set; } /// /// Configures the RabbitMQ bus. /// public Action? ConfigureServiceBus { get; set; } /// public override void Configure() { Module.Configure(massTransitFeature => { massTransitFeature.BusConfigurator = configure => { var tempConsumers = massTransitFeature.GetConsumers() .Where(c => c.IsTemporary) .ToList(); configure.AddConsumers(tempConsumers.Select(c => c.ConsumerType).ToArray()); configure.UsingRabbitMq((context, serviceBus) => { var options = context.GetRequiredService>().Value; var instanceNameRetriever = context.GetRequiredService(); if (!string.IsNullOrEmpty(ConnectionString)) serviceBus.Host(ConnectionString); ConfigureServiceBus?.Invoke(serviceBus); foreach (var consumer in tempConsumers) { serviceBus.ReceiveEndpoint($"{instanceNameRetriever.GetName()}-{consumer.Name}", configurator => { configurator.QueueExpiration = options.ShortTermQueueLifetime ?? TimeSpan.FromHours(1); configurator.ConcurrentMessageLimit = options.ConcurrentMessageLimit; configurator.ConfigureConsumer(context); }); } serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); }); }; }); } /// public override void Apply() { if (TransportOptions != null) Services.Configure(TransportOptions); } }