SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Made per instance queues temporary

Browse source
This commit is contained in:
Raymond den Haan 2024-01-25 14:51:48 +01:00
parent 7251c79f40
commit 3b66d7edbb
12 changed files with 120 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs
View file

@ -1,8 +1,14 @@
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;
@ -31,15 +37,34 @@ public class AzureServiceBusFeature : FeatureBase
{
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));
});
};

43
src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs
View file

@ -1,9 +1,14 @@
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;
@ -32,18 +37,40 @@ public class RabbitMqServiceBusFeature : FeatureBase
/// <inheritdoc />
public override void Configure()
{
Module.Configure<MassTransitFeature>().BusConfigurator = configure =>
Module.Configure<MassTransitFeature>(massTransitFeature =>
{
configure.UsingRabbitMq((context, serviceBus) =>
massTransitFeature.BusConfigurator = configure =>
{
if (!string.IsNullOrEmpty(ConnectionString))
serviceBus.Host(ConnectionString);
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<IOptions<MassTransitWorkflowDispatcherOptions>>().Value;
var instanceNameRetriever = context.GetRequiredService<IInstanceNameRetriever>();
if (!string.IsNullOrEmpty(ConnectionString))
serviceBus.Host(ConnectionString);
ConfigureServiceBus?.Invoke(serviceBus);
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<DispatchCancelWorkflowsRequestConsumer>(context);
});
}
serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false));
});
};
serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false));
});
};
});
}
/// <inheritdoc />

31
src/modules/Elsa.MassTransit/ConsumerDefinitions/DispatchCancelWorkflowsRequestConsumerDefinition.cs
View file

@ -1,31 +0,0 @@
using Elsa.MassTransit.Consumers;
using Elsa.MassTransit.Options;
using MassTransit;
using MassTransit.Configuration;
using Microsoft.Extensions.Options;
namespace Elsa.MassTransit.ConsumerDefinitions;
/// <summary>
/// Configures the endpoint for <see cref="DispatchCancelWorkflowsRequestConsumer"/>
/// </summary>
public class DispatchCancelWorkflowsRequestConsumerDefinition : ConsumerDefinition<DispatchCancelWorkflowsRequestConsumer>
{
private readonly IOptions<MassTransitWorkflowDispatcherOptions> _options;
/// <inheritdoc />
public DispatchCancelWorkflowsRequestConsumerDefinition(IOptions<MassTransitWorkflowDispatcherOptions> options)
{
_options = options;
ConcurrentMessageLimit = _options.Value.ConcurrentMessageLimit;
EndpointName = "random-queue-name" + DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
this.EndpointDefinition!.IsTemporary = true;
}
/// <inheritdoc />
protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<DispatchCancelWorkflowsRequestConsumer> consumerConfigurator, IRegistrationContext context)
{
endpointConfigurator.UseMessageRetry(r => r.Interval(5, 1000));
endpointConfigurator.tem
}
}

13
src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs
View file

