diff --git a/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings b/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings new file mode 100644 index 000000000..3d80b14ab --- /dev/null +++ b/src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj.DotSettings @@ -0,0 +1,2 @@ + + True \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs b/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs new file mode 100644 index 000000000..a93bc8f91 --- /dev/null +++ b/src/bundles/Elsa.Server.Web/Enums/MassTransitBroker.cs @@ -0,0 +1,10 @@ +namespace Elsa.Server.Web; + +/// +/// Represents the type of messaging broker used in MassTransit. +/// +public enum MassTransitBroker +{ + AzureServiceBus, + RabbitMq +} \ No newline at end of file diff --git a/src/bundles/Elsa.Server.Web/Program.cs b/src/bundles/Elsa.Server.Web/Program.cs index 78f1468e5..3078e1fc5 100644 --- a/src/bundles/Elsa.Server.Web/Program.cs +++ b/src/bundles/Elsa.Server.Web/Program.cs @@ -15,6 +15,7 @@ using Elsa.MongoDb.Extensions; using Elsa.MongoDb.Modules.Identity; using Elsa.MongoDb.Modules.Management; using Elsa.MongoDb.Modules.Runtime; +using Elsa.Server.Web; using Elsa.Workflows.Management.Compression; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Options; @@ -27,10 +28,9 @@ const bool useDapper = false; const bool useProtoActor = false; const bool useHangfire = false; const bool useQuartz = true; -const bool useMassTransit = false; -const bool useMassTransitAzureServiceBus = true; -const bool useMassTransitRabbitMq = false; +const bool useMassTransit = true; const bool useZipCompression = true; +const MassTransitBroker useMassTransitBroker = MassTransitBroker.AzureServiceBus; var builder = WebApplication.CreateBuilder(args); var services = builder.Services; @@ -142,6 +142,7 @@ services runtime.UseMassTransitDispatcher(); } runtime.WorkflowInboxCleanupOptions = options => configuration.GetSection("Runtime:WorkflowInboxCleanup").Bind(options); + runtime.WorkflowDispatcherOptions = options => configuration.GetSection("Runtime:WorkflowDispatcher").Bind(options); }) .UseEnvironments(environments => environments.EnvironmentsOptions = options => configuration.GetSection("Environments").Bind(options)) .UseScheduling(scheduling => @@ -210,7 +211,7 @@ services { elsa.UseMassTransit(massTransit => { - if (useMassTransitAzureServiceBus) + if (useMassTransitBroker == MassTransitBroker.AzureServiceBus) { massTransit.UseAzureServiceBus(azureServiceBusConnectionString, serviceBusFeature => serviceBusFeature.ConfigureServiceBus = bus => { @@ -222,7 +223,7 @@ services }); } - if (useMassTransitRabbitMq) + if (useMassTransitBroker == MassTransitBroker.RabbitMq) { massTransit.UseRabbitMq(rabbitMqConnectionString, rabbit => rabbit.ConfigureServiceBus = bus => { diff --git a/src/bundles/Elsa.Server.Web/appsettings.json b/src/bundles/Elsa.Server.Web/appsettings.json index 131ccbe59..396e4e0bd 100644 --- a/src/bundles/Elsa.Server.Web/appsettings.json +++ b/src/bundles/Elsa.Server.Web/appsettings.json @@ -97,6 +97,19 @@ "WorkflowInboxCleanup": { "SweepInterval": "00:00:10:00", "BatchSize": 1000 + }, + "WorkflowDispatcher": { + "Channels": [ + { + "Name": "Low" + }, + { + "Name": "Medium" + }, + { + "Name": "High" + } + ] } }, "Scripting": { diff --git a/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs b/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs index 0fa9e1706..f2b6be4d0 100644 --- a/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs +++ b/src/modules/Elsa.Alterations.Core/Contracts/DefaultAlteredWorkflowScheduler.cs @@ -27,6 +27,6 @@ public class DefaultAlteredWorkflowDispatcher : IAlteredWorkflowDispatcher /// public async Task DispatchAsync(RunAlterationsResult result, CancellationToken cancellationToken = default) { - await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(result.WorkflowInstanceId), cancellationToken); + await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(result.WorkflowInstanceId), cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs b/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs index b2a8e97bf..fa2c33cd6 100644 --- a/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs +++ b/src/modules/Elsa.Alterations/Endpoints/Workflows/Retry/Endpoint.cs @@ -56,7 +56,7 @@ public class Retry : ElsaEndpoint allResults.AddRange(results); // Schedule updated workflow. - await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(workflowInstance.Id), cancellationToken); + await _workflowDispatcher.DispatchAsync(new DispatchWorkflowInstanceRequest(workflowInstance.Id), cancellationToken: cancellationToken); } // Write response. diff --git a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs index 22eba245a..da4604a12 100644 --- a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs +++ b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs @@ -37,6 +37,6 @@ public class ResumeWorkflowInstance : INotificationHandlerThe name of the job. /// The workflow request. /// The cancellation token. - public async Task ExecuteAsync(string name, DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken); + public async Task ExecuteAsync(string name, DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs index 0d656ec96..3d1fb829f 100644 --- a/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Hangfire/Jobs/RunWorkflowJob.cs @@ -24,5 +24,5 @@ public class RunWorkflowJob /// The name of the job. /// The workflow request. /// The cancellation token. - public async Task ExecuteAsync(string name, DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken); + public async Task ExecuteAsync(string name, DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken) => await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs index 5050a7c16..81fed0955 100644 --- a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs @@ -1,7 +1,10 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Features; +using Elsa.MassTransit.Messages; +using Elsa.Workflows.Runtime.Activities; using MassTransit; namespace Elsa.MassTransit.AzureServiceBus.Features; @@ -33,14 +36,15 @@ public class AzureServiceBusFeature : FeatureBase { configure.AddServiceBusMessageScheduler(); - configure.UsingAzureServiceBus((context, serviceBus) => + configure.UsingAzureServiceBus((context, configurator) => { if (ConnectionString != null) - serviceBus.Host(ConnectionString); + configurator.Host(ConnectionString); - serviceBus.UseServiceBusMessageScheduler(); - ConfigureServiceBus?.Invoke(serviceBus); - serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + configurator.UseServiceBusMessageScheduler(); + configurator.SetupWorkflowDispatcherEndpoints(context); + ConfigureServiceBus?.Invoke(configurator); + configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); }); }; }); diff --git a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs index af1271514..53283432e 100644 --- a/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs @@ -1,6 +1,8 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; +using Elsa.MassTransit.Consumers; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Features; using MassTransit; using Microsoft.Extensions.DependencyInjection; @@ -21,9 +23,8 @@ public class RabbitMqServiceBusFeature : FeatureBase /// A RabbitMQ connection string. public string? ConnectionString { get; set; } - /// Configures the RabbitMQ transport options. public Action? TransportOptions { get; set; } - + /// /// Configures the RabbitMQ bus. /// @@ -34,14 +35,15 @@ public class RabbitMqServiceBusFeature : FeatureBase { Module.Configure().BusConfigurator = configure => { - configure.UsingRabbitMq((context, serviceBus) => + configure.UsingRabbitMq((context, configurator) => { if (!string.IsNullOrEmpty(ConnectionString)) - serviceBus.Host(ConnectionString); + configurator.Host(ConnectionString); - ConfigureServiceBus?.Invoke(serviceBus); + ConfigureServiceBus?.Invoke(configurator); - serviceBus.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); + configurator.SetupWorkflowDispatcherEndpoints(context); + configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false)); }); }; } diff --git a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs index de1121995..d0a32890a 100644 --- a/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs +++ b/src/modules/Elsa.MassTransit/Activities/MessageReceived.cs @@ -2,7 +2,7 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Expressions.Models; using Elsa.Extensions; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Services; using Elsa.Workflows; namespace Elsa.MassTransit.Activities; diff --git a/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs b/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs index 1802f3bd8..2e1770159 100644 --- a/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs +++ b/src/modules/Elsa.MassTransit/Activities/PublishMessage.cs @@ -2,7 +2,7 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Expressions.Helpers; using Elsa.Extensions; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Services; using Elsa.Workflows; using Elsa.Workflows.Attributes; using Elsa.Workflows.Models; diff --git a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs index fcfcf255a..4912ba6a6 100644 --- a/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/DispatchWorkflowRequestConsumer.cs @@ -1,6 +1,7 @@ using Elsa.MassTransit.Messages; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Options; +using JetBrains.Annotations; using MassTransit; namespace Elsa.MassTransit.Consumers; @@ -8,6 +9,7 @@ namespace Elsa.MassTransit.Consumers; /// /// A consumer of various dispatch message types to asynchronously execute workflows. /// +[UsedImplicitly] public class DispatchWorkflowRequestConsumer : IConsumer, IConsumer, diff --git a/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs index 17ca203bf..e8bc5afbb 100644 --- a/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs @@ -36,6 +36,6 @@ public class WorkflowMessageConsumer : IConsumer where T : class CorrelationId = correlationId, Input = input }; - await _workflowRuntime.DispatchAsync(request, cancellationToken); + await _workflowRuntime.DispatchAsync(request, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs b/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs new file mode 100644 index 000000000..1a2612397 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Contracts/IEndpointChannelFormatter.cs @@ -0,0 +1,14 @@ +namespace Elsa.MassTransit.Contracts; + +/// +/// Represents a contract for formatting channel queue names. +/// +public interface IEndpointChannelFormatter +{ + /// + /// Formats the queue name based on the provided channel name. + /// + /// The channel name to be formatted. + /// The formatted queue name. + string FormatEndpointName(string? channelName = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs b/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs index 80e03d6b1..ee05200e9 100644 --- a/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs +++ b/src/modules/Elsa.MassTransit/Extensions/MassTransitFeatureExtensions.cs @@ -1,7 +1,7 @@ using Elsa.Features.Services; using Elsa.MassTransit.Features; -using Elsa.MassTransit.Implementations; using Elsa.MassTransit.Models; +using Elsa.MassTransit.Services; using MassTransit; // ReSharper disable once CheckNamespace diff --git a/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs b/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs new file mode 100644 index 000000000..45622b6d5 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Extensions/ServiceBusChannelsConfigurationExtensions.cs @@ -0,0 +1,42 @@ +using Elsa.MassTransit.Consumers; +using Elsa.MassTransit.Contracts; +using Elsa.Workflows.Runtime.Options; +using MassTransit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Elsa.MassTransit.Extensions; + +/// +/// Provides extension methods for configuring service bus channel endpoints. +/// +public static class ServiceBusChannelsConfigurationExtensions +{ + /// + /// Sets up channel endpoints for receiving workflow dispatch requests. + /// + /// The MassTransit receive configurator. + /// The MassTransit bus registration context. + public static void SetupWorkflowDispatcherEndpoints(this IReceiveConfigurator configurator, IBusRegistrationContext context) + { + var dispatcherOptions = context.GetRequiredService>(); + var formatter = context.GetRequiredService(); + var channelDescriptors = dispatcherOptions.Value.Channels; + var defaultEndpointName = formatter.FormatEndpointName(); + + var endpointNames = new List + { + defaultEndpointName + }; + + endpointNames.AddRange(channelDescriptors.Select(x => formatter.FormatEndpointName(x.Name))); + + foreach (string endpointName in endpointNames) + { + configurator.ReceiveEndpoint(endpointName, endpoint => + { + endpoint.ConfigureConsumer(context); + }); + } + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs index d30250b18..527a9d5bc 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs @@ -5,9 +5,10 @@ using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Services; using Elsa.MassTransit.Consumers; -using Elsa.MassTransit.Implementations; +using Elsa.MassTransit.Extensions; using Elsa.MassTransit.Models; using Elsa.MassTransit.Options; +using Elsa.MassTransit.Services; using Elsa.Workflows.Attributes; using Elsa.Workflows.Management.Models; using Elsa.Workflows.Management.Options; @@ -49,20 +50,23 @@ public class MassTransitFeature : FeatureBase Services.Configure(x => { }); Services.AddActivityProvider(); - var busConfigurator = BusConfigurator ??= configure => + void Configurator(IBusRegistrationConfigurator configure) { configure.UsingInMemory((context, configurator) => { + configurator.SetupWorkflowDispatcherEndpoints(context); configurator.ConfigureEndpoints(context); - if (PrefetchCount != null) - configurator.PrefetchCount = PrefetchCount.Value; + if (PrefetchCount != null) configurator.PrefetchCount = PrefetchCount.Value; }); - }; + } + + var busConfigurator = BusConfigurator ??= Configurator; AddMassTransit(busConfigurator); // Add collected message types to options. - Services.Configure(options => options.MessageTypes = new HashSet(messageTypes)); + Services.Configure( + options => options.MessageTypes = new HashSet(messageTypes)); // Add collected message types as available variable types. Services.Configure(options => @@ -79,7 +83,8 @@ public class MassTransitFeature : FeatureBase }); // Configure message serializer. - SystemTextJsonMessageSerializer.Options.Converters.Add(new TypeJsonConverter(WellKnownTypeRegistry.CreateDefault())); + SystemTextJsonMessageSerializer.Options.Converters.Add( + new TypeJsonConverter(WellKnownTypeRegistry.CreateDefault())); } /// @@ -89,7 +94,8 @@ public class MassTransitFeature : FeatureBase { // For each message type, create a concrete WorkflowMessageConsumer. var workflowMessageConsumerType = typeof(WorkflowMessageConsumer<>); - var workflowMessageConsumers = this.GetMessages().Select(x => new ConsumerTypeDefinition(workflowMessageConsumerType.MakeGenericType(x))); + 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(); diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs index af2f2f4c6..30cba3ded 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs @@ -4,12 +4,13 @@ using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.MassTransit.ConsumerDefinitions; using Elsa.MassTransit.Consumers; -using Elsa.MassTransit.Implementations; -using Elsa.MassTransit.Messages; +using Elsa.MassTransit.Contracts; +using Elsa.MassTransit.Formatters; using Elsa.MassTransit.Options; +using Elsa.MassTransit.Services; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Features; -using MassTransit; +using Elsa.Workflows.Runtime.Services; using Microsoft.Extensions.DependencyInjection; namespace Elsa.MassTransit.Features; @@ -31,11 +32,20 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase /// public Action? ConfigureDispatcherOptions { get; set; } + /// + /// A factory that creates a . + /// + public Func ChannelQueueFormatterFactory { get; set; } = _ => new DefaultEndpointChannelFormatter(); + /// public override void Configure() { Module.AddMassTransitConsumer(); - Module.Configure(f => f.WorkflowDispatcher = sp => sp.GetRequiredService()); + Module.Configure(f => f.WorkflowDispatcher = sp => + { + var decoratedService = ActivatorUtilities.CreateInstance(sp); + return ActivatorUtilities.CreateInstance(sp, decoratedService); + }); } /// @@ -46,13 +56,6 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase if (ConfigureDispatcherOptions != null) options.Configure(ConfigureDispatcherOptions); - var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); - var queueAddress = new Uri($"queue:elsa-{queueName}"); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - EndpointConvention.Map(queueAddress); - - Services.AddScoped(); + Services.AddSingleton(ChannelQueueFormatterFactory); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs b/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs new file mode 100644 index 000000000..beca6be16 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Formatters/DefaultEndpointChannelFormatter.cs @@ -0,0 +1,18 @@ +using Elsa.MassTransit.Contracts; +using Humanizer; + +namespace Elsa.MassTransit.Formatters; + +/// +/// The DefaultChannelQueueFormatter class is an implementation of the IChannelQueueFormatter interface. +/// It is responsible for formatting channel queue names. +/// +public class DefaultEndpointChannelFormatter : IEndpointChannelFormatter +{ + /// + public string FormatEndpointName(string? channelName) + { + var channelSuffix = !string.IsNullOrEmpty(channelName) ? "-" + channelName.Kebaberize() : ""; + return $"elsa-dispatch-workflow-request{channelSuffix}"; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs deleted file mode 100644 index c0ccee005..000000000 --- a/src/modules/Elsa.MassTransit/Implementations/MassTransitWorkflowDispatcher.cs +++ /dev/null @@ -1,80 +0,0 @@ -using Elsa.MassTransit.Messages; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Requests; -using MassTransit; - -namespace Elsa.MassTransit.Implementations; - -/// -/// Implements by leveraging MassTransit. -/// -public class MassTransitWorkflowDispatcher : IWorkflowDispatcher -{ - private readonly IBus _bus; - - /// - /// Constructor. - /// - public MassTransitWorkflowDispatcher(IBus bus) - { - _bus = bus; - } - - /// - public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchWorkflowDefinition( - request.DefinitionId, - request.VersionOptions, - request.Input, - request.Properties, - request.CorrelationId, - request.InstanceId, - request.TriggerActivityId - ), cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchWorkflowInstance(request.InstanceId) - { - BookmarkId = request.BookmarkId, - ActivityId = request.ActivityId, - ActivityNodeId = request.ActivityNodeId, - ActivityInstanceId = request.ActivityInstanceId, - ActivityHash = request.ActivityHash, - Input = request.Input, - Properties = request.Properties, - CorrelationId = request.CorrelationId - }, cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchTriggerWorkflows(request.ActivityTypeName, request.BookmarkPayload) - { - CorrelationId = request.CorrelationId, - WorkflowInstanceId = request.WorkflowInstanceId, - ActivityInstanceId = request.ActivityInstanceId, - Input = request.Input - }, cancellationToken); - return new(); - } - - /// - public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default) - { - await _bus.Publish(new DispatchResumeWorkflows(request.ActivityTypeName, request.BookmarkPayload) - { - CorrelationId = request.CorrelationId, - WorkflowInstanceId = request.WorkflowInstanceId, - ActivityInstanceId = request.ActivityInstanceId, - Input = request.Input - }, cancellationToken); - return new(); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs b/src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs similarity index 98% rename from src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs rename to src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs index e2dabe143..9ecfff2c7 100644 --- a/src/modules/Elsa.MassTransit/Implementations/MassTransitActivityTypeProvider.cs +++ b/src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs @@ -10,13 +10,15 @@ using Elsa.Workflows.Helpers; using Elsa.Workflows.Management; using Elsa.Workflows.Models; using Humanizer; +using JetBrains.Annotations; using Microsoft.Extensions.Options; -namespace Elsa.MassTransit.Implementations; +namespace Elsa.MassTransit.Services; /// /// Provides activities to the system from the configured MassTransit message types. /// +[UsedImplicitly] public class MassTransitActivityTypeProvider(IActivityFactory activityFactory, IOptions options, IActivityDescriber activityDescriber) : IActivityProvider { /// diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs new file mode 100644 index 000000000..37f17a8a2 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Services/MassTransitWorkflowDispatcher.cs @@ -0,0 +1,98 @@ +using Elsa.MassTransit.Contracts; +using Elsa.MassTransit.Messages; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; +using MassTransit; + +namespace Elsa.MassTransit.Services; + +/// +/// Implements by leveraging MassTransit. +/// +public class MassTransitWorkflowDispatcher : IWorkflowDispatcher +{ + private readonly IBus _bus; + private readonly IEndpointChannelFormatter _endpointChannelFormatter; + + /// + /// Initializes a new instance of the class. + /// + public MassTransitWorkflowDispatcher(IBus bus, IEndpointChannelFormatter endpointChannelFormatter) + { + _bus = bus; + _endpointChannelFormatter = endpointChannelFormatter; + } + + /// + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + + await sendEndpoint.Send(new DispatchWorkflowDefinition( + request.DefinitionId, + request.VersionOptions, + request.Input, + request.Properties, + request.CorrelationId, + request.InstanceId, + request.TriggerActivityId + ), cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + + await sendEndpoint.Send(new DispatchWorkflowInstance(request.InstanceId) + { + BookmarkId = request.BookmarkId, + ActivityId = request.ActivityId, + ActivityNodeId = request.ActivityNodeId, + ActivityInstanceId = request.ActivityInstanceId, + ActivityHash = request.ActivityHash, + Input = request.Input, + Properties = request.Properties, + CorrelationId = request.CorrelationId + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + await sendEndpoint.Send(new DispatchTriggerWorkflows(request.ActivityTypeName, request.BookmarkPayload) + { + CorrelationId = request.CorrelationId, + WorkflowInstanceId = request.WorkflowInstanceId, + ActivityInstanceId = request.ActivityInstanceId, + Input = request.Input + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + /// + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + var sendEndpoint = await GetSendEndpointAsync(options); + await sendEndpoint.Send(new DispatchResumeWorkflows(request.ActivityTypeName, request.BookmarkPayload) + { + CorrelationId = request.CorrelationId, + WorkflowInstanceId = request.WorkflowInstanceId, + ActivityInstanceId = request.ActivityInstanceId, + Input = request.Input + }, cancellationToken); + return DispatchWorkflowResponse.Success(); + } + + private async Task GetSendEndpointAsync(DispatchWorkflowOptions? options = default) + { + var endpointName = _endpointChannelFormatter.FormatEndpointName(options?.Channel); + var sendEndpoint = await _bus.GetSendEndpoint(new Uri($"queue:{endpointName}")); + return sendEndpoint; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs b/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs index 6a51977d9..f4cd46fd9 100644 --- a/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs +++ b/src/modules/Elsa.Quartz/Jobs/ResumeWorkflowJob.cs @@ -39,6 +39,6 @@ public class ResumeWorkflowJob : IJob CorrelationId = (string?)map.Get(nameof(DispatchWorkflowInstanceRequest.CorrelationId)), Input = map.GetDictionary(nameof(DispatchWorkflowInstanceRequest.Input)) }; - await _workflowDispatcher.DispatchAsync(request, context.CancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs b/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs index c81c2181c..fd4795f71 100644 --- a/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs +++ b/src/modules/Elsa.Quartz/Jobs/RunWorkflowJob.cs @@ -44,6 +44,6 @@ public class RunWorkflowJob : IJob CorrelationId = (string?)map.Get(nameof(DispatchWorkflowDefinitionRequest.CorrelationId)), Input = map.GetDictionary(nameof(DispatchWorkflowDefinitionRequest.Input)) }; - await _workflowDispatcher.DispatchAsync(request, context.CancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs b/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs index ab3448f1f..a9298f9f1 100644 --- a/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs +++ b/src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs @@ -26,6 +26,6 @@ public class ResumeWorkflowTask : ITask public async ValueTask ExecuteAsync(TaskExecutionContext context) { var workflowDispatcher = context.ServiceProvider.GetRequiredService(); - await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, context.CancellationToken); + await workflowDispatcher.DispatchAsync(_dispatchWorkflowInstanceRequest, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs b/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs index fdf100098..15cb29916 100644 --- a/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs +++ b/src/modules/Elsa.Scheduling/Tasks/RunWorkflowTask.cs @@ -25,6 +25,6 @@ public class RunWorkflowTask : ITask public async ValueTask ExecuteAsync(TaskExecutionContext context) { var workflowDispatcher = context.ServiceProvider.GetRequiredService(); - await workflowDispatcher.DispatchAsync(_request, context.CancellationToken); + await workflowDispatcher.DispatchAsync(_request, cancellationToken: context.CancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs index 903c88e2c..c269a6d33 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkDispatch/Endpoint.cs @@ -64,7 +64,7 @@ internal class Endpoint : ElsaEndpoint TriggerActivityId = triggerActivityId }; - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken); instanceIds.Add(instanceId); } diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs index 3a83d8e18..bf33ed80c 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Endpoint.cs @@ -4,25 +4,15 @@ using Elsa.Workflows.Contracts; using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; using JetBrains.Annotations; namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Dispatch; -[PublicAPI] -internal class Endpoint : ElsaEndpoint +[UsedImplicitly] +internal class Endpoint(IWorkflowDefinitionStore store, IWorkflowDispatcher workflowDispatcher, IIdentityGenerator identityGenerator) : ElsaEndpoint { - private readonly IWorkflowDefinitionStore _store; - private readonly IWorkflowDispatcher _workflowDispatcher; - private readonly IIdentityGenerator _identityGenerator; - - public Endpoint(IWorkflowDefinitionStore store, IWorkflowDispatcher workflowDispatcher, IIdentityGenerator identityGenerator) - { - _store = store; - _workflowDispatcher = workflowDispatcher; - _identityGenerator = identityGenerator; - } - public override void Configure() { Post("/workflow-definitions/{definitionId}/dispatch"); @@ -34,7 +24,7 @@ internal class Endpoint : ElsaEndpoint var definitionId = request.DefinitionId; var versionOptions = request.VersionOptions ?? VersionOptions.Published; - var exists = await _store.AnyAsync( + var exists = await store.AnyAsync( new WorkflowDefinitionFilter { DefinitionId = definitionId, @@ -47,23 +37,41 @@ internal class Endpoint : ElsaEndpoint await SendNotFoundAsync(cancellationToken); return; } + + var input = request.Input; + + if(input != null && input is not IDictionary) + { + AddError("Input must be a dictionary."); + await SendErrorsAsync(cancellation: cancellationToken); + return; + } - var instanceId = request.InstanceId ?? _identityGenerator.GenerateId(); - var correlationId = request.CorrelationId; - var triggerActivityId = request.TriggerActivityId; - var input = (IDictionary?)request.Input; + var instanceId = request.InstanceId ?? identityGenerator.GenerateId(); var dispatchRequest = new DispatchWorkflowDefinitionRequest { DefinitionId = definitionId, VersionOptions = versionOptions, - Input = input, + Input = input as IDictionary, InstanceId = instanceId, - CorrelationId = correlationId, - TriggerActivityId = triggerActivityId + CorrelationId = request.CorrelationId, + TriggerActivityId = request.TriggerActivityId, }; - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + var options = new DispatchWorkflowOptions + { + Channel = request.Channel + }; + var result = await workflowDispatcher.DispatchAsync(dispatchRequest, options, cancellationToken); + + if(!result.Succeeded) + { + AddError(result.ErrorMessage!); + await SendErrorsAsync(cancellation: cancellationToken); + return; + } + await SendOkAsync(new Response(instanceId), cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs index 2acb4f36c..0a10a7a21 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Dispatch/Models.cs @@ -15,6 +15,8 @@ internal class Request [JsonConverter(typeof(ExpandoObjectConverterFactory))] public object? Input { get; set; } + + public string? Channel { get; set; } } internal record Response(string WorkflowInstanceId); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs index 2a5cbe656..f45cc34f2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs @@ -7,13 +7,16 @@ using Elsa.Extensions; using Elsa.Workflows.Activities.Flowchart.Attributes; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Exceptions; using Elsa.Workflows.UIHints; using Elsa.Workflows.Memory; using Elsa.Workflows.Models; using Elsa.Workflows.Options; using Elsa.Workflows.Runtime.Bookmarks; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.UIHints; using Elsa.Workflows.Services; using JetBrains.Annotations; @@ -73,6 +76,17 @@ public class BulkDispatchWorkflows : Activity DefaultValue = true)] public Input WaitForCompletion { get; set; } = new(true); + /// + /// The channel to dispatch the workflow to. + /// + [Input( + DisplayName = "Channel", + Description = "The channel to dispatch the workflow to.", + UIHint = InputUIHints.DropDown, + UIHandler = typeof(DispatcherChannelOptionsProvider) + )] + public Input ChannelName { get; set; } = default!; + /// /// An activity to execute when the child workflow finishes. /// @@ -97,7 +111,7 @@ public class BulkDispatchWorkflows : Activity await ProcessItem(context, item); dispatchedInstancesCount++; } - + context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount); // If we need to wait for the child workflows to complete (if any), create a bookmark. @@ -144,6 +158,7 @@ public class BulkDispatchWorkflows : Activity var workflowDefinitionId = WorkflowDefinitionId.Get(context); var parentInstanceId = context.WorkflowExecutionContext.Id; var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); var properties = new Dictionary { ["ParentInstanceId"] = parentInstanceId @@ -173,12 +188,19 @@ public class BulkDispatchWorkflows : Activity CorrelationId = correlationId, InstanceId = instanceId }; + var options = new DispatchWorkflowOptions + { + Channel = channelName + }; - await workflowDispatcher.DispatchAsync(request, context.CancellationToken); + var dispatchResponse = await workflowDispatcher.DispatchAsync(request, options, context.CancellationToken); + + if (!dispatchResponse.Succeeded) + throw new FaultException(dispatchResponse.ErrorMessage); return instanceId; } - + private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context) { var input = context.WorkflowInput; diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs index dfba4ac65..eff1280c3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs @@ -3,11 +3,14 @@ using Elsa.Common.Models; using Elsa.Extensions; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; +using Elsa.Workflows.Exceptions; using Elsa.Workflows.UIHints; using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Bookmarks; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.UIHints; using JetBrains.Annotations; namespace Elsa.Workflows.Runtime.Activities; @@ -55,6 +58,17 @@ public class DispatchWorkflow : Activity [Input(Description = "Wait for the child workflow to complete before completing this activity.")] public Input WaitForCompletion { get; set; } = default!; + /// + /// The channel to dispatch the workflow to. + /// + [Input( + DisplayName = "Channel", + Description = "The channel to dispatch the workflow to.", + UIHint = InputUIHints.DropDown, + UIHandler = typeof(DispatcherChannelOptionsProvider) + )] + public Input ChannelName { get; set; } = default!; + /// protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { @@ -85,6 +99,7 @@ public class DispatchWorkflow : Activity { var workflowDefinitionId = WorkflowDefinitionId.Get(context); var input = Input.GetOrDefault(context) ?? new Dictionary(); + var channelName = ChannelName.GetOrDefault(context); input["ParentInstanceId"] = context.WorkflowExecutionContext.Id; @@ -98,11 +113,18 @@ public class DispatchWorkflow : Activity VersionOptions = VersionOptions.Published, Input = input, CorrelationId = correlationId, - InstanceId = instanceId + InstanceId = instanceId, + }; + var options = new DispatchWorkflowOptions + { + Channel = channelName }; // Dispatch the child workflow. - await workflowDispatcher.DispatchAsync(request, context.CancellationToken); + var dispatchResponse = await workflowDispatcher.DispatchAsync(request, options, context.CancellationToken); + + if (!dispatchResponse.Succeeded) + throw new FaultException(dispatchResponse.ErrorMessage); return instanceId; } diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs index ca20040af..744fccf7f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDispatcher.cs @@ -1,4 +1,6 @@ +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; namespace Elsa.Workflows.Runtime.Contracts; @@ -10,20 +12,20 @@ public interface IWorkflowDispatcher /// /// Dispatches a request to execute the specified workflow definition. /// - Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Dispatches a request to execute the specified workflow instance. /// - Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload. /// - Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); /// /// Resumes the workflow waiting for the specified bookmark. /// - Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default); + Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 23919dcde..744b73be2 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -12,6 +12,7 @@ using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Handlers; using Elsa.Workflows.Runtime.HostedServices; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Providers; using Elsa.Workflows.Runtime.Services; @@ -33,6 +34,8 @@ public class WorkflowRuntimeFeature : FeatureBase { } + private IDictionary WorkflowDispatcherChannels { get; set; } = new Dictionary(); + /// /// A list of workflow builders configured during application startup. /// @@ -46,7 +49,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// /// A factory that instantiates an . /// - public Func WorkflowDispatcher { get; set; } = sp => ActivatorUtilities.CreateInstance(sp); + public Func WorkflowDispatcher { get; set; } = sp => + { + var decoratedService = ActivatorUtilities.CreateInstance(sp); + return ActivatorUtilities.CreateInstance(sp, decoratedService); + }; /// /// A factory that instantiates an . @@ -102,6 +109,11 @@ public class WorkflowRuntimeFeature : FeatureBase /// A delegate to configure the . /// public Action WorkflowInboxCleanupOptions { get; set; } = _ => { }; + + /// + /// A delegate to configure the . + /// + public Action WorkflowDispatcherOptions { get; set; } = _ => { }; /// /// Register the specified workflow type. @@ -127,6 +139,26 @@ public class WorkflowRuntimeFeature : FeatureBase return this; } + /// + /// Adds a dispatcher channel. + /// + public WorkflowRuntimeFeature AddDispatcherChannel(string channel) + { + return AddDispatcherChannel(new DispatcherChannel + { + Name = channel + }); + } + + /// + /// Adds a dispatcher channel. + /// + public WorkflowRuntimeFeature AddDispatcherChannel(DispatcherChannel channel) + { + WorkflowDispatcherChannels[channel.Name] = channel; + return this; + } + /// public override void Configure() { @@ -146,8 +178,13 @@ public class WorkflowRuntimeFeature : FeatureBase { // Options. Services.Configure(DistributedLockingOptions); - Services.Configure(options => { options.Workflows = Workflows; }); Services.Configure(WorkflowInboxCleanupOptions); + Services.Configure(WorkflowDispatcherOptions); + Services.Configure(options => { options.Workflows = Workflows; }); + Services.Configure(options => + { + options.Channels.AddRange(WorkflowDispatcherChannels.Values); + }); Services // Core. @@ -214,7 +251,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddNotificationHandler() .AddNotificationHandler() .AddNotificationHandler() - .AddNotificationHandler() + .AddNotificationHandler() .AddNotificationHandler() // Workflow activation strategies. diff --git a/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs b/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs new file mode 100644 index 000000000..cd276011b --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/DispatchWorkflowOptions.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.Runtime.Models; + +public class DispatchWorkflowOptions +{ + public string? Channel { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs b/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs new file mode 100644 index 000000000..ee785c8aa --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/DispatcherChannel.cs @@ -0,0 +1,13 @@ +namespace Elsa.Workflows.Runtime.Models; + +/// +/// Describes a workflow dispatcher channel. +/// +public class DispatcherChannel +{ + /// + /// Gets or sets the name of the channel. + /// + public string Name { get; set; } = default!; + +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs new file mode 100644 index 000000000..385f13acc --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Options/WorkflowDispatcherOptions.cs @@ -0,0 +1,17 @@ +using Elsa.Workflows.Runtime.Models; + +namespace Elsa.Workflows.Runtime.Options; + +/// +/// Provides options to the workflow dispatcher. +/// +public class WorkflowDispatcherOptions +{ + /// + /// A list of available channels to dispatch workflows to. + /// + /// + /// Channels are used to dispatch workflows to different queues or endpoints. + /// + public ICollection Channels { get; set; } = new List(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs deleted file mode 100644 index 0d63e9fe2..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchResumeWorkflowsResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchResumeWorkflowsResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs deleted file mode 100644 index 601460c67..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchTriggerWorkflowsResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchTriggerWorkflowsResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs index bd88eaf9f..103b0f760 100644 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs +++ b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionRequest.cs @@ -60,5 +60,5 @@ public class DispatchWorkflowDefinitionRequest /// /// The ID of the activity that triggered the workflow. /// - public string? TriggerActivityId { get; init; } + public string? TriggerActivityId { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs deleted file mode 100644 index 6320f22ef..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowDefinitionResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchWorkflowDefinitionResponse(); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs b/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs deleted file mode 100644 index 79f91af3b..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Requests/DispatchWorkflowInstanceResponse.cs +++ /dev/null @@ -1,3 +0,0 @@ -namespace Elsa.Workflows.Runtime.Requests; - -public record DispatchWorkflowInstanceResponse; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs new file mode 100644 index 000000000..744eeb8a2 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Responses/DispatchWorkflowDefinitionResponse.cs @@ -0,0 +1,18 @@ +namespace Elsa.Workflows.Runtime.Responses; + +/// +/// Represents the response of a dispatch action for a workflow definition. +/// +public record DispatchWorkflowResponse(bool Succeeded, string? ErrorMessage) +{ + /// + /// Creates a response indicating that the dispatch action was successful. + /// + /// + public static DispatchWorkflowResponse Success() => new DispatchWorkflowResponse(true, default); + + /// + /// Creates a response indicating that the specified channel does not exist. + /// + public static DispatchWorkflowResponse UnknownChannel() => new DispatchWorkflowResponse(false, "The specified channel does not exist."); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs index e08df6bef..cd8176a2e 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs @@ -2,7 +2,9 @@ using Elsa.Mediator; using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Commands; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; namespace Elsa.Workflows.Runtime.Services; @@ -22,7 +24,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher } /// - public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchWorkflowDefinitionCommand(request.DefinitionId, request.VersionOptions) { @@ -34,11 +36,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchWorkflowDefinitionResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchWorkflowInstanceCommand(request.InstanceId){ BookmarkId = request.BookmarkId, @@ -51,11 +53,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher CorrelationId = request.CorrelationId}; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchWorkflowInstanceResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchTriggerWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload) { @@ -66,11 +68,11 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher Properties = request.Properties }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchTriggerWorkflowsResponse(); + return DispatchWorkflowResponse.Success(); } /// - public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, CancellationToken cancellationToken = default) + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) { var command = new DispatchResumeWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload) { @@ -80,6 +82,6 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher Input = request.Input }; await _commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); - return new DispatchResumeWorkflowsResponse(); + return DispatchWorkflowResponse.Success(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs index 7da481ea0..75932fb66 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs @@ -142,6 +142,6 @@ public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker return; } - await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken); + await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs index 1b9259f69..af952d15d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowInbox.cs @@ -133,7 +133,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox WorkflowInstanceId = workflowInstanceId, ActivityInstanceId = activityInstanceId, Input = input - }, cancellationToken); + }, cancellationToken: cancellationToken); return new DeliverWorkflowInboxMessageResult(new List()); } @@ -153,7 +153,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox WorkflowInstanceId = workflowInstanceId, ActivityInstanceId = activityInstanceId, Input = input - }, cancellationToken); + }, cancellationToken: cancellationToken); } private async Task> ResumeWorkflowsSynchronouslyAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs b/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs index 4ceeadcb2..ff00e451a 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/TaskReporter.cs @@ -31,6 +31,6 @@ public class TaskReporter : ITaskReporter var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(); var request = new DispatchResumeWorkflowsRequest(activityTypeName, bookmarkPayload, Input: input); - await _workflowDispatcher.DispatchAsync(request, cancellationToken); + await _workflowDispatcher.DispatchAsync(request, cancellationToken: cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs new file mode 100644 index 000000000..d0ddfad58 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/ValidatingWorkflowDispatcher.cs @@ -0,0 +1,63 @@ +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime.Services; + +/// +/// Validates the workflow request before dispatching it to the workflow dispatcher. +/// +/// The workflow dispatcher to decorate. +public class ValidatingWorkflowDispatcher(IWorkflowDispatcher decoratedService, IOptions dispatcherOptions) : IWorkflowDispatcher +{ + private IWorkflowDispatcher DecoratedService { get; set; } = decoratedService; + + /// + public async Task DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + /// + public async Task DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default) + { + if (!ValidateChannel(options?.Channel)) + return DispatchWorkflowResponse.UnknownChannel(); + + return await DecoratedService.DispatchAsync(request, options, cancellationToken); + } + + private bool ValidateChannel(string? channelName) + { + return string.IsNullOrEmpty(channelName) || GetChannelExists(channelName); + } + + private bool GetChannelExists(string channelName) + { + return dispatcherOptions.Value.Channels.Any(x => x.Name == channelName); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs b/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs new file mode 100644 index 000000000..f094903d9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/UIHints/DispatcherChannelOptionsProvider.cs @@ -0,0 +1,31 @@ +using System.Reflection; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.UIHints.Dropdown; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime.UIHints; + +/// +/// Provides options for activities that dispatch workflows to channels. +/// +public class DispatcherChannelOptionsProvider : DropDownOptionsProviderBase +{ + private readonly WorkflowDispatcherOptions _options; + + /// + public DispatcherChannelOptionsProvider(IOptions options) + { + _options = options.Value; + } + + /// + protected override ValueTask> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken) + { + var channelNames = _options.Channels.Select(x => x.Name).ToList(); + var selectListItems = new List { new("Default", "") }; + + selectListItems.AddRange(channelNames.Select(x => new SelectListItem(x, x))); + + return new(selectListItems); + } +} \ No newline at end of file