elsa-core/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs

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.AzureServiceBus.Features;

/// Configures MassTransit to use the Azure Service Bus transport.
/// See https://masstransit.io/documentation/configuration/transports/azure-service-bus
[DependsOn(typeof(MassTransitFeature))]
public class AzureServiceBusFeature : FeatureBase
{
    /// <inheritdoc />
    public AzureServiceBusFeature(IModule module) : base(module)
    {
    }
    
    /// An Azure Service Bus connection string.
    public string? ConnectionString { get; set; }

    /// <summary>
    /// A delegate that configures the Azure Service Bus transport options.
    /// </summary>
    public Action<IServiceBusBusFactoryConfigurator>? ConfigureServiceBus { get; set; }

    /// <inheritdoc />
    public override void Configure()
    {
        Module.Configure<MassTransitFeature>(massTransitFeature =>
        {
            massTransitFeature.BusConfigurator = configure =>
            {
                var tempConsumers = massTransitFeature.GetConsumers()
                    .Where(c => c.IsTemporary)
                    .ToList();
                
                configure.AddServiceBusMessageScheduler();
                configure.AddConsumers(tempConsumers.Select(c => c.ConsumerType).ToArray());
                
                configure.UsingAzureServiceBus((context, serviceBus) =>
                {
                    var options = context.GetRequiredService<IOptions<MassTransitWorkflowDispatcherOptions>>().Value;
                    var instanceNameRetriever = context.GetRequiredService<IInstanceNameRetriever>();

                    if (ConnectionString != null) 
                        serviceBus.Host(ConnectionString);
                    serviceBus.UseServiceBusMessageScheduler();
                    ConfigureServiceBus?.Invoke(serviceBus);

                    foreach (var consumer in tempConsumers)
                    {
                        //Throw on no known name?
                        serviceBus.ReceiveEndpoint($"{instanceNameRetriever.GetName()}-{consumer.Name}", configurator =>
                        {
                            configurator.AutoDeleteOnIdle = options.ShortTermQueueLifetime ?? TimeSpan.FromHours(1);
                            configurator.ConcurrentMessageLimit = options.ConcurrentMessageLimit;
                            configurator.ConfigureConsumer<DispatchCancelWorkflowsRequestConsumer>(context);
                        });
                    }

                    serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false));
                });
            };
        });
    }
}