@ -18,27 +18,28 @@ public static class MassTransitFeatureExtensions
/// <summary>
/// Registers the specified type for MassTransit service bus consumer discovery.
/// </summary>
public static MassTransitFeature AddConsumer<T>(this MassTransitFeature feature) where T : IConsumer => feature.AddConsumer(typeof(T));
public static MassTransitFeature AddConsumer<T>(this MassTransitFeature feature, string? name, bool isTemporary) where T : IConsumer => feature.AddConsumer(typeof(T), name, isTemporary);
/// <summary>
/// Registers the specified type for MassTransit service bus consumer discovery.
/// </summary>
/// <typeparam name="T">The consumer type.</typeparam>
/// <typeparam name="TDefinition">The consumer definition type.</typeparam>
public static MassTransitFeature AddConsumer<T, TDefinition>(this MassTransitFeature feature)
public static MassTransitFeature AddConsumer<T, TDefinition>(this MassTransitFeature feature, string? name, bool isTemporary)
where T : IConsumer
where TDefinition : IConsumerDefinition
{
return feature.AddConsumer(typeof(T), typeof(TDefinition));
return feature.AddConsumer(typeof(T), name, isTemporary, typeof(TDefinition));
}
/// <summary>
/// Registers the specified type for MassTransit service bus consumer discovery.
/// </summary>
public static MassTransitFeature AddConsumer(this MassTransitFeature feature, Type type, Type? consumerDefinitionType = default)
public static MassTransitFeature AddConsumer(this MassTransitFeature feature, Type type, string? name, bool isTemporary,
Type? consumerDefinitionType = default)
{
var types = feature.Module.Properties.GetOrAdd(ServiceBusConsumerTypesKey, () => new HashSet<ConsumerTypeDefinition>());
types.Add(new ConsumerTypeDefinition(type, consumerDefinitionType));
types.Add(new ConsumerTypeDefinition(type, consumerDefinitionType, name, isTemporary));
return feature;
}
@ -60,7 +61,7 @@ public static class MassTransitFeatureExtensions
/// <summary>
/// Returns all collected consumer types.
/// </summary>
internal static IEnumerable<ConsumerTypeDefinition> GetConsumers(this MassTransitFeature feature) => feature.Module.Properties.GetOrAdd(ServiceBusConsumerTypesKey, () => new HashSet<ConsumerTypeDefinition>());
public static IEnumerable<ConsumerTypeDefinition> GetConsumers(this MassTransitFeature feature) => feature.Module.Properties.GetOrAdd(ServiceBusConsumerTypesKey, () => new HashSet<ConsumerTypeDefinition>());
/// <summary>
/// Returns all collected message types.

8
src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs
View file

@ -18,20 +18,20 @@ public static class ModuleExtensions
/// <summary>
/// Registers the specified consumer with MassTransit.
/// </summary>
public static IModule AddMassTransitConsumer<T>(this IModule module) where T : IConsumer
public static IModule AddMassTransitConsumer<T>(this IModule module, string? name = null, bool isTemporary = false) where T : IConsumer
{
module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<T>());
module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<T>(name, isTemporary));
return module;
}
/// <summary>
/// Registers the specified consumer and consumer definition with MassTransit.
/// </summary>
public static IModule AddMassTransitConsumer<T, TDefinition>(this IModule module)
public static IModule AddMassTransitConsumer<T, TDefinition>(this IModule module, string? name = null, bool isTemporary = false)
where T : IConsumer
where TDefinition : IConsumerDefinition
{
module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<T, TDefinition>());
module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<T, TDefinition>(name, isTemporary));
return module;
}

7
src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs
View file

@ -48,7 +48,7 @@ public class MassTransitFeature : FeatureBase
Services.Configure<MassTransitWorkflowDispatcherOptions>(x => { });
Services.AddActivityProvider<MassTransitActivityTypeProvider>();
var busConfigurator = BusConfigurator ??= configure =>
{
configure.UsingInMemory((context, configurator) =>
@ -92,7 +92,10 @@ public class MassTransitFeature : FeatureBase
var workflowMessageConsumers = this.GetMessages().Select(x => new ConsumerTypeDefinition(workflowMessageConsumerType.MakeGenericType(x)));
// Concatenate the manually registered consumers with the workflow message consumers.
var consumerTypeDefinitions = this.GetConsumers().Concat(workflowMessageConsumers).ToArray();
var consumerTypeDefinitions = this.GetConsumers()
//Temporary queues require implementation specific variables which will be handled in their respective projects
.Where(c => c.IsTemporary == false)
.Concat(workflowMessageConsumers).ToArray();
Services.AddMassTransit(bus =>
{

2
src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs
View file

@ -35,7 +35,7 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
public override void Configure()
{
Module.AddMassTransitConsumer<DispatchWorkflowRequestConsumer, DispatchWorkflowRequestConsumerDefinition>();
Module.AddMassTransitConsumer<DispatchCancelWorkflowsRequestConsumer, DispatchCancelWorkflowsRequestConsumerDefinition>();
Module.AddMassTransitConsumer<DispatchCancelWorkflowsRequestConsumer>("dispatch-cancel-workflow", true);
Module.Configure<WorkflowRuntimeFeature>(f => f.WorkflowDispatcher = sp => sp.GetRequiredService<MassTransitWorkflowDispatcher>());
}

2
src/modules/Elsa.MassTransit/Models/ConsumerTypeDefinition.cs
View file

@ -1,3 +1,3 @@
namespace Elsa.MassTransit.Models;
internal record ConsumerTypeDefinition(Type ConsumerType, Type? ConsumerDefinitionType = default);
public record ConsumerTypeDefinition(Type ConsumerType, Type? ConsumerDefinitionType = default, string? Name = null, bool IsTemporary = false);

2
src/modules/Elsa.MassTransit/Options/MassTransitWorkflowDispatcherOptions.cs
View file

@ -5,6 +5,8 @@ namespace Elsa.MassTransit.Options;
/// Provides options to the <see cref="DispatchWorkflowRequestConsumerDefinition"/>
public class MassTransitWorkflowDispatcherOptions
{
/// The TTL of queues that are seen as short lived (typically queues that are created per running instance).
public TimeSpan? ShortTermQueueLifetime { get; set; }
/// The number of concurrent messages to process.
public int? ConcurrentMessageLimit { get; set; }
}

12
src/modules/Elsa.Workflows.Core/Contracts/IInstanceNameRetriever.cs Normal file
View file

@ -0,0 +1,12 @@
namespace Elsa.Workflows.Contracts;
/// <summary>
/// Retrieves a name of the current instance.
/// </summary>
public interface IInstanceNameRetriever
{
/// <summary>
/// Returns a name for the instance.
/// </summary>
public string GetName();
}

19
src/modules/Elsa.Workflows.Core/Services/RandomInstanceNameRetriever.cs Normal file
View file

@ -0,0 +1,19 @@
using Elsa.Workflows.Contracts;
namespace Elsa.Workflows.Services;
/// <summary>
/// Returns a randomly generated instance name.
/// </summary>
public class RandomInstanceNameRetriever : IInstanceNameRetriever
{
private readonly string _instanceName;
public RandomInstanceNameRetriever(RandomLongIdentityGenerator identityGenerator)
{
_instanceName = identityGenerator.GenerateId();
}
/// <inheritdoc />
public string GetName() => _instanceName;
}

8
src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs
View file

@ -16,6 +16,7 @@ using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Providers;
using Elsa.Workflows.Runtime.Services;
using Elsa.Workflows.Runtime.Stores;
using Elsa.Workflows.Services;
using Medallion.Threading;
using Medallion.Threading.FileSystem;
using Microsoft.Extensions.DependencyInjection;
@ -98,6 +99,11 @@ public class WorkflowRuntimeFeature : FeatureBase
/// </summary>
public Func<IServiceProvider, IBackgroundActivityScheduler> BackgroundActivityScheduler { get; set; } = sp => ActivatorUtilities.CreateInstance<LocalBackgroundActivityScheduler>(sp);
/// <summary>
/// A factory that instantiates an <see cref="IInstanceNameRetriever"/>.
/// </summary>
public Func<IServiceProvider, IInstanceNameRetriever> InstanceNameRetriever { get; set; } = sp => ActivatorUtilities.CreateInstance<RandomInstanceNameRetriever>(sp);
/// <summary>
/// A delegate to configure the <see cref="DistributedLockingOptions"/>.
/// </summary>
@ -171,6 +177,8 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddScoped(WorkflowExecutionContextStore)
.AddSingleton(RunTaskDispatcher)
.AddSingleton(BackgroundActivityScheduler)
.AddSingleton(InstanceNameRetriever)
.AddSingleton<RandomLongIdentityGenerator>()
.AddScoped<IBookmarkManager, DefaultBookmarkManager>()
.AddScoped<IActivityExecutionManager, DefaultActivityExecutionManager>()
.AddScoped<IActivityExecutionStatsService, ActivityExecutionStatsService>